bahman/scylla
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Merge "Convert result builders to v2" from Botond

Browse Source 
"
Namely the query result writer and the reconcilable result builder, used
for building results for regular queries and mutation queries (used in
read repair) respectively.
With this, there are no users left for the v1 output of the compactor,
so we remove that, making the compactor v2 all-the-way (and simpler).
This means that for regular queries, a downgrade phase is eliminated
completely, as regular queries don't store range tombstone in their
result, so no need to convert them.

Tests: unit(dev, release, debug)
"

* 'result-builders-v2/v1' of https://github.com/denesb/scylla:
  reconcilable_result_builder: remove v1 support
  query_result_builder: remove v1 support
  mutation_compactor: drop v1 related code-paths
  mutation_compactor: drop v1 support altogether from the API
  tree: migrate to the v2 consumer APIs
  test/boost/mutation_test: remove v1 specific test code
  querier: switch to v2 compactor output
  reconcilable_result_builder: add v2 support
  query_result_writer: add v2 support
  query_result_builder: make consume(range_tombstone) noop
This commit is contained in:
Avi Kivity
2022-03-15 14:32:58 +02:00
parent 7ce0d380d4 0632114a9b
commit 39504dea61
10 changed files with 72 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
db/view/view.cc
View File

@@ -364,7 +364,7 @@ public:
void consume(tombstone t) { _builder.consume(t); }
stop_iteration consume(static_row&& sr, tombstone t, bool is_alive) { return _builder.consume(std::move(sr), t, is_alive); }
stop_iteration consume(clustering_row&& cr, row_tombstone t, bool is_alive) { return _builder.consume(std::move(cr), t, is_alive); }
stop_iteration consume(range_tombstone&& rt) { return _builder.consume(std::move(rt)); }
stop_iteration consume(range_tombstone_change&& rtc) { return _builder.consume(std::move(rtc)); }
stop_iteration consume_end_of_partition() { return _builder.consume_end_of_partition(); }
result_type consume_end_of_stream() {
_builder.consume_end_of_stream();
@@ -1950,7 +1950,7 @@ public:
return stop_iteration::no;
}
stop_iteration consume(range_tombstone&&) {
stop_iteration consume(range_tombstone_change&&) {
inject_failure("view_builder_consume_range_tombstone");
return stop_iteration::no;
}
@@ -2008,7 +2008,7 @@ public:
// Called in the context of a seastar::thread.
void view_builder::execute(build_step& step, exponential_backoff_retry r) {
gc_clock::time_point now = gc_clock::now();
auto consumer = compact_for_query<emit_only_live_rows::yes, view_builder::consumer>(
auto consumer = compact_for_query_v2<emit_only_live_rows::yes, view_builder::consumer>(
*step.reader.schema(),
now,
step.pslice,

6
multishard_mutation_query.cc
View File

@@ -610,7 +610,7 @@ future<> read_context::save_readers(flat_mutation_reader_v2::tracked_buffer unco
namespace {
template <typename ResultType>
using compact_for_result_state = compact_for_query_state<ResultType::only_live>;
using compact_for_result_state = compact_for_query_state_v2<ResultType::only_live>;
template <typename ResultBuilder>
requires std::is_nothrow_move_constructible_v<typename ResultBuilder::result_type>
@@ -821,7 +821,7 @@ public:
void consume(tombstone t) { _builder.consume(t); }
stop_iteration consume(static_row&& sr, tombstone t, bool is_alive) { return _builder.consume(std::move(sr), t, is_alive); }
stop_iteration consume(clustering_row&& cr, row_tombstone t, bool is_alive) { return _builder.consume(std::move(cr), t, is_alive); }
stop_iteration consume(range_tombstone&& rt) { return _builder.consume(std::move(rt)); }
stop_iteration consume(range_tombstone_change&& rtc) { return _builder.consume(std::move(rtc)); }
stop_iteration consume_end_of_partition() { return _builder.consume_end_of_partition(); }
result_type consume_end_of_stream() { return _builder.consume_end_of_stream(); }
};
@@ -844,7 +844,7 @@ public:
void consume(tombstone t) { _builder.consume(t); }
stop_iteration consume(static_row&& sr, tombstone t, bool is_alive) { return _builder.consume(std::move(sr), t, is_alive); }
stop_iteration consume(clustering_row&& cr, row_tombstone t, bool is_alive) { return _builder.consume(std::move(cr), t, is_alive); }
stop_iteration consume(range_tombstone&& rt) { return _builder.consume(std::move(rt)); }
stop_iteration consume(range_tombstone_change&& rtc) { return _builder.consume(std::move(rtc)); }
stop_iteration consume_end_of_partition() { return _builder.consume_end_of_partition(); }
result_type consume_end_of_stream() {
_builder.consume_end_of_stream();

184
mutation_compactor.hh
View File

@@ -30,23 +30,6 @@ enum class compact_for_sstables {
yes,
};
enum class compactor_output_format {
v1,
v2
};
template<typename T>
concept CompactedFragmentsConsumer = requires(T obj, tombstone t, const dht::decorated_key& dk, static_row sr,
clustering_row cr, range_tombstone rt, tombstone current_tombstone, row_tombstone current_row_tombstone, bool is_alive) {
obj.consume_new_partition(dk);
obj.consume(t);
{ obj.consume(std::move(sr), current_tombstone, is_alive) } -> std::same_as<stop_iteration>;
{ obj.consume(std::move(cr), current_row_tombstone, is_alive) } -> std::same_as<stop_iteration>;
{ obj.consume(std::move(rt)) } -> std::same_as<stop_iteration>;
{ obj.consume_end_of_partition() } -> std::same_as<stop_iteration>;
obj.consume_end_of_stream();
};
template<typename T>
concept CompactedFragmentsConsumerV2 = requires(T obj, tombstone t, const dht::decorated_key& dk, static_row sr,
clustering_row cr, range_tombstone_change rtc, tombstone current_tombstone, row_tombstone current_row_tombstone, bool is_alive) {
@@ -59,12 +42,6 @@ concept CompactedFragmentsConsumerV2 = requires(T obj, tombstone t, const dht::d
obj.consume_end_of_stream();
};
// TODO: I want to make this choose the right concept for OutputFormat but
// probably not worth the effort for the (hopefully) brief time for which we
// have to support both.
template<typename T, compactor_output_format OutputFormat>
concept CompactedFragmentsConsumerWithVersion = CompactedFragmentsConsumer<T> || CompactedFragmentsConsumerV2<T>;
struct detached_compaction_state {
::partition_start partition_start;
std::optional<::static_row> static_row;
@@ -77,7 +54,6 @@ public:
void consume(tombstone t) {}
stop_iteration consume(static_row&& sr, tombstone, bool) { return stop_iteration::no; }
stop_iteration consume(clustering_row&& cr, row_tombstone, bool) { return stop_iteration::no; }
stop_iteration consume(range_tombstone&& rt) { return stop_iteration::no; }
stop_iteration consume(range_tombstone_change&& rtc) { return stop_iteration::no; }
stop_iteration consume_end_of_partition() { return stop_iteration::no; }
void consume_end_of_stream() {}
@@ -159,7 +135,7 @@ struct compaction_stats {
// emit_only_live::yes will cause compact_for_query to emit only live
// static and clustering rows. It doesn't affect the way range tombstones are
// emitted.
template<emit_only_live_rows OnlyLive, compact_for_sstables SSTableCompaction, compactor_output_format OutputFormat = compactor_output_format::v1>
template<emit_only_live_rows OnlyLive, compact_for_sstables SSTableCompaction>
class compact_mutation_state {
const schema& _schema;
gc_clock::time_point _query_time;
@@ -173,7 +149,6 @@ class compact_mutation_state {
uint64_t _partition_row_limit{};
tombstone _partition_tombstone;
range_tombstone_assembler _rt_assembler;
bool _static_row_live{};
uint64_t _rows_in_current_partition;
@@ -200,21 +175,7 @@ class compact_mutation_state {
compaction_stats _stats;
private:
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
stop_iteration do_consume(range_tombstone&& rt, Consumer& consumer, GCConsumer& gc_consumer) {
if (rt.tomb <= _partition_tombstone) {
return stop_iteration::no;
}
if (can_purge_tombstone(rt.tomb)) {
partition_is_not_empty_for_gc_consumer(gc_consumer);
return gc_consumer.consume(std::move(rt));
} else {
partition_is_not_empty(consumer);
return consumer.consume(std::move(rt));
}
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
stop_iteration do_consume(range_tombstone_change&& rtc, Consumer& consumer, GCConsumer& gc_consumer) {
stop_iteration gc_consumer_stop = stop_iteration::no;
stop_iteration consumer_stop = stop_iteration::no;
@@ -240,19 +201,6 @@ private:
}
return gc_consumer_stop || consumer_stop;
}
template <typename Consumer, typename GCConsumer>
tombstone tombstone_for_row(const clustering_key& ckey, Consumer& consumer, GCConsumer& gc_consumer) {
if constexpr (OutputFormat == compactor_output_format::v2) {
return std::max(_partition_tombstone, _effective_tombstone);
} else {
if (_rt_assembler.needs_flush()) {
if (auto rt_opt = _rt_assembler.flush(_schema, position_in_partition::after_key(ckey))) {
do_consume(std::move(*rt_opt), consumer, gc_consumer);
}
}
return std::max(_partition_tombstone, _rt_assembler.get_current_tombstone());
}
}
static constexpr bool only_live() {
return OnlyLive == emit_only_live_rows::yes;
}
@@ -368,7 +316,6 @@ public:
_rows_in_current_partition = 0;
_static_row_live = false;
_partition_tombstone = {};
_rt_assembler.reset();
_current_partition_limit = std::min(_row_limit, _partition_row_limit);
_max_purgeable = api::missing_timestamp;
_gc_before = std::nullopt;
@@ -380,7 +327,7 @@ public:
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
void consume(tombstone t, Consumer& consumer, GCConsumer& gc_consumer) {
_partition_tombstone = t;
if (!only_live()) {
@@ -393,13 +340,13 @@ public:
}
template <typename Consumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer>
void force_partition_not_empty(Consumer& consumer) {
partition_is_not_empty(consumer);
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
stop_iteration consume(static_row&& sr, Consumer& consumer, GCConsumer& gc_consumer) {
_last_static_row = static_row(_schema, sr);
auto current_tombstone = _partition_tombstone;
@@ -430,12 +377,12 @@ public:
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
stop_iteration consume(clustering_row&& cr, Consumer& consumer, GCConsumer& gc_consumer) {
if (!sstable_compaction()) {
_last_clustering_pos = cr.position();
}
auto current_tombstone = tombstone_for_row(cr.key(), consumer, gc_consumer);
auto current_tombstone = std::max(_partition_tombstone, _effective_tombstone);
auto t = cr.tomb();
t.apply(current_tombstone);
@@ -497,39 +444,25 @@ public:
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
stop_iteration consume(range_tombstone_change&& rtc, Consumer& consumer, GCConsumer& gc_consumer) {
if (!sstable_compaction()) {
_last_clustering_pos = rtc.position();
}
++_stats.range_tombstones;
if constexpr (OutputFormat == compactor_output_format::v1) {
_effective_tombstone = rtc.tombstone();
if (auto rt_opt = _rt_assembler.consume(_schema, std::move(rtc))) {
return do_consume(std::move(*rt_opt), consumer, gc_consumer);
}
} else {
do_consume(std::move(rtc), consumer, gc_consumer);
}
do_consume(std::move(rtc), consumer, gc_consumer);
return stop_iteration::no;
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
stop_iteration consume_end_of_partition(Consumer& consumer, GCConsumer& gc_consumer) {
if (_effective_tombstone) {
auto rtc = range_tombstone_change(position_in_partition::after_key(_last_clustering_pos), tombstone{});
if constexpr (OutputFormat == compactor_output_format::v1) {
if (auto rt_opt = _rt_assembler.consume(_schema, std::move(rtc))) {
do_consume(std::move(*rt_opt), consumer, gc_consumer);
}
_rt_assembler.on_end_of_stream();
} else {
// do_consume() overwrites _effective_tombstone with {}, so save and restore it.
auto prev_tombstone = _effective_tombstone;
do_consume(std::move(rtc), consumer, gc_consumer);
_effective_tombstone = prev_tombstone;
}
// do_consume() overwrites _effective_tombstone with {}, so save and restore it.
auto prev_tombstone = _effective_tombstone;
do_consume(std::move(rtc), consumer, gc_consumer);
_effective_tombstone = prev_tombstone;
}
if (!_empty_partition_in_gc_consumer) {
gc_consumer.consume_end_of_partition();
@@ -554,7 +487,7 @@ public:
}
template <typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat> && CompactedFragmentsConsumerWithVersion<GCConsumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
auto consume_end_of_stream(Consumer& consumer, GCConsumer& gc_consumer) {
if (_dk) {
_last_dk = *_dk;
@@ -578,7 +511,7 @@ public:
/// partition-header and static row if there are clustering rows or range
/// tombstones left in the partition.
template <typename Consumer>
requires CompactedFragmentsConsumerWithVersion<Consumer, OutputFormat>
requires CompactedFragmentsConsumerV2<Consumer>
void start_new_page(uint64_t row_limit,
uint32_t partition_limit,
gc_clock::time_point query_time,
@@ -601,11 +534,7 @@ public:
}
if (_effective_tombstone) {
auto rtc = range_tombstone_change(position_in_partition_view::after_key(_last_clustering_pos), _effective_tombstone);
if constexpr (OutputFormat == compactor_output_format::v2) {
do_consume(std::move(rtc), consumer, nc);
} else if (auto rt_opt = _rt_assembler.consume(_schema, std::move(rtc))) {
do_consume(std::move(*rt_opt), consumer, nc);
}
do_consume(std::move(rtc), consumer, nc);
}
}
@@ -632,70 +561,10 @@ public:
const compaction_stats& stats() const { return _stats; }
};
template<emit_only_live_rows OnlyLive, compact_for_sstables SSTableCompaction, typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumer<Consumer> && CompactedFragmentsConsumer<GCConsumer>
class compact_mutation {
lw_shared_ptr<compact_mutation_state<OnlyLive, SSTableCompaction>> _state;
Consumer _consumer;
// Garbage Collected Consumer
GCConsumer _gc_consumer;
public:
compact_mutation(const schema& s, gc_clock::time_point query_time, const query::partition_slice& slice, uint64_t limit,
uint32_t partition_limit, Consumer consumer, GCConsumer gc_consumer = GCConsumer())
: _state(make_lw_shared<compact_mutation_state<OnlyLive, SSTableCompaction>>(s, query_time, slice, limit, partition_limit))
, _consumer(std::move(consumer))
, _gc_consumer(std::move(gc_consumer)) {
}
compact_mutation(const schema& s, gc_clock::time_point compaction_time,
std::function<api::timestamp_type(const dht::decorated_key&)> get_max_purgeable,
Consumer consumer, GCConsumer gc_consumer = GCConsumer())
: _state(make_lw_shared<compact_mutation_state<OnlyLive, SSTableCompaction>>(s, compaction_time, get_max_purgeable))
, _consumer(std::move(consumer))
, _gc_consumer(std::move(gc_consumer)) {
}
compact_mutation(lw_shared_ptr<compact_mutation_state<OnlyLive, SSTableCompaction>> state, Consumer consumer,
GCConsumer gc_consumer = GCConsumer())
: _state(std::move(state))
, _consumer(std::move(consumer))
, _gc_consumer(std::move(gc_consumer)) {
}
void consume_new_partition(const dht::decorated_key& dk) {
_state->consume_new_partition(dk);
}
void consume(tombstone t) {
_state->consume(std::move(t), _consumer, _gc_consumer);
}
stop_iteration consume(static_row&& sr) {
return _state->consume(std::move(sr), _consumer, _gc_consumer);
}
stop_iteration consume(clustering_row&& cr) {
return _state->consume(std::move(cr), _consumer, _gc_consumer);
}
stop_iteration consume(range_tombstone_change&& rtc) {
return _state->consume(std::move(rtc), _consumer, _gc_consumer);
}
stop_iteration consume_end_of_partition() {
return _state->consume_end_of_partition(_consumer, _gc_consumer);
}
auto consume_end_of_stream() {
return _state->consume_end_of_stream(_consumer, _gc_consumer);
}
};
template<emit_only_live_rows OnlyLive, compact_for_sstables SSTableCompaction, typename Consumer, typename GCConsumer>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>
class compact_mutation_v2 {
lw_shared_ptr<compact_mutation_state<OnlyLive, SSTableCompaction, compactor_output_format::v2>> _state;
lw_shared_ptr<compact_mutation_state<OnlyLive, SSTableCompaction>> _state;
Consumer _consumer;
// Garbage Collected Consumer
GCConsumer _gc_consumer;
@@ -703,7 +572,7 @@ class compact_mutation_v2 {
public:
compact_mutation_v2(const schema& s, gc_clock::time_point query_time, const query::partition_slice& slice, uint64_t limit,
uint32_t partition_limit, Consumer consumer, GCConsumer gc_consumer = GCConsumer())
: _state(make_lw_shared<compact_mutation_state<OnlyLive, SSTableCompaction, compactor_output_format::v2>>(s, query_time, slice, limit, partition_limit))
: _state(make_lw_shared<compact_mutation_state<OnlyLive, SSTableCompaction>>(s, query_time, slice, limit, partition_limit))
, _consumer(std::move(consumer))
, _gc_consumer(std::move(gc_consumer)) {
}
@@ -711,12 +580,12 @@ public:
compact_mutation_v2(const schema& s, gc_clock::time_point compaction_time,
std::function<api::timestamp_type(const dht::decorated_key&)> get_max_purgeable,
Consumer consumer, GCConsumer gc_consumer = GCConsumer())
: _state(make_lw_shared<compact_mutation_state<OnlyLive, SSTableCompaction, compactor_output_format::v2>>(s, compaction_time, get_max_purgeable))
: _state(make_lw_shared<compact_mutation_state<OnlyLive, SSTableCompaction>>(s, compaction_time, get_max_purgeable))
, _consumer(std::move(consumer))
, _gc_consumer(std::move(gc_consumer)) {
}
compact_mutation_v2(lw_shared_ptr<compact_mutation_state<OnlyLive, SSTableCompaction, compactor_output_format::v2>> state, Consumer consumer,
compact_mutation_v2(lw_shared_ptr<compact_mutation_state<OnlyLive, SSTableCompaction>> state, Consumer consumer,
GCConsumer gc_consumer = GCConsumer())
: _state(std::move(state))
, _consumer(std::move(consumer))
@@ -752,15 +621,6 @@ public:
}
};
template<emit_only_live_rows only_live, typename Consumer>
requires CompactedFragmentsConsumer<Consumer>
struct compact_for_query : compact_mutation<only_live, compact_for_sstables::no, Consumer, noop_compacted_fragments_consumer> {
using compact_mutation<only_live, compact_for_sstables::no, Consumer, noop_compacted_fragments_consumer>::compact_mutation;
};
template<emit_only_live_rows OnlyLive>
using compact_for_query_state = compact_mutation_state<OnlyLive, compact_for_sstables::no>;
template<emit_only_live_rows only_live, typename Consumer>
requires CompactedFragmentsConsumerV2<Consumer>
struct compact_for_query_v2 : compact_mutation_v2<only_live, compact_for_sstables::no, Consumer, noop_compacted_fragments_consumer> {
@@ -768,7 +628,7 @@ struct compact_for_query_v2 : compact_mutation_v2<only_live, compact_for_sstable
};
template<emit_only_live_rows OnlyLive>
using compact_for_query_state_v2 = compact_mutation_state<OnlyLive, compact_for_sstables::no, compactor_output_format::v2>;
using compact_for_query_state_v2 = compact_mutation_state<OnlyLive, compact_for_sstables::no>;
template<typename Consumer, typename GCConsumer = noop_compacted_fragments_consumer>
requires CompactedFragmentsConsumerV2<Consumer> && CompactedFragmentsConsumerV2<GCConsumer>

25
mutation_partition.cc
View File

@@ -1949,8 +1949,7 @@ stop_iteration query_result_builder::consume(clustering_row&& cr, row_tombstone
_stop = _mutation_consumer->consume(std::move(cr), t);
return _stop;
}
stop_iteration query_result_builder::consume(range_tombstone&& rt) {
_stop = _mutation_consumer->consume(std::move(rt));
stop_iteration query_result_builder::consume(range_tombstone_change&& rtc) {
return _stop;
}
@@ -1988,6 +1987,7 @@ stop_iteration query::result_memory_accounter::check_local_limit() const {
}
void reconcilable_result_builder::consume_new_partition(const dht::decorated_key& dk) {
_rt_assembler.reset();
_return_static_content_on_partition_with_no_rows =
_slice.options.contains(query::partition_slice::option::always_return_static_content) ||
!has_ck_selector(_slice.row_ranges(_schema, dk.key()));
@@ -2007,6 +2007,11 @@ stop_iteration reconcilable_result_builder::consume(static_row&& sr, tombstone,
}
stop_iteration reconcilable_result_builder::consume(clustering_row&& cr, row_tombstone, bool is_alive) {
if (_rt_assembler.needs_flush()) {
if (auto rt_opt = _rt_assembler.flush(_schema, position_in_partition::after_key(cr.key()))) {
consume(std::move(*rt_opt));
}
}
_live_rows += is_alive;
auto stop = _memory_accounter.update_and_check(cr.memory_usage(_schema));
if (is_alive) {
@@ -2029,7 +2034,15 @@ stop_iteration reconcilable_result_builder::consume(range_tombstone&& rt) {
return _mutation_consumer->consume(std::move(rt));
}
stop_iteration reconcilable_result_builder::consume(range_tombstone_change&& rtc) {
if (auto rt_opt = _rt_assembler.consume(_schema, std::move(rtc))) {
return consume(std::move(*rt_opt));
}
return stop_iteration::no;
}
stop_iteration reconcilable_result_builder::consume_end_of_partition() {
_rt_assembler.on_end_of_stream();
if (_live_rows == 0 && _static_row_is_alive && _return_static_content_on_partition_with_no_rows) {
++_live_rows;
// Normally we count only live clustering rows, to guarantee that
@@ -2056,7 +2069,7 @@ to_data_query_result(const reconcilable_result& r, schema_ptr s, const query::pa
query::result_options opts) {
// This result was already built with a limit, don't apply another one.
query::result::builder builder(slice, opts, query::result_memory_accounter{ query::result_memory_limiter::unlimited_result_size });
auto consumer = compact_for_query<emit_only_live_rows::yes, query_result_builder>(*s, gc_clock::time_point::min(), slice, max_rows,
auto consumer = compact_for_query_v2<emit_only_live_rows::yes, query_result_builder>(*s, gc_clock::time_point::min(), slice, max_rows,
max_partitions, query_result_builder(*s, builder));
const auto reverse = slice.options.contains(query::partition_slice::option::reversed) ? consume_in_reverse::legacy_half_reverse : consume_in_reverse::no;
@@ -2075,7 +2088,7 @@ to_data_query_result(const reconcilable_result& r, schema_ptr s, const query::pa
query::result
query_mutation(mutation&& m, const query::partition_slice& slice, uint64_t row_limit, gc_clock::time_point now, query::result_options opts) {
query::result::builder builder(slice, opts, query::result_memory_accounter{ query::result_memory_limiter::unlimited_result_size });
auto consumer = compact_for_query<emit_only_live_rows::yes, query_result_builder>(*m.schema(), now, slice, row_limit,
auto consumer = compact_for_query_v2<emit_only_live_rows::yes, query_result_builder>(*m.schema(), now, slice, row_limit,
query::max_partitions, query_result_builder(*m.schema(), builder));
const auto reverse = slice.options.contains(query::partition_slice::option::reversed) ? consume_in_reverse::legacy_half_reverse : consume_in_reverse::no;
std::move(m).consume(consumer, reverse);
@@ -2099,7 +2112,7 @@ public:
_mutation->partition().insert_row(_schema, cr.key(), std::move(cr).as_deletable_row());
return stop_iteration::no;
}
stop_iteration consume(range_tombstone&& rt) {
stop_iteration consume(range_tombstone_change&& rtc) {
return stop_iteration::no;
}
stop_iteration consume_end_of_partition() {
@@ -2298,7 +2311,7 @@ future<mutation_opt> counter_write_query(schema_ptr s, const mutation_source& so
// do_with() doesn't support immovable objects
auto r_a_r = std::make_unique<range_and_reader>(s, source, std::move(permit), dk, slice, std::move(trace_ptr));
auto cwqrb = counter_write_query_result_builder(*s);
auto cfq = compact_for_query<emit_only_live_rows::yes, counter_write_query_result_builder>(
auto cfq = compact_for_query_v2<emit_only_live_rows::yes, counter_write_query_result_builder>(
*s, gc_clock::now(), slice, query::max_rows, query::max_partitions, std::move(cwqrb));
auto f = r_a_r->reader.consume(std::move(cfq));
return f.finally([r_a_r = std::move(r_a_r)] {

8
mutation_query.hh
View File

@@ -14,6 +14,7 @@
#include "db/timeout_clock.hh"
#include "mutation.hh"
#include "utils/chunked_vector.hh"
#include "range_tombstone_assembler.hh"
class reconcilable_result;
class frozen_reconcilable_result;
@@ -132,10 +133,15 @@ class reconcilable_result_builder {
query::result_memory_accounter _memory_accounter;
stop_iteration _stop;
std::optional<streamed_mutation_freezer> _mutation_consumer;
range_tombstone_assembler _rt_assembler;
uint64_t _live_rows{};
// make this the last member so it is destroyed first. #7240
utils::chunked_vector<partition> _result;
private:
stop_iteration consume(range_tombstone&& rt);
public:
// Expects table schema (non-reversed) and half-reversed (legacy) slice when building results for reverse query.
reconcilable_result_builder(const schema& s, const query::partition_slice& slice,
@@ -148,7 +154,7 @@ public:
void consume(tombstone t);
stop_iteration consume(static_row&& sr, tombstone, bool is_alive);
stop_iteration consume(clustering_row&& cr, row_tombstone, bool is_alive);
stop_iteration consume(range_tombstone&& rt);
stop_iteration consume(range_tombstone_change&& rtc);
stop_iteration consume_end_of_partition();
reconcilable_result consume_end_of_stream();
};

4
mutation_reader.cc
View File

@@ -2307,11 +2307,11 @@ std::pair<flat_mutation_reader_v2, queue_reader_handle_v2> make_queue_reader_v2(
namespace {
class compacting_reader : public flat_mutation_reader_v2::impl {
friend class compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::yes, compactor_output_format::v2>;
friend class compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::yes>;
private:
flat_mutation_reader_v2 _reader;
compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::yes, compactor_output_format::v2> _compactor;
compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::yes> _compactor;
noop_compacted_fragments_consumer _gc_consumer;
// Uncompacted stream

49
querier.hh
View File

@@ -44,8 +44,8 @@ public:
*_last_ckey = cr.key();
return _consumer.consume(std::move(cr), std::move(t), is_live);
}
stop_iteration consume(range_tombstone&& rt) {
return _consumer.consume(std::move(rt));
stop_iteration consume(range_tombstone_change&& rtc) {
return _consumer.consume(std::move(rtc));
}
stop_iteration consume_end_of_partition() {
return _consumer.consume_end_of_partition();
@@ -63,42 +63,9 @@ public:
/// or std::nullopt if the last row wasn't a clustering row, and whatever the
/// consumer's `consume_end_of_stream()` method returns.
template <emit_only_live_rows OnlyLive, typename Consumer>
requires CompactedFragmentsConsumer<Consumer>
auto consume_page(flat_mutation_reader& reader,
lw_shared_ptr<compact_for_query_state<OnlyLive>> compaction_state,
const query::partition_slice& slice,
Consumer&& consumer,
uint64_t row_limit,
uint32_t partition_limit,
gc_clock::time_point query_time) {
return reader.peek().then([=, &reader, consumer = std::move(consumer), &slice] (
mutation_fragment* next_fragment) mutable {
const auto next_fragment_region = next_fragment ? next_fragment->position().region() : partition_region::partition_end;
compaction_state->start_new_page(row_limit, partition_limit, query_time, next_fragment_region, consumer);
auto last_ckey = make_lw_shared<std::optional<clustering_key_prefix>>();
auto reader_consumer = compact_for_query<OnlyLive, clustering_position_tracker<Consumer>>(
compaction_state,
clustering_position_tracker(std::move(consumer), last_ckey));
return reader.consume(std::move(reader_consumer)).then([last_ckey] (auto&&... results) mutable {
static_assert(sizeof...(results) <= 1);
return make_ready_future<std::tuple<std::optional<clustering_key_prefix>, std::decay_t<decltype(results)>...>>(std::tuple(std::move(*last_ckey), std::move(results)...));
});
});
}
/// Consume a page worth of data from the reader.
///
/// Uses `compaction_state` for compacting the fragments and `consumer` for
/// building the results.
/// Returns a future containing a tuple with the last consumed clustering key,
/// or std::nullopt if the last row wasn't a clustering row, and whatever the
/// consumer's `consume_end_of_stream()` method returns.
template <emit_only_live_rows OnlyLive, typename Consumer>
requires CompactedFragmentsConsumer<Consumer>
requires CompactedFragmentsConsumerV2<Consumer>
auto consume_page(flat_mutation_reader_v2& reader,
lw_shared_ptr<compact_for_query_state<OnlyLive>> compaction_state,
lw_shared_ptr<compact_for_query_state_v2<OnlyLive>> compaction_state,
const query::partition_slice& slice,
Consumer&& consumer,
uint64_t row_limit,
@@ -110,7 +77,7 @@ auto consume_page(flat_mutation_reader_v2& reader,
compaction_state->start_new_page(row_limit, partition_limit, query_time, next_fragment_region, consumer);
auto last_ckey = make_lw_shared<std::optional<clustering_key_prefix>>();
auto reader_consumer = compact_for_query<OnlyLive, clustering_position_tracker<Consumer>>(
auto reader_consumer = compact_for_query_v2<OnlyLive, clustering_position_tracker<Consumer>>(
compaction_state,
clustering_position_tracker(std::move(consumer), last_ckey));
@@ -211,7 +178,7 @@ public:
/// instead.
template <emit_only_live_rows OnlyLive>
class querier : public querier_base {
lw_shared_ptr<compact_for_query_state<OnlyLive>> _compaction_state;
lw_shared_ptr<compact_for_query_state_v2<OnlyLive>> _compaction_state;
std::optional<clustering_key_prefix> _last_ckey;
public:
@@ -223,7 +190,7 @@ public:
const io_priority_class& pc,
tracing::trace_state_ptr trace_ptr)
: querier_base(schema, permit, std::move(range), std::move(slice), ms, pc, std::move(trace_ptr))
, _compaction_state(make_lw_shared<compact_for_query_state<OnlyLive>>(*schema, gc_clock::time_point{}, *_slice, 0, 0)) {
, _compaction_state(make_lw_shared<compact_for_query_state_v2<OnlyLive>>(*schema, gc_clock::time_point{}, *_slice, 0, 0)) {
}
bool are_limits_reached() const {
@@ -231,7 +198,7 @@ public:
}
template <typename Consumer>
requires CompactedFragmentsConsumer<Consumer>
requires CompactedFragmentsConsumerV2<Consumer>
auto consume_page(Consumer&& consumer,
uint64_t row_limit,
uint32_t partition_limit,

6
query-result-writer.hh
View File

@@ -185,7 +185,7 @@ public:
class row;
class static_row;
class clustering_row;
class range_tombstone;
class range_tombstone_change;
// Adds mutation to query::result.
class mutation_querier {
@@ -207,7 +207,7 @@ public:
stop_iteration consume(static_row&& sr, tombstone current_tombstone);
// Requires that cr.has_any_live_data()
stop_iteration consume(clustering_row&& cr, row_tombstone current_tombstone);
stop_iteration consume(range_tombstone&&) { return stop_iteration::no; }
stop_iteration consume(range_tombstone_change&&) { return stop_iteration::no; }
uint64_t consume_end_of_stream();
};
@@ -223,7 +223,7 @@ public:
void consume(tombstone t);
stop_iteration consume(static_row&& sr, tombstone t, bool);
stop_iteration consume(clustering_row&& cr, row_tombstone t, bool);
stop_iteration consume(range_tombstone&& rt);
stop_iteration consume(range_tombstone_change&& rtc);
stop_iteration consume_end_of_partition();
void consume_end_of_stream();
};

125
test/boost/mutation_test.cc
View File

@@ -2689,37 +2689,6 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
ref_mut.partition().compact_for_query(*s, pk, query_time, {query::clustering_range::make_open_ended_both_sides()}, true, false, max_rows);
}
struct consumer {
reader_permit permit;
mutation& mut;
const uint64_t row_limit;
uint64_t rows = 0;
void consume_new_partition(const dht::decorated_key& dk) {
BOOST_REQUIRE(mut.decorated_key().equal(*mut.schema(), dk));
}
void consume(const tombstone& t) {
BOOST_REQUIRE_EQUAL(t, mut.partition().partition_tombstone());
}
stop_iteration consume(static_row&& sr, tombstone, bool) {
mut.apply(mutation_fragment(*mut.schema(), permit, std::move(sr)));
return stop_iteration(++rows >= row_limit);
}
stop_iteration consume(clustering_row&& cr, row_tombstone t, bool is_alive) {
mut.apply(mutation_fragment(*mut.schema(), permit, std::move(cr)));
return stop_iteration(++rows >= row_limit);
}
stop_iteration consume(range_tombstone&& rt) {
mut.apply(mutation_fragment(*mut.schema(), permit, std::move(rt)));
return stop_iteration(++rows >= row_limit);
}
stop_iteration consume_end_of_partition() {
return stop_iteration::yes;
}
void consume_end_of_stream() {
}
};
struct consumer_v2 {
reader_permit permit;
mutation& mut;
@@ -2762,18 +2731,6 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
}
};
testlog.info("non-paged");
{
mutation res_mut(s, pk);
auto c = compact_for_query<emit_only_live_rows::no, consumer>(*s, query_time, s->full_slice(), max_rows, max_partitions, consumer{permit, res_mut, max_rows});
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
reader.consume(std::move(c)).get();
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("non-paged v2");
{
mutation res_mut(s, pk);
@@ -2786,26 +2743,10 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("limited pages");
{
mutation res_mut(s, pk);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), 1, max_partitions);
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
while (!reader.is_buffer_empty() || !reader.is_end_of_stream()) {
auto c = consumer{permit, res_mut, max_rows};
compaction_state->start_new_page(1, max_partitions, query_time, reader.peek().get()->position().region(), c);
reader.consume(compact_for_query<emit_only_live_rows::no, consumer>(compaction_state, std::move(c))).get();
}
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("limited pages v2");
{
mutation res_mut(s, pk);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no, compactor_output_format::v2>>(*s, query_time, s->full_slice(), 1, max_partitions);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), 1, max_partitions);
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
@@ -2818,29 +2759,13 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("short pages");
testlog.info("short pages v2");
{
mutation res_mut(s, pk);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), max_rows, max_partitions);
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
while (!reader.is_buffer_empty() || !reader.is_end_of_stream()) {
auto c = consumer{permit, res_mut, 2};
compaction_state->start_new_page(max_rows, max_partitions, query_time, reader.peek().get()->position().region(), c);
reader.consume(compact_for_query<emit_only_live_rows::no, consumer>(compaction_state, std::move(c))).get();
}
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("short pages v2");
{
mutation res_mut(s, pk);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no, compactor_output_format::v2>>(*s, query_time, s->full_slice(), max_rows, max_partitions);
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
while (!reader.is_buffer_empty() || !reader.is_end_of_stream()) {
auto c = consumer_v2{permit, res_mut, 2};
compaction_state->start_new_page(max_rows, max_partitions, query_time, reader.peek().get()->position().region(), c);
@@ -2850,27 +2775,6 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("limited pages - detach state");
{
mutation res_mut(s, pk);
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
std::optional<detached_compaction_state> detached_state;
while (!reader.is_buffer_empty() || !reader.is_end_of_stream()) {
if (detached_state) {
restore_state(reader, std::move(*detached_state));
}
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), 1, max_partitions);
auto c = consumer{permit, res_mut, max_rows};
reader.consume(compact_for_query<emit_only_live_rows::no, consumer>(compaction_state, std::move(c))).get();
detached_state = std::move(*compaction_state).detach_state();
}
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("limited pages - detach state v2");
{
mutation res_mut(s, pk);
@@ -2883,7 +2787,7 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
if (detached_state) {
restore_state(reader, std::move(*detached_state));
}
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no, compactor_output_format::v2>>(*s, query_time, s->full_slice(), 1, max_partitions);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), 1, max_partitions);
auto c = consumer_v2{permit, res_mut, max_rows};
reader.consume(compact_for_query_v2<emit_only_live_rows::no, consumer_v2>(compaction_state, std::move(c))).get();
detached_state = std::move(*compaction_state).detach_state();
@@ -2892,27 +2796,6 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("short pages - detach state");
{
mutation res_mut(s, pk);
auto reader = make_flat_mutation_reader_from_fragments(s, permit, make_frags());
auto close_reader = deferred_close(reader);
std::optional<detached_compaction_state> detached_state;
while (!reader.is_buffer_empty() || !reader.is_end_of_stream()) {
if (detached_state) {
restore_state(reader, std::move(*detached_state));
}
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), max_rows, max_partitions);
auto c = consumer{permit, res_mut, 2};
reader.consume(compact_for_query<emit_only_live_rows::no, consumer>(compaction_state, std::move(c))).get();
detached_state = std::move(*compaction_state).detach_state();
}
BOOST_REQUIRE_EQUAL(res_mut, ref_mut);
}
testlog.info("short pages - detach state v2");
{
mutation res_mut(s, pk);
@@ -2925,7 +2808,7 @@ SEASTAR_THREAD_TEST_CASE(test_compactor_range_tombstone_spanning_many_pages) {
if (detached_state) {
restore_state(reader, std::move(*detached_state));
}
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no, compactor_output_format::v2>>(*s, query_time, s->full_slice(), max_rows, max_partitions);
auto compaction_state = make_lw_shared<compact_mutation_state<emit_only_live_rows::no, compact_for_sstables::no>>(*s, query_time, s->full_slice(), max_rows, max_partitions);
auto c = consumer_v2{permit, res_mut, 2};
reader.consume(compact_for_query_v2<emit_only_live_rows::no, consumer_v2>(compaction_state, std::move(c))).get();
detached_state = std::move(*compaction_state).detach_state();

2
test/boost/querier_cache_test.cc
View File

@@ -47,7 +47,7 @@ public:
_ck = cr.key();
return stop_iteration::no;
}
stop_iteration consume(range_tombstone&& rt) {
stop_iteration consume(range_tombstone_change&& rtc) {
return stop_iteration::no;
}
stop_iteration consume_end_of_partition() {