d624413ddd
As requested in #22120, moved the files and fixed other includes and build system. Moved files: - query.cc - query-request.hh - query-result.hh - query-result-reader.hh - query-result-set.cc - query-result-set.hh - query-result-writer.hh - query_id.hh - query_result_merger.hh Fixes: #22120 This is a cleanup, no need to backport Closes scylladb/scylladb#25105
240 lines
10 KiB
C++
240 lines
10 KiB
C++
/*
|
|
* Copyright (C) 2018-present ScyllaDB
|
|
*/
|
|
|
|
/*
|
|
* SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
|
|
*/
|
|
|
|
#pragma once
|
|
|
|
#include "replica/database.hh"
|
|
#include "db/system_keyspace.hh"
|
|
#include "readers/mutation_reader.hh"
|
|
#include "mutation/mutation_fragment_v2.hh"
|
|
#include "query/query-request.hh"
|
|
#include "schema/schema_fwd.hh"
|
|
#include "secondary_index_manager.hh"
|
|
#include "view_info.hh"
|
|
|
|
#include <memory>
|
|
|
|
namespace tracing { class trace_state_ptr; }
|
|
|
|
namespace db::index {
|
|
|
|
class built_indexes_virtual_reader {
|
|
replica::database& _db;
|
|
|
|
struct built_indexes_reader : mutation_reader::impl {
|
|
replica::database& _db;
|
|
schema_ptr _schema;
|
|
query::partition_slice _view_names_slice;
|
|
mutation_reader _underlying;
|
|
sstring _current_keyspace;
|
|
|
|
// Convert a key holding an index name (e.g., xyz) to a key holding
|
|
// the backing materialized-view name (e.g., xyz_index).
|
|
static clustering_key_prefix index_name_to_view_name(
|
|
const clustering_key_prefix& c,
|
|
const schema& built_views_schema,
|
|
const schema& built_indexes_schema)
|
|
{
|
|
sstring index_name = value_cast<sstring>(utf8_type->deserialize(*c.begin(built_indexes_schema)));
|
|
sstring view_name = ::secondary_index::index_table_name(index_name);
|
|
return clustering_key_prefix::from_single_value(built_views_schema, utf8_type->decompose(view_name));
|
|
}
|
|
|
|
// Convert a clustering_range holding a range of index names (e.g.,
|
|
// abc..xyz) to a range of backing view names (abc_index..xyz_index).
|
|
static query::clustering_range index_name_range_to_view_name_range(
|
|
const query::clustering_range& range,
|
|
const schema& built_views_schema,
|
|
const schema& built_indexes_schema)
|
|
{
|
|
bool singular = range.is_singular();
|
|
auto start = range.start_copy();
|
|
auto end = range.end_copy();
|
|
if (start) {
|
|
*start = query::clustering_range::bound(
|
|
index_name_to_view_name(start->value(), built_views_schema, built_indexes_schema),
|
|
start->is_inclusive());
|
|
}
|
|
if (end) {
|
|
*end = query::clustering_range::bound(
|
|
index_name_to_view_name(end->value(), built_views_schema, built_indexes_schema),
|
|
end->is_inclusive());
|
|
}
|
|
return query::clustering_range(std::move(start), std::move(end), singular);
|
|
}
|
|
|
|
// The index list reader is created with a slice of index names
|
|
// (e.g., xyz) but the underlying reader needs a slice of view names,
|
|
// e.g., xyz_index, so we need to add the _index suffix with
|
|
// index_table_name().
|
|
static query::partition_slice index_slice_to_view_slice(
|
|
const query::partition_slice& slice,
|
|
const schema& built_views_schema,
|
|
const schema& built_indexes_schema)
|
|
{
|
|
// Most of the slice - such as options and which columns to read -
|
|
// should be copied unchanged. Just the clustering row ranges need
|
|
// changing.
|
|
// Fix _row_ranges (used for WHERE specifying a clustering range):
|
|
query::partition_slice ret = slice;
|
|
for (query::clustering_range &range : ret._row_ranges) {
|
|
range = index_name_range_to_view_name_range(range, built_views_schema, built_indexes_schema);
|
|
}
|
|
// Fix _specific_ranges (used for continuing on paging).
|
|
auto& specific_ranges = ret.get_specific_ranges();
|
|
if (specific_ranges) {
|
|
partition_key pk = specific_ranges->pk();
|
|
std::vector<query::clustering_range> ranges;
|
|
for (const query::clustering_range &range : specific_ranges->ranges()) {
|
|
ranges.push_back(index_name_range_to_view_name_range(range, built_views_schema, built_indexes_schema));
|
|
}
|
|
ret.clear_ranges();
|
|
ret.set_range(built_indexes_schema, pk, ranges);
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
built_indexes_reader(
|
|
replica::database& db,
|
|
schema_ptr schema,
|
|
reader_permit permit,
|
|
replica::column_family& built_views,
|
|
const dht::partition_range& range,
|
|
const query::partition_slice& slice,
|
|
tracing::trace_state_ptr trace_state,
|
|
streamed_mutation::forwarding fwd,
|
|
mutation_reader::forwarding fwd_mr)
|
|
: mutation_reader::impl(schema, permit)
|
|
, _db(db)
|
|
, _schema(std::move(schema))
|
|
, _view_names_slice(index_slice_to_view_slice(slice, *built_views.schema(), *_schema))
|
|
, _underlying(built_views.make_mutation_reader(
|
|
built_views.schema(),
|
|
std::move(permit),
|
|
range,
|
|
_view_names_slice,
|
|
std::move(trace_state),
|
|
fwd,
|
|
fwd_mr)) {
|
|
}
|
|
|
|
bool is_index(const clustering_row& built_view_row) const {
|
|
auto view_name = value_cast<sstring>(utf8_type->deserialize(*built_view_row.key().begin(*_underlying.schema())));
|
|
try {
|
|
auto s = _db.find_schema(_current_keyspace, view_name);
|
|
return s->is_view() && _db.find_column_family(s->view_info()->base_id()).get_index_manager().is_index(view_ptr(s));
|
|
} catch (const replica::no_such_column_family&) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
virtual future<> fill_buffer() override {
|
|
return _underlying.fill_buffer().then([this] {
|
|
_end_of_stream = _underlying.is_end_of_stream();
|
|
while (!_underlying.is_buffer_empty()) {
|
|
auto mf = _underlying.pop_mutation_fragment();
|
|
if (mf.is_partition_start()) {
|
|
_current_keyspace = value_cast<sstring>(utf8_type->deserialize(*mf.as_partition_start().key().key().begin(*_underlying.schema())));
|
|
} else if (mf.is_clustering_row()) {
|
|
if (is_index(mf.as_clustering_row())) {
|
|
// mf's clustering key is the materialized view's
|
|
// name, xyz_index, while we need the index name
|
|
// xyz. So modify mf:
|
|
mf.mutate_as_clustering_row(*_underlying.schema(),
|
|
[&] (clustering_row& cr) mutable {
|
|
sstring view_name = value_cast<sstring>(utf8_type->deserialize(*cr.key().begin(*_underlying.schema())));
|
|
sstring index_name = ::secondary_index::index_name_from_table_name(view_name);
|
|
cr.key() = clustering_key_prefix::from_single_value(*_underlying.schema(), utf8_type->decompose(index_name));
|
|
});
|
|
} else {
|
|
continue;
|
|
}
|
|
}
|
|
push_mutation_fragment(std::move(mf));
|
|
}
|
|
});
|
|
}
|
|
|
|
virtual future<> next_partition() override {
|
|
_end_of_stream = false;
|
|
clear_buffer_to_next_partition();
|
|
if (is_buffer_empty()) {
|
|
return _underlying.next_partition();
|
|
}
|
|
return make_ready_future<>();
|
|
}
|
|
|
|
virtual future<> fast_forward_to(const dht::partition_range& pr) override {
|
|
clear_buffer();
|
|
_end_of_stream = false;
|
|
return _underlying.fast_forward_to(pr);
|
|
}
|
|
|
|
virtual future<> fast_forward_to(position_range range) override {
|
|
clear_buffer();
|
|
_end_of_stream = false;
|
|
// range contains index names (e.g., xyz) but the underlying table
|
|
// contains view names (e.g., xyz_index) so we need to add the
|
|
// _index suffix with index_table_name():
|
|
position_in_partition start = range.start();
|
|
position_in_partition end = range.end();
|
|
if (start.get_clustering_key_prefix()) {
|
|
clustering_key_prefix ck = index_name_to_view_name(
|
|
*start.get_clustering_key_prefix(),
|
|
*_underlying.schema(), *_schema);
|
|
start = position_in_partition(start.region(),
|
|
start.get_bound_weight(),
|
|
std::move(ck));
|
|
}
|
|
if (end.get_clustering_key_prefix()) {
|
|
clustering_key_prefix ck = index_name_to_view_name(
|
|
*end.get_clustering_key_prefix(),
|
|
*_underlying.schema(), *_schema);
|
|
start = position_in_partition(end.region(),
|
|
end.get_bound_weight(),
|
|
std::move(ck));
|
|
}
|
|
range = position_range(std::move(start), std::move(end));
|
|
return _underlying.fast_forward_to(std::move(range));
|
|
}
|
|
|
|
virtual future<> close() noexcept override {
|
|
return _underlying.close();
|
|
}
|
|
};
|
|
|
|
public:
|
|
built_indexes_virtual_reader(replica::database& db)
|
|
: _db(db) {
|
|
}
|
|
|
|
mutation_reader operator()(
|
|
schema_ptr s,
|
|
reader_permit permit,
|
|
const dht::partition_range& range,
|
|
const query::partition_slice& slice,
|
|
tracing::trace_state_ptr trace_state,
|
|
streamed_mutation::forwarding fwd,
|
|
mutation_reader::forwarding fwd_mr) {
|
|
// Use of BUILT_VIEWS as filtering table should be in sync with
|
|
// cf::get_built_indexes's filtering with load_built_views()
|
|
return make_mutation_reader<built_indexes_reader>(
|
|
_db,
|
|
s,
|
|
std::move(permit),
|
|
_db.find_column_family(s->ks_name(), system_keyspace::v3::BUILT_VIEWS),
|
|
range,
|
|
slice,
|
|
std::move(trace_state),
|
|
fwd,
|
|
fwd_mr);
|
|
}
|
|
};
|
|
|
|
}
|