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

flat_mutation_reader: Split readers by file and remove unnecessary includes.

Browse Source 
The flat_mutation_reader files were conflated and contained multiple
readers, which were not strictly necessary. Splitting optimizes both
iterative compilation times, as touching rarely used readers doesn't
recompile large chunks of codebase. Total compilation times are also
improved, as the size of flat_mutation_reader.hh and
flat_mutation_reader_v2.hh have been reduced and those files are
included by many file in the codebase.

With changes

real	29m14.051s
user	168m39.071s
sys	5m13.443s

Without changes

real	30m36.203s
user	175m43.354s
sys	5m26.376s

Closes #10194
This commit is contained in:
Mikołaj Sielużycki
2022-03-10 12:07:32 +01:00
committed by Botond Dénes
parent 26b1be0b8f
commit 1d84a254c0
83 changed files with 2063 additions and 1566 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
cache_flat_mutation_reader.hh
View File

@@ -15,7 +15,8 @@
#include "query-request.hh"
#include "partition_snapshot_row_cursor.hh"
#include "read_context.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/delegating.hh"
#include "clustering_key_filter.hh"
namespace cache {

2
compaction/compaction_strategy.hh
View File

@@ -16,7 +16,7 @@
#include "sstables/shared_sstable.hh"
#include "exceptions/exceptions.hh"
#include "compaction_strategy_type.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "table_state.hh"
#include "strategy_control.hh"

3
configure.py
View File

@@ -698,7 +698,8 @@ scylla_core = (['replica/database.cc',
'mutation_partition_serializer.cc',
'converting_mutation_partition_applier.cc',
'mutation_reader.cc',
'flat_mutation_reader.cc',
'readers/mutation_reader.cc',
'readers/mutation_readers.cc',
'mutation_query.cc',
'keys.cc',
'counters.cc',

2
db/chained_delegating_reader.hh
View File

@@ -10,7 +10,7 @@
#include <seastar/core/shared_future.hh>
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
// A reader which allows to insert a deferring operation before reading.
// All calls will first wait for a future to resolve, then forward to a given underlying reader.

2
db/data_listeners.hh
View File

@@ -15,7 +15,7 @@
#include "utils/hash.hh"
#include "schema_fwd.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "mutation_reader.hh"
#include "utils/top_k.hh"
#include "schema_registry.hh"

1
db/size_estimates_virtual_reader.cc
View File

@@ -25,6 +25,7 @@
#include "replica/database.hh"
#include "db/size_estimates_virtual_reader.hh"
#include "readers/from_mutations.hh"
namespace db {

3
db/size_estimates_virtual_reader.hh
View File

@@ -8,7 +8,8 @@
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader.hh"
#include "db/system_keyspace.hh"
namespace replica {

2
db/view/build_progress_virtual_reader.hh
View File

@@ -10,7 +10,7 @@
#include "db/system_keyspace.hh"
#include "db/timeout_clock.hh"
#include "dht/i_partitioner.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "mutation_fragment.hh"
#include "mutation_reader.hh"
#include "query-request.hh"

1
db/view/view.cc
View File

@@ -56,6 +56,7 @@
#include "utils/exponential_backoff_retry.hh"
#include "utils/fb_utilities.hh"
#include "query-result-writer.hh"
#include "readers/from_fragments.hh"
using namespace std::chrono_literals;

2
db/view/view.hh
View File

@@ -12,7 +12,7 @@
#include "gc_clock.hh"
#include "query-request.hh"
#include "schema_fwd.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "frozen_mutation.hh"
class frozen_mutation_and_schema;

1
db/view/view_updating_consumer.hh
View File

@@ -19,6 +19,7 @@
#include "mutation_rebuilder.hh"
class evictable_reader_handle;
class evictable_reader_handle_v2;
namespace db::view {

2
db/virtual_table.cc
View File

@@ -12,6 +12,8 @@
#include "db/virtual_table.hh"
#include "db/chained_delegating_reader.hh"
#include "readers/reversing.hh"
#include "readers/forwardable.hh"
namespace db {

36
dht/i_partitioner_fwd.hh Normal file
View File

@@ -0,0 +1,36 @@
/*
* Modified by ScyllaDB
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: (AGPL-3.0-or-later and Apache-2.0)
*/
#pragma once
#include <vector>
#include "range.hh"
namespace sstables {
class key_view;
class decorated_key_view;
}
namespace dht {
class decorated_key;
class ring_position;
class token;
using partition_range = nonwrapping_range<ring_position>;
using token_range = nonwrapping_range<token>;
using partition_range_vector = std::vector<partition_range>;
using token_range_vector = std::vector<token_range>;
class decorated_key;
using decorated_key_opt = std::optional<decorated_key>;
}

2
frozen_mutation.cc
View File

@@ -26,7 +26,7 @@
#include "idl/uuid.dist.impl.hh"
#include "idl/keys.dist.impl.hh"
#include "idl/mutation.dist.impl.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "converting_mutation_partition_applier.hh"
#include "mutation_partition_view.hh"

2
index/built_indexes_virtual_reader.hh
View File

@@ -8,7 +8,7 @@
#include "replica/database.hh"
#include "db/system_keyspace.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "mutation_fragment_v2.hh"
#include "mutation_reader.hh"
#include "query-request.hh"

2
message/messaging_service.cc
View File

@@ -88,7 +88,7 @@
#include <boost/range/adaptor/filtered.hpp>
#include <boost/range/adaptor/indirected.hpp>
#include "frozen_mutation.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "streaming/stream_manager.hh"
#include "streaming/stream_mutation_fragments_cmd.hh"
#include "locator/snitch_base.hh"

2
mutation.cc
View File

@@ -8,7 +8,7 @@
#include "mutation.hh"
#include "query-result-writer.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "mutation_rebuilder.hh"
mutation::data::data(dht::decorated_key&& key, schema_ptr&& schema)

27
mutation_fragment.hh
View File

@@ -18,6 +18,7 @@
#include "db/timeout_clock.hh"
#include "reader_permit.hh"
#include "mutation_fragment_fwd.hh"
// mutation_fragments are the objects that streamed_mutation are going to
// stream. They can represent:
@@ -525,32 +526,6 @@ inline position_in_partition_view partition_end::position() const
std::ostream& operator<<(std::ostream&, partition_region);
std::ostream& operator<<(std::ostream&, mutation_fragment::kind);
using mutation_fragment_opt = optimized_optional<mutation_fragment>;
namespace streamed_mutation {
// Determines whether streamed_mutation is in forwarding mode or not.
//
// In forwarding mode the stream does not return all fragments right away,
// but only those belonging to the current clustering range. Initially
// current range only covers the static row. The stream can be forwarded
// (even before end-of- stream) to a later range with fast_forward_to().
// Forwarding doesn't change initial restrictions of the stream, it can
// only be used to skip over data.
//
// Monotonicity of positions is preserved by forwarding. That is fragments
// emitted after forwarding will have greater positions than any fragments
// emitted before forwarding.
//
// For any range, all range tombstones relevant for that range which are
// present in the original stream will be emitted. Range tombstones
// emitted before forwarding which overlap with the new range are not
// necessarily re-emitted.
//
// When streamed_mutation is not in forwarding mode, fast_forward_to()
// cannot be used.
class forwarding_tag;
using forwarding = bool_class<forwarding_tag>;
}
// range_tombstone_stream is a helper object that simplifies producing a stream
// of range tombstones and merging it with a stream of clustering rows.

20
mutation_fragment_fwd.hh Normal file
View File

@@ -0,0 +1,20 @@
/*
* Copyright (C) 2016-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include <seastar/util/bool_class.hh>
#include <seastar/util/optimized_optional.hh>
using namespace seastar;
class mutation_fragment;
class mutation_fragment_v2;
using mutation_fragment_opt = optimized_optional<mutation_fragment>;
using mutation_fragment_v2_opt = optimized_optional<mutation_fragment_v2>;

3
mutation_fragment_v2.hh
View File

@@ -354,9 +354,6 @@ private:
std::ostream& operator<<(std::ostream&, mutation_fragment_v2::kind);
using mutation_fragment_v2_opt = optimized_optional<mutation_fragment_v2>;
// F gets a stream element as an argument and returns the new value which replaces that element
// in the transformed stream.
template<typename F>

4
mutation_reader.cc
View File

@@ -16,10 +16,12 @@
#include <seastar/util/closeable.hh>
#include "mutation_reader.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/empty.hh"
#include "schema_registry.hh"
#include "mutation_compactor.hh"
#include "dht/sharder.hh"
#include "readers/empty_v2.hh"
logging::logger mrlog("mutation_reader");

5
mutation_reader.hh
View File

@@ -14,9 +14,10 @@
#include <seastar/core/future-util.hh>
#include <seastar/core/do_with.hh>
#include "tracing/trace_state.hh"
#include "flat_mutation_reader.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "reader_concurrency_semaphore.hh"
#include <seastar/core/io_priority_class.hh>
class reader_selector {
protected:

2
mutation_writer/feed_writers.hh
View File

@@ -8,7 +8,7 @@
#pragma once
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "mutation_reader.hh"
#include "seastar/core/coroutine.hh"

2
mutation_writer/multishard_writer.hh
View File

@@ -9,7 +9,7 @@
#pragma once
#include "schema_fwd.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "dht/i_partitioner.hh"
#include "utils/phased_barrier.hh"

4
partition_snapshot_reader.hh
View File

@@ -9,8 +9,8 @@
#pragma once
#include "partition_version.hh"
#include "flat_mutation_reader.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "clustering_key_filter.hh"
#include "query-request.hh"
#include <boost/range/algorithm/heap_algorithm.hpp>

1
reader_concurrency_semaphore.cc
View File

@@ -8,6 +8,7 @@
#include <seastar/core/seastar.hh>
#include <seastar/core/print.hh>
#include <seastar/core/file.hh>
#include <seastar/util/lazy.hh>
#include <seastar/util/log.hh>
#include <seastar/core/coroutine.hh>

2
reader_concurrency_semaphore.hh
View File

@@ -14,7 +14,7 @@
#include <seastar/core/queue.hh>
#include <seastar/core/expiring_fifo.hh>
#include "reader_permit.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
namespace bi = boost::intrusive;

18
readers/conversion.hh Normal file
View File

@@ -0,0 +1,18 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
class flat_mutation_reader;
class flat_mutation_reader_v2;
// Adapts a v2 reader to v1 reader
flat_mutation_reader downgrade_to_v1(flat_mutation_reader_v2);
// Adapts a v1 reader to v2 reader
flat_mutation_reader_v2 upgrade_to_v2(flat_mutation_reader);

42
readers/delegating.hh Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "readers/flat_mutation_reader.hh"
class flat_mutation_reader;
flat_mutation_reader make_delegating_reader(flat_mutation_reader&);
class delegating_reader : public flat_mutation_reader::impl {
flat_mutation_reader_opt _underlying_holder;
flat_mutation_reader* _underlying;
public:
// when passed a lvalue reference to the reader
// we don't own it and the caller is responsible
// for evenetually closing the reader.
delegating_reader(flat_mutation_reader& r)
: impl(r.schema(), r.permit())
, _underlying_holder()
, _underlying(&r)
{ }
// when passed a rvalue reference to the reader
// we assume ownership of it and will close it
// in close().
delegating_reader(flat_mutation_reader&& r)
: impl(r.schema(), r.permit())
, _underlying_holder(std::move(r))
, _underlying(&*_underlying_holder)
{ }
virtual future<> fill_buffer() override;
virtual future<> fast_forward_to(position_range pr) override;
virtual future<> next_partition() override;
virtual future<> fast_forward_to(const dht::partition_range& pr) override;
virtual future<> close() noexcept override;
};

67
readers/delegating_v2.hh Normal file
View File

@@ -0,0 +1,67 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "readers/flat_mutation_reader_v2.hh"
class delegating_reader_v2 : public flat_mutation_reader_v2::impl {
flat_mutation_reader_v2_opt _underlying_holder;
flat_mutation_reader_v2* _underlying;
public:
// when passed a lvalue reference to the reader
// we don't own it and the caller is responsible
// for evenetually closing the reader.
delegating_reader_v2(flat_mutation_reader_v2& r)
: impl(r.schema(), r.permit())
, _underlying_holder()
, _underlying(&r)
{ }
// when passed a rvalue reference to the reader
// we assume ownership of it and will close it
// in close().
delegating_reader_v2(flat_mutation_reader_v2&& r)
: impl(r.schema(), r.permit())
, _underlying_holder(std::move(r))
, _underlying(&*_underlying_holder)
{ }
virtual future<> fill_buffer() override {
if (is_buffer_full()) {
return make_ready_future<>();
}
return _underlying->fill_buffer().then([this] {
_end_of_stream = _underlying->is_end_of_stream();
_underlying->move_buffer_content_to(*this);
});
}
virtual future<> fast_forward_to(position_range pr) override {
_end_of_stream = false;
forward_buffer_to(pr.start());
return _underlying->fast_forward_to(std::move(pr));
}
virtual future<> next_partition() override {
clear_buffer_to_next_partition();
auto maybe_next_partition = make_ready_future<>();
if (is_buffer_empty()) {
maybe_next_partition = _underlying->next_partition();
}
return maybe_next_partition.then([this] {
_end_of_stream = _underlying->is_end_of_stream() && _underlying->is_buffer_empty();
});
}
virtual future<> fast_forward_to(const dht::partition_range& pr) override {
_end_of_stream = false;
clear_buffer();
return _underlying->fast_forward_to(pr);
}
virtual future<> close() noexcept override {
return _underlying_holder ? _underlying_holder->close() : make_ready_future<>();
}
};
flat_mutation_reader_v2 make_delegating_reader_v2(flat_mutation_reader_v2&);

16
readers/empty.hh Normal file
View File

@@ -0,0 +1,16 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
class flat_mutation_reader;
class reader_permit;
flat_mutation_reader make_empty_flat_reader(schema_ptr s, reader_permit permit);

16
readers/empty_v2.hh Normal file
View File

@@ -0,0 +1,16 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
class flat_mutation_reader_v2;
class reader_permit;
flat_mutation_reader_v2 make_empty_flat_reader_v2(schema_ptr s, reader_permit permit);

174
flat_mutation_reader.hh → readers/flat_mutation_reader.hh
View File

@@ -14,22 +14,11 @@
#include "dht/i_partitioner.hh"
#include "mutation_fragment.hh"
#include "tracing/trace_state.hh"
#include "mutation.hh"
#include "mutation_consumer_concepts.hh"
#include <seastar/core/thread.hh>
#include <seastar/core/file.hh>
#include "reader_permit.hh"
#include <deque>
using seastar::future;
class mutation_source;
class position_in_partition;
class flat_mutation_reader_v2;
#include "readers/flat_mutation_reader_fwd.hh"
/// \brief Represents a stream of mutation fragments.
///
@@ -405,9 +394,6 @@ private:
friend flat_mutation_reader downgrade_to_v1(flat_mutation_reader_v2);
friend flat_mutation_reader_v2 upgrade_to_v2(flat_mutation_reader);
public:
// Documented in mutation_reader::forwarding.
class partition_range_forwarding_tag;
using partition_range_forwarding = bool_class<partition_range_forwarding_tag>;
flat_mutation_reader(std::unique_ptr<impl> impl) noexcept : _impl(std::move(impl)) {}
flat_mutation_reader(const flat_mutation_reader&) = delete;
@@ -666,7 +652,33 @@ namespace mutation_reader {
// from streamed_mutation::forwarding - the former is about skipping to
// a different partition range, while the latter is about skipping
// inside a large partition.
using forwarding = flat_mutation_reader::partition_range_forwarding;
class partition_range_forwarding_tag;
using forwarding = bool_class<partition_range_forwarding_tag>;
}
namespace streamed_mutation {
// Determines whether streamed_mutation is in forwarding mode or not.
//
// In forwarding mode the stream does not return all fragments right away,
// but only those belonging to the current clustering range. Initially
// current range only covers the static row. The stream can be forwarded
// (even before end-of- stream) to a later range with fast_forward_to().
// Forwarding doesn't change initial restrictions of the stream, it can
// only be used to skip over data.
//
// Monotonicity of positions is preserved by forwarding. That is fragments
// emitted after forwarding will have greater positions than any fragments
// emitted before forwarding.
//
// For any range, all range tombstones relevant for that range which are
// present in the original stream will be emitted. Range tombstones
// emitted before forwarding which overlap with the new range are not
// necessarily re-emitted.
//
// When streamed_mutation is not in forwarding mode, fast_forward_to()
// cannot be used.
class forwarding_tag;
using forwarding = bool_class<forwarding_tag>;
}
// Consumes mutation fragments until StopCondition is true.
@@ -752,102 +764,6 @@ flat_mutation_reader transform(flat_mutation_reader r, T t) {
return make_flat_mutation_reader<transforming_reader>(std::move(r), std::move(t));
}
class delegating_reader : public flat_mutation_reader::impl {
flat_mutation_reader_opt _underlying_holder;
flat_mutation_reader* _underlying;
public:
// when passed a lvalue reference to the reader
// we don't own it and the caller is responsible
// for evenetually closing the reader.
delegating_reader(flat_mutation_reader& r)
: impl(r.schema(), r.permit())
, _underlying_holder()
, _underlying(&r)
{ }
// when passed a rvalue reference to the reader
// we assume ownership of it and will close it
// in close().
delegating_reader(flat_mutation_reader&& r)
: impl(r.schema(), r.permit())
, _underlying_holder(std::move(r))
, _underlying(&*_underlying_holder)
{ }
virtual future<> fill_buffer() override {
if (is_buffer_full()) {
return make_ready_future<>();
}
return _underlying->fill_buffer().then([this] {
_end_of_stream = _underlying->is_end_of_stream();
_underlying->move_buffer_content_to(*this);
});
}
virtual future<> fast_forward_to(position_range pr) override {
_end_of_stream = false;
forward_buffer_to(pr.start());
return _underlying->fast_forward_to(std::move(pr));
}
virtual future<> next_partition() override {
clear_buffer_to_next_partition();
auto maybe_next_partition = make_ready_future<>();
if (is_buffer_empty()) {
maybe_next_partition = _underlying->next_partition();
}
return maybe_next_partition.then([this] {
_end_of_stream = _underlying->is_end_of_stream() && _underlying->is_buffer_empty();
});
}
virtual future<> fast_forward_to(const dht::partition_range& pr) override {
_end_of_stream = false;
clear_buffer();
return _underlying->fast_forward_to(pr);
}
virtual future<> close() noexcept override {
return _underlying_holder ? _underlying_holder->close() : make_ready_future<>();
}
};
flat_mutation_reader make_delegating_reader(flat_mutation_reader&);
flat_mutation_reader make_forwardable(flat_mutation_reader m);
flat_mutation_reader make_nonforwardable(flat_mutation_reader, bool);
flat_mutation_reader make_empty_flat_reader(schema_ptr s, reader_permit permit);
// All mutations should have the same schema.
flat_mutation_reader make_flat_mutation_reader_from_mutations(schema_ptr schema, reader_permit permit, std::vector<mutation>,
const dht::partition_range& pr = query::full_partition_range, streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
inline flat_mutation_reader make_flat_mutation_reader_from_mutations(schema_ptr schema, reader_permit permit, std::vector<mutation> ms, streamed_mutation::forwarding fwd) {
return make_flat_mutation_reader_from_mutations(std::move(schema), std::move(permit), std::move(ms), query::full_partition_range, fwd);
}
// All mutations should have the same schema.
flat_mutation_reader
make_flat_mutation_reader_from_mutations(schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
flat_mutation_reader
make_flat_mutation_reader_from_mutations(schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const dht::partition_range& pr,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
flat_mutation_reader
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment>);
flat_mutation_reader
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment>, const dht::partition_range& pr);
flat_mutation_reader
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment>, const dht::partition_range& pr, const query::partition_slice& slice);
// Calls the consumer for each element of the reader's stream until end of stream
// is reached or the consumer requests iteration to stop by returning stop_iteration::yes.
// The consumer should accept mutation as the argument and return stop_iteration.
@@ -868,40 +784,6 @@ future<> consume_partitions(flat_mutation_reader& reader, Consumer consumer) {
});
}
flat_mutation_reader
make_generating_reader(schema_ptr s, reader_permit permit, std::function<future<mutation_fragment_opt> ()> get_next_fragment);
/// A reader that emits partitions in native reverse order.
///
/// 1. The reader's schema() method will return a reversed schema (see
/// \ref schema::make_reversed()).
/// 2. Static row is still emitted first.
/// 3. Range tombstones' bounds are reversed (see \ref range_tombstone::reverse()).
/// 4. Clustered rows and range tombstones are emitted in descending order.
/// Because of 3 and 4 the guarantee that a range tombstone is emitted before
/// any mutation fragment affected by it still holds.
/// Ordering of partitions themselves remains unchanged.
/// For more details see docs/design-notes/reverse-reads.md.
///
/// The reader's schema (returned by `schema()`) is the reverse of `original`'s schema.
///
/// \param original the reader to be reversed.
/// \param max_size the maximum amount of memory the reader is allowed to use
/// for reversing and conversely the maximum size of the results. The
/// reverse reader reads entire partitions into memory, before reversing
/// them. Since partitions can be larger than the available memory, we need
/// to enforce a limit on memory consumption. When reaching the soft limit
/// a warning will be logged. When reaching the hard limit the read will be
/// aborted.
/// \param slice serves as a convenience slice storage for reads that have to
/// store an edited slice somewhere. This is common for reads that work
/// with a native-reversed slice and so have to convert the one used in the
/// query -- which is in half-reversed format.
///
/// FIXME: reversing should be done in the sstable layer, see #1413.
flat_mutation_reader
make_reversing_reader(flat_mutation_reader original, query::max_result_size max_size, std::unique_ptr<query::partition_slice> slice = {});
/// A cosumer function that is passed a flat_mutation_reader to be consumed from
/// and returns a future<> resolved when the reader is fully consumed, and closed.
/// Note: the function assumes ownership of the reader and must close it in all cases.

26
readers/flat_mutation_reader_fwd.hh Normal file
View File

@@ -0,0 +1,26 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include <seastar/util/bool_class.hh>
using namespace seastar;
class mutation_source;
class position_in_partition;
class flat_mutation_reader_v2;
namespace streamed_mutation {
class forwarding_tag;
using forwarding = bool_class<forwarding_tag>;
}
namespace mutation_reader {
class partition_range_forwarding_tag;
using forwarding = bool_class<partition_range_forwarding_tag>;
}

147
flat_mutation_reader_v2.hh → readers/flat_mutation_reader_v2.hh
View File

@@ -10,23 +10,16 @@
#include <seastar/util/bool_class.hh>
#include <seastar/core/future.hh>
#include <seastar/core/circular_buffer.hh>
#include "dht/i_partitioner.hh"
#include "position_in_partition.hh"
#include "flat_mutation_reader.hh"
#include "mutation_fragment_v2.hh"
#include "tracing/trace_state.hh"
#include "mutation.hh"
#include "query_class_config.hh"
#include "mutation_consumer_concepts.hh"
#include <seastar/core/thread.hh>
#include <seastar/core/file.hh>
#include "reader_permit.hh"
#include <deque>
using seastar::future;
class flat_mutation_reader;
/// \brief Represents a stream of mutation fragments.
///
@@ -413,10 +406,6 @@ private:
friend flat_mutation_reader downgrade_to_v1(flat_mutation_reader_v2);
friend flat_mutation_reader_v2 upgrade_to_v2(flat_mutation_reader);
public:
// Documented in mutation_reader::forwarding.
class partition_range_forwarding_tag;
using partition_range_forwarding = bool_class<partition_range_forwarding_tag>;
flat_mutation_reader_v2(std::unique_ptr<impl> impl) noexcept : _impl(std::move(impl)) {}
flat_mutation_reader_v2(const flat_mutation_reader_v2&) = delete;
flat_mutation_reader_v2(flat_mutation_reader_v2&&) = default;
@@ -748,139 +737,10 @@ flat_mutation_reader_v2 transform(flat_mutation_reader_v2 r, T t) {
return make_flat_mutation_reader_v2<transforming_reader>(std::move(r), std::move(t));
}
class delegating_reader_v2 : public flat_mutation_reader_v2::impl {
flat_mutation_reader_v2_opt _underlying_holder;
flat_mutation_reader_v2* _underlying;
public:
// when passed a lvalue reference to the reader
// we don't own it and the caller is responsible
// for evenetually closing the reader.
delegating_reader_v2(flat_mutation_reader_v2& r)
: impl(r.schema(), r.permit())
, _underlying_holder()
, _underlying(&r)
{ }
// when passed a rvalue reference to the reader
// we assume ownership of it and will close it
// in close().
delegating_reader_v2(flat_mutation_reader_v2&& r)
: impl(r.schema(), r.permit())
, _underlying_holder(std::move(r))
, _underlying(&*_underlying_holder)
{ }
virtual future<> fill_buffer() override {
if (is_buffer_full()) {
return make_ready_future<>();
}
return _underlying->fill_buffer().then([this] {
_end_of_stream = _underlying->is_end_of_stream();
_underlying->move_buffer_content_to(*this);
});
}
virtual future<> fast_forward_to(position_range pr) override {
_end_of_stream = false;
forward_buffer_to(pr.start());
return _underlying->fast_forward_to(std::move(pr));
}
virtual future<> next_partition() override {
clear_buffer_to_next_partition();
auto maybe_next_partition = make_ready_future<>();
if (is_buffer_empty()) {
maybe_next_partition = _underlying->next_partition();
}
return maybe_next_partition.then([this] {
_end_of_stream = _underlying->is_end_of_stream() && _underlying->is_buffer_empty();
});
}
virtual future<> fast_forward_to(const dht::partition_range& pr) override {
_end_of_stream = false;
clear_buffer();
return _underlying->fast_forward_to(pr);
}
virtual future<> close() noexcept override {
return _underlying_holder ? _underlying_holder->close() : make_ready_future<>();
}
};
flat_mutation_reader_v2 make_delegating_reader_v2(flat_mutation_reader_v2&);
// Adapts a v2 reader to v1 reader
flat_mutation_reader downgrade_to_v1(flat_mutation_reader_v2);
// Adapts a v1 reader to v2 reader
flat_mutation_reader_v2 upgrade_to_v2(flat_mutation_reader);
// Reads a single partition from a reader. Returns empty optional if there are no more partitions to be read.
future<mutation_opt> read_mutation_from_flat_mutation_reader(flat_mutation_reader_v2&);
flat_mutation_reader_v2 make_forwardable(flat_mutation_reader_v2 m);
flat_mutation_reader_v2 make_empty_flat_reader_v2(schema_ptr s, reader_permit permit);
// All mutations should have the same schema.
flat_mutation_reader_v2 make_flat_mutation_reader_from_mutations_v2(schema_ptr schema, reader_permit permit, std::vector<mutation>,
const dht::partition_range& pr = query::full_partition_range, streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
inline flat_mutation_reader_v2 make_flat_mutation_reader_from_mutations_v2(schema_ptr schema, reader_permit permit, std::vector<mutation> ms, streamed_mutation::forwarding fwd) {
return make_flat_mutation_reader_from_mutations_v2(std::move(schema), std::move(permit), std::move(ms), query::full_partition_range, fwd);
}
// All mutations should have the same schema.
flat_mutation_reader_v2
make_flat_mutation_reader_from_mutations_v2(schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
flat_mutation_reader_v2
make_flat_mutation_reader_from_mutations_v2(schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const dht::partition_range& pr,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
/// Make a reader that enables the wrapped reader to work with multiple ranges.
///
/// \param ranges An range vector that has to contain strictly monotonic
/// partition ranges, such that successively calling
/// `flat_mutation_reader::fast_forward_to()` with each one is valid.
/// An range vector range with 0 or 1 elements is also valid.
/// \param fwd_mr It is only respected when `ranges` contains 0 or 1 partition
/// ranges. Otherwise the reader is created with
/// mutation_reader::forwarding::yes.
flat_mutation_reader_v2
make_flat_multi_range_reader(schema_ptr s, reader_permit permit, mutation_source source, const dht::partition_range_vector& ranges,
const query::partition_slice& slice, const io_priority_class& pc = default_priority_class(),
tracing::trace_state_ptr trace_state = nullptr,
flat_mutation_reader::partition_range_forwarding fwd_mr = flat_mutation_reader::partition_range_forwarding::yes);
/// Make a reader that enables the wrapped reader to work with multiple ranges.
///
/// Generator overload. The ranges returned by the generator have to satisfy the
/// same requirements as the `ranges` param of the vector overload.
flat_mutation_reader_v2
make_flat_multi_range_reader(
schema_ptr s,
reader_permit permit,
mutation_source source,
std::function<std::optional<dht::partition_range>()> generator,
const query::partition_slice& slice,
const io_priority_class& pc = default_priority_class(),
tracing::trace_state_ptr trace_state = nullptr,
flat_mutation_reader::partition_range_forwarding fwd_mr = flat_mutation_reader::partition_range_forwarding::yes);
flat_mutation_reader_v2
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment_v2>);
flat_mutation_reader_v2
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment_v2>, const dht::partition_range& pr);
flat_mutation_reader_v2
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment_v2>, const dht::partition_range& pr, const query::partition_slice& slice);
// Calls the consumer for each element of the reader's stream until end of stream
// is reached or the consumer requests iteration to stop by returning stop_iteration::yes.
// The consumer should accept mutation as the argument and return stop_iteration.
@@ -901,9 +761,6 @@ future<> consume_partitions(flat_mutation_reader_v2& reader, Consumer consumer)
});
}
flat_mutation_reader_v2
make_generating_reader(schema_ptr s, reader_permit permit, std::function<future<mutation_fragment_v2_opt> ()> get_next_fragment);
/// A cosumer function that is passed a flat_mutation_reader to be consumed from
/// and returns a future<> resolved when the reader is fully consumed, and closed.
/// Note: the function assumes ownership of the reader and must close it in all cases.

14
readers/forwardable.hh Normal file
View File

@@ -0,0 +1,14 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
class flat_mutation_reader;
flat_mutation_reader make_forwardable(flat_mutation_reader m);

14
readers/forwardable_v2.hh Normal file
View File

@@ -0,0 +1,14 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
class flat_mutation_reader_v2;
flat_mutation_reader_v2 make_forwardable(flat_mutation_reader_v2 m);

33
readers/from_fragments.hh Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright (C) 2022-present ScyllaDB
*
* Modified by ScyllaDB
*/
/*
* SPDX-License-Identifier: (AGPL-3.0-or-later and Apache-2.0)
*/
#pragma once
#include "schema_fwd.hh"
#include <deque>
#include "dht/i_partitioner_fwd.hh"
class flat_mutation_reader;
class reader_permit;
class mutation_fragment;
class ring_position;
namespace query {
class partition_slice;
}
flat_mutation_reader
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment>);
flat_mutation_reader
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment>, const dht::partition_range& pr);
flat_mutation_reader
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment>, const dht::partition_range& pr, const query::partition_slice& slice);

31
readers/from_fragments_v2.hh Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
#include <deque>
#include "dht/i_partitioner_fwd.hh"
class flat_mutation_reader_v2;
class reader_permit;
class mutation_fragment_v2;
class ring_position;
namespace query {
class partition_slice;
}
flat_mutation_reader_v2
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment_v2>);
flat_mutation_reader_v2
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment_v2>, const dht::partition_range& pr);
flat_mutation_reader_v2
make_flat_mutation_reader_from_fragments(schema_ptr, reader_permit, std::deque<mutation_fragment_v2>, const dht::partition_range& pr, const query::partition_slice& slice);

51
readers/from_mutations.hh Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
#include "dht/i_partitioner_fwd.hh"
#include <vector>
#include "mutation_fragment_fwd.hh"
class flat_mutation_reader;
class reader_permit;
class mutation;
namespace query {
class partition_slice;
extern const dht::partition_range full_partition_range;
}
// All mutations should have the same schema.
flat_mutation_reader make_flat_mutation_reader_from_mutations(
schema_ptr schema,
reader_permit permit,
std::vector<mutation>,
const dht::partition_range& pr = query::full_partition_range,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
flat_mutation_reader make_flat_mutation_reader_from_mutations(schema_ptr schema, reader_permit permit, std::vector<mutation> ms, streamed_mutation::forwarding fwd);
// All mutations should have the same schema.
flat_mutation_reader
make_flat_mutation_reader_from_mutations(schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
flat_mutation_reader
make_flat_mutation_reader_from_mutations(schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const dht::partition_range& pr,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);

57
readers/from_mutations_v2.hh Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
#include <vector>
#include "dht/i_partitioner_fwd.hh"
#include "mutation_fragment_fwd.hh"
class flat_mutation_reader_v2;
class reader_permit;
class mutation;
namespace query {
class partition_slice;
extern const dht::partition_range full_partition_range;
}
// All mutations should have the same schema.
flat_mutation_reader_v2 make_flat_mutation_reader_from_mutations_v2(
schema_ptr schema,
reader_permit permit,
std::vector<mutation>,
const dht::partition_range& pr = query::full_partition_range,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
flat_mutation_reader_v2 make_flat_mutation_reader_from_mutations_v2(
schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
streamed_mutation::forwarding fwd);
// All mutations should have the same schema.
flat_mutation_reader_v2
make_flat_mutation_reader_from_mutations_v2(
schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);
// All mutations should have the same schema.
flat_mutation_reader_v2
make_flat_mutation_reader_from_mutations_v2(
schema_ptr schema,
reader_permit permit,
std::vector<mutation> ms,
const dht::partition_range& pr,
const query::partition_slice& slice,
streamed_mutation::forwarding fwd = streamed_mutation::forwarding::no);

25
readers/generating.hh Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
#include <seastar/core/future.hh>
#include <seastar/util/optimized_optional.hh>
#include <functional>
using namespace seastar;
class flat_mutation_reader;
class reader_permit;
class mutation_fragment;
using mutation_fragment_opt = optimized_optional<mutation_fragment>;
flat_mutation_reader
make_generating_reader(schema_ptr s, reader_permit permit, std::function<future<mutation_fragment_opt> ()> get_next_fragment);

21
readers/generating_v2.hh Normal file
View File

@@ -0,0 +1,21 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
#include <functional>
#include <seastar/core/future.hh>
#include "mutation_fragment_fwd.hh"
using namespace seastar;
class flat_mutation_reader_v2;
class reader_permit;
flat_mutation_reader_v2
make_generating_reader(schema_ptr s, reader_permit permit, std::function<future<mutation_fragment_v2_opt> ()> get_next_fragment);

58
readers/multi_range.hh Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include "schema_fwd.hh"
#include "dht/i_partitioner_fwd.hh"
#include <functional>
#include <optional>
#include <seastar/core/io_priority_class.hh>
#include "readers/flat_mutation_reader_fwd.hh"
#include "tracing/trace_state.hh"
using namespace seastar;
class flat_mutation_reader_v2;
class reader_permit;
class mutation_source;
namespace query {
class partition_slice;
}
// Make a reader that enables the wrapped reader to work with multiple ranges.
///
/// \param ranges An range vector that has to contain strictly monotonic
/// partition ranges, such that successively calling
/// `flat_mutation_reader::fast_forward_to()` with each one is valid.
/// An range vector range with 0 or 1 elements is also valid.
/// \param fwd_mr It is only respected when `ranges` contains 0 or 1 partition
/// ranges. Otherwise the reader is created with
/// mutation_reader::forwarding::yes.
flat_mutation_reader_v2
make_flat_multi_range_reader(
schema_ptr s, reader_permit permit, mutation_source source, const dht::partition_range_vector& ranges,
const query::partition_slice& slice, const io_priority_class& pc = default_priority_class(),
tracing::trace_state_ptr trace_state = nullptr,
mutation_reader::forwarding fwd_mr = mutation_reader::forwarding::yes);
/// Make a reader that enables the wrapped reader to work with multiple ranges.
///
/// Generator overload. The ranges returned by the generator have to satisfy the
/// same requirements as the `ranges` param of the vector overload.
flat_mutation_reader_v2
make_flat_multi_range_reader(
schema_ptr s,
reader_permit permit,
mutation_source source,
std::function<std::optional<dht::partition_range>()> generator,
const query::partition_slice& slice,
const io_priority_class& pc = default_priority_class(),
tracing::trace_state_ptr trace_state = nullptr,
mutation_reader::forwarding fwd_mr = mutation_reader::forwarding::yes);

403
readers/mutation_reader.cc Normal file
View File

@@ -0,0 +1,403 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#include "readers/flat_mutation_reader.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "mutation_rebuilder.hh"
#include "mutation_fragment_stream_validator.hh"
#include "schema_upgrader.hh"
logging::logger fmr_logger("flat_mutation_reader");
flat_mutation_reader& flat_mutation_reader::operator=(flat_mutation_reader&& o) noexcept {
if (_impl && _impl->is_close_required()) {
impl* ip = _impl.get();
// Abort to enforce calling close() before readers are closed
// to prevent leaks and potential use-after-free due to background
// tasks left behind.
on_internal_error_noexcept(fmr_logger, format("{} [{}]: permit {}: was not closed before overwritten by move-assign", typeid(*ip).name(), fmt::ptr(ip), ip->_permit.description()));
abort();
}
_impl = std::move(o._impl);
return *this;
}
flat_mutation_reader::~flat_mutation_reader() {
if (_impl && _impl->is_close_required()) {
impl* ip = _impl.get();
// Abort to enforce calling close() before readers are closed
// to prevent leaks and potential use-after-free due to background
// tasks left behind.
on_internal_error_noexcept(fmr_logger, format("{} [{}]: permit {}: was not closed before destruction", typeid(*ip).name(), fmt::ptr(ip), ip->_permit.description()));
abort();
}
}
static size_t compute_buffer_size(const schema& s, const flat_mutation_reader::tracked_buffer& buffer)
{
return boost::accumulate(
buffer
| boost::adaptors::transformed([&s] (const mutation_fragment& mf) {
return mf.memory_usage();
}), size_t(0)
);
}
void flat_mutation_reader::impl::forward_buffer_to(const position_in_partition& pos) {
_buffer.erase(std::remove_if(_buffer.begin(), _buffer.end(), [this, &pos] (mutation_fragment& f) {
return !f.relevant_for_range_assuming_after(*_schema, pos);
}), _buffer.end());
_buffer_size = compute_buffer_size(*_schema, _buffer);
}
void flat_mutation_reader::impl::clear_buffer_to_next_partition() {
auto next_partition_start = std::find_if(_buffer.begin(), _buffer.end(), [] (const mutation_fragment& mf) {
return mf.is_partition_start();
});
_buffer.erase(_buffer.begin(), next_partition_start);
_buffer_size = compute_buffer_size(*_schema, _buffer);
}
template<typename Source>
future<bool> flat_mutation_reader::impl::fill_buffer_from(Source& source) {
if (source.is_buffer_empty()) {
if (source.is_end_of_stream()) {
return make_ready_future<bool>(true);
}
return source.fill_buffer().then([this, &source] {
return fill_buffer_from(source);
});
} else {
while (!source.is_buffer_empty() && !is_buffer_full()) {
push_mutation_fragment(source.pop_mutation_fragment());
}
return make_ready_future<bool>(source.is_end_of_stream() && source.is_buffer_empty());
}
}
template future<bool> flat_mutation_reader::impl::fill_buffer_from<flat_mutation_reader>(flat_mutation_reader&);
void flat_mutation_reader::do_upgrade_schema(const schema_ptr& s) {
*this = transform(std::move(*this), schema_upgrader(s));
}
void flat_mutation_reader::on_close_error(std::unique_ptr<impl> i, std::exception_ptr ep) noexcept {
impl* ip = i.get();
on_internal_error_noexcept(fmr_logger,
format("Failed to close {} [{}]: permit {}: {}", typeid(*ip).name(), fmt::ptr(ip), ip->_permit.description(), ep));
}
invalid_mutation_fragment_stream::invalid_mutation_fragment_stream(std::runtime_error e) : std::runtime_error(std::move(e)) {
}
static mutation_fragment_v2::kind to_mutation_fragment_kind_v2(mutation_fragment::kind k) {
switch (k) {
case mutation_fragment::kind::partition_start:
return mutation_fragment_v2::kind::partition_start;
case mutation_fragment::kind::static_row:
return mutation_fragment_v2::kind::static_row;
case mutation_fragment::kind::clustering_row:
return mutation_fragment_v2::kind::clustering_row;
case mutation_fragment::kind::range_tombstone:
return mutation_fragment_v2::kind::range_tombstone_change;
case mutation_fragment::kind::partition_end:
return mutation_fragment_v2::kind::partition_end;
}
}
mutation_fragment_stream_validator::mutation_fragment_stream_validator(const ::schema& s)
: _schema(s)
, _prev_kind(mutation_fragment_v2::kind::partition_end)
, _prev_pos(position_in_partition::end_of_partition_tag_t{})
, _prev_partition_key(dht::minimum_token(), partition_key::make_empty()) {
}
bool mutation_fragment_stream_validator::operator()(const dht::decorated_key& dk) {
if (_prev_partition_key.less_compare(_schema, dk)) {
_prev_partition_key = dk;
return true;
}
return false;
}
bool mutation_fragment_stream_validator::operator()(dht::token t) {
if (_prev_partition_key.token() <= t) {
_prev_partition_key._token = t;
return true;
}
return false;
}
bool mutation_fragment_stream_validator::operator()(mutation_fragment_v2::kind kind, position_in_partition_view pos) {
if (_prev_kind == mutation_fragment_v2::kind::partition_end) {
const bool valid = (kind == mutation_fragment_v2::kind::partition_start);
if (valid) {
_prev_kind = mutation_fragment_v2::kind::partition_start;
_prev_pos = pos;
}
return valid;
}
auto cmp = position_in_partition::tri_compare(_schema);
auto res = cmp(_prev_pos, pos);
bool valid = true;
if (_prev_kind == mutation_fragment_v2::kind::range_tombstone_change) {
valid = res <= 0;
} else {
valid = res < 0;
}
if (valid) {
_prev_kind = kind;
_prev_pos = pos;
}
return valid;
}
bool mutation_fragment_stream_validator::operator()(mutation_fragment::kind kind, position_in_partition_view pos) {
return (*this)(to_mutation_fragment_kind_v2(kind), pos);
}
bool mutation_fragment_stream_validator::operator()(const mutation_fragment_v2& mf) {
return (*this)(mf.mutation_fragment_kind(), mf.position());
}
bool mutation_fragment_stream_validator::operator()(const mutation_fragment& mf) {
return (*this)(to_mutation_fragment_kind_v2(mf.mutation_fragment_kind()), mf.position());
}
bool mutation_fragment_stream_validator::operator()(mutation_fragment_v2::kind kind) {
bool valid = true;
switch (_prev_kind) {
case mutation_fragment_v2::kind::partition_start:
valid = kind != mutation_fragment_v2::kind::partition_start;
break;
case mutation_fragment_v2::kind::static_row: // fall-through
case mutation_fragment_v2::kind::clustering_row: // fall-through
case mutation_fragment_v2::kind::range_tombstone_change:
valid = kind != mutation_fragment_v2::kind::partition_start &&
kind != mutation_fragment_v2::kind::static_row;
break;
case mutation_fragment_v2::kind::partition_end:
valid = kind == mutation_fragment_v2::kind::partition_start;
break;
}
if (valid) {
_prev_kind = kind;
}
return valid;
}
bool mutation_fragment_stream_validator::operator()(mutation_fragment::kind kind) {
return (*this)(to_mutation_fragment_kind_v2(kind));
}
bool mutation_fragment_stream_validator::on_end_of_stream() {
return _prev_kind == mutation_fragment_v2::kind::partition_end;
}
void mutation_fragment_stream_validator::reset(dht::decorated_key dk) {
_prev_partition_key = dk;
_prev_pos = position_in_partition::for_partition_start();
_prev_kind = mutation_fragment_v2::kind::partition_start;
}
void mutation_fragment_stream_validator::reset(const mutation_fragment_v2& mf) {
_prev_pos = mf.position();
_prev_kind = mf.mutation_fragment_kind();
}
void mutation_fragment_stream_validator::reset(const mutation_fragment& mf) {
_prev_pos = mf.position();
_prev_kind = to_mutation_fragment_kind_v2(mf.mutation_fragment_kind());
}
namespace {
[[noreturn]] void on_validation_error(seastar::logger& l, const seastar::sstring& reason) {
try {
on_internal_error(l, reason);
} catch (std::runtime_error& e) {
throw invalid_mutation_fragment_stream(e);
}
}
}
bool mutation_fragment_stream_validating_filter::operator()(const dht::decorated_key& dk) {
if (_validation_level < mutation_fragment_stream_validation_level::token) {
return true;
}
if (_validation_level == mutation_fragment_stream_validation_level::token) {
if (_validator(dk.token())) {
return true;
}
on_validation_error(fmr_logger, format("[validator {} for {}] Unexpected token: previous {}, current {}",
static_cast<void*>(this), _name, _validator.previous_token(), dk.token()));
} else {
if (_validator(dk)) {
return true;
}
on_validation_error(fmr_logger, format("[validator {} for {}] Unexpected partition key: previous {}, current {}",
static_cast<void*>(this), _name, _validator.previous_partition_key(), dk));
}
}
mutation_fragment_stream_validating_filter::mutation_fragment_stream_validating_filter(sstring_view name, const schema& s,
mutation_fragment_stream_validation_level level)
: _validator(s)
, _name(format("{} ({}.{} {})", name, s.ks_name(), s.cf_name(), s.id()))
, _validation_level(level)
{
if (fmr_logger.level() <= log_level::debug) {
std::string_view what;
switch (_validation_level) {
case mutation_fragment_stream_validation_level::partition_region:
what = "partition region";
break;
case mutation_fragment_stream_validation_level::token:
what = "partition region and token";
break;
case mutation_fragment_stream_validation_level::partition_key:
what = "partition region and partition key";
break;
case mutation_fragment_stream_validation_level::clustering_key:
what = "partition region, partition key and clustering key";
break;
}
fmr_logger.debug("[validator {} for {}] Will validate {} monotonicity.", static_cast<void*>(this), _name, what);
}
}
bool mutation_fragment_stream_validating_filter::operator()(mutation_fragment_v2::kind kind, position_in_partition_view pos) {
bool valid = false;
fmr_logger.debug("[validator {}] {}:{}", static_cast<void*>(this), kind, pos);
if (_validation_level >= mutation_fragment_stream_validation_level::clustering_key) {
valid = _validator(kind, pos);
} else {
valid = _validator(kind);
}
if (__builtin_expect(!valid, false)) {
if (_validation_level >= mutation_fragment_stream_validation_level::clustering_key) {
on_validation_error(fmr_logger, format("[validator {} for {}] Unexpected mutation fragment: partition key {}: previous {}:{}, current {}:{}",
static_cast<void*>(this), _name, _validator.previous_partition_key(), _validator.previous_mutation_fragment_kind(), _validator.previous_position(), kind, pos));
} else if (_validation_level >= mutation_fragment_stream_validation_level::partition_key) {
on_validation_error(fmr_logger, format("[validator {} for {}] Unexpected mutation fragment: partition key {}: previous {}, current {}",
static_cast<void*>(this), _name, _validator.previous_partition_key(), _validator.previous_mutation_fragment_kind(), kind));
} else {
on_validation_error(fmr_logger, format("[validator {} for {}] Unexpected mutation fragment: previous {}, current {}",
static_cast<void*>(this), _name, _validator.previous_mutation_fragment_kind(), kind));
}
}
return true;
}
bool mutation_fragment_stream_validating_filter::operator()(mutation_fragment::kind kind, position_in_partition_view pos) {
return (*this)(to_mutation_fragment_kind_v2(kind), pos);
}
bool mutation_fragment_stream_validating_filter::operator()(const mutation_fragment_v2& mv) {
return (*this)(mv.mutation_fragment_kind(), mv.position());
}
bool mutation_fragment_stream_validating_filter::operator()(const mutation_fragment& mv) {
return (*this)(to_mutation_fragment_kind_v2(mv.mutation_fragment_kind()), mv.position());
}
bool mutation_fragment_stream_validating_filter::on_end_of_partition() {
return (*this)(mutation_fragment::kind::partition_end, position_in_partition_view(position_in_partition_view::end_of_partition_tag_t()));
}
void mutation_fragment_stream_validating_filter::on_end_of_stream() {
fmr_logger.debug("[validator {}] EOS", static_cast<const void*>(this));
if (!_validator.on_end_of_stream()) {
on_validation_error(fmr_logger, format("[validator {} for {}] Stream ended with unclosed partition: {}", static_cast<const void*>(this), _name,
_validator.previous_mutation_fragment_kind()));
}
}
static size_t compute_buffer_size(const schema& s, const flat_mutation_reader_v2::tracked_buffer& buffer)
{
return boost::accumulate(
buffer
| boost::adaptors::transformed([&s] (const mutation_fragment_v2& mf) {
return mf.memory_usage();
}), size_t(0)
);
}
flat_mutation_reader_v2& flat_mutation_reader_v2::operator=(flat_mutation_reader_v2&& o) noexcept {
if (_impl && _impl->is_close_required()) {
impl* ip = _impl.get();
// Abort to enforce calling close() before readers are closed
// to prevent leaks and potential use-after-free due to background
// tasks left behind.
on_internal_error_noexcept(fmr_logger, format("{} [{}]: permit {}: was not closed before overwritten by move-assign", typeid(*ip).name(), fmt::ptr(ip), ip->_permit.description()));
abort();
}
_impl = std::move(o._impl);
return *this;
}
flat_mutation_reader_v2::~flat_mutation_reader_v2() {
if (_impl && _impl->is_close_required()) {
impl* ip = _impl.get();
// Abort to enforce calling close() before readers are closed
// to prevent leaks and potential use-after-free due to background
// tasks left behind.
on_internal_error_noexcept(fmr_logger, format("{} [{}]: permit {}: was not closed before destruction", typeid(*ip).name(), fmt::ptr(ip), ip->_permit.description()));
abort();
}
}
void flat_mutation_reader_v2::impl::forward_buffer_to(const position_in_partition& pos) {
clear_buffer();
_buffer_size = compute_buffer_size(*_schema, _buffer);
}
void flat_mutation_reader_v2::impl::clear_buffer_to_next_partition() {
auto next_partition_start = std::find_if(_buffer.begin(), _buffer.end(), [] (const mutation_fragment_v2& mf) {
return mf.is_partition_start();
});
_buffer.erase(_buffer.begin(), next_partition_start);
_buffer_size = compute_buffer_size(*_schema, _buffer);
}
template<typename Source>
future<bool> flat_mutation_reader_v2::impl::fill_buffer_from(Source& source) {
if (source.is_buffer_empty()) {
if (source.is_end_of_stream()) {
return make_ready_future<bool>(true);
}
return source.fill_buffer().then([this, &source] {
return fill_buffer_from(source);
});
} else {
while (!source.is_buffer_empty() && !is_buffer_full()) {
push_mutation_fragment(source.pop_mutation_fragment());
}
return make_ready_future<bool>(source.is_end_of_stream() && source.is_buffer_empty());
}
}
template future<bool> flat_mutation_reader_v2::impl::fill_buffer_from<flat_mutation_reader_v2>(flat_mutation_reader_v2&);
void flat_mutation_reader_v2::do_upgrade_schema(const schema_ptr& s) {
*this = transform(std::move(*this), schema_upgrader_v2(s));
}
void flat_mutation_reader_v2::on_close_error(std::unique_ptr<impl> i, std::exception_ptr ep) noexcept {
impl* ip = i.get();
on_internal_error_noexcept(fmr_logger,
format("Failed to close {} [{}]: permit {}: {}", typeid(*ip).name(), fmt::ptr(ip), ip->_permit.description(), ep));
}
future<mutation_opt> read_mutation_from_flat_mutation_reader(flat_mutation_reader_v2& r) {
return r.consume(mutation_rebuilder_v2(r.schema()));
}

2100
flat_mutation_reader.cc → readers/mutation_readers.cc
View File

File diff suppressed because it is too large Load Diff

14
readers/nonforwardable.hh Normal file
View File

@@ -0,0 +1,14 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
class flat_mutation_reader;
flat_mutation_reader make_nonforwardable(flat_mutation_reader, bool);

49
readers/reversing.hh Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include <memory>
class flat_mutation_reader;
namespace query {
struct max_result_size;
class partition_slice;
}
/// A reader that emits partitions in native reverse order.
///
/// 1. The reader's schema() method will return a reversed schema (see
/// \ref schema::make_reversed()).
/// 2. Static row is still emitted first.
/// 3. Range tombstones' bounds are reversed (see \ref range_tombstone::reverse()).
/// 4. Clustered rows and range tombstones are emitted in descending order.
/// Because of 3 and 4 the guarantee that a range tombstone is emitted before
/// any mutation fragment affected by it still holds.
/// Ordering of partitions themselves remains unchanged.
/// For more details see docs/design-notes/reverse-reads.md.
///
/// The reader's schema (returned by `schema()`) is the reverse of `original`'s schema.
///
/// \param original the reader to be reversed.
/// \param max_size the maximum amount of memory the reader is allowed to use
/// for reversing and conversely the maximum size of the results. The
/// reverse reader reads entire partitions into memory, before reversing
/// them. Since partitions can be larger than the available memory, we need
/// to enforce a limit on memory consumption. When reaching the soft limit
/// a warning will be logged. When reaching the hard limit the read will be
/// aborted.
/// \param slice serves as a convenience slice storage for reads that have to
/// store an edited slice somewhere. This is common for reads that work
/// with a native-reversed slice and so have to convert the one used in the
/// query -- which is in half-reversed format.
///
/// FIXME: reversing should be done in the sstable layer, see #1413.
flat_mutation_reader
make_reversing_reader(flat_mutation_reader original, query::max_result_size max_size, std::unique_ptr<query::partition_slice> slice = {});

20
readers/slice_mutations.hh Normal file
View File

@@ -0,0 +1,20 @@
/*
* Copyright (C) 2022-present ScyllaDB
*/
/*
* SPDX-License-Identifier: AGPL-3.0-or-later
*/
#pragma once
#include <vector>
#include "schema_fwd.hh"
class mutation;
namespace query {
class partition_slice;
}
std::vector<mutation> slice_mutations(schema_ptr schema, std::vector<mutation> ms, const query::partition_slice& slice);

1
repair/row_level.cc
View File

@@ -46,6 +46,7 @@
#include "db/batchlog_manager.hh"
#include "cql3/untyped_result_set.hh"
#include "idl/partition_checksum.dist.hh"
#include "readers/empty.hh"
extern logging::logger rlogger;

1
replica/database.cc
View File

@@ -59,6 +59,7 @@
#include "tombstone_gc.hh"
#include "data_dictionary/impl.hh"
#include "readers/multi_range.hh"
using namespace std::chrono_literals;
using namespace db;

2
replica/memtable.cc
View File

@@ -12,6 +12,8 @@
#include "partition_snapshot_reader.hh"
#include "partition_builder.hh"
#include "mutation_partition_view.hh"
#include "readers/empty_v2.hh"
#include "readers/forwardable_v2.hh"
namespace replica {

4
replica/table.cc
View File

@@ -43,6 +43,10 @@
#include <boost/range/algorithm/remove_if.hpp>
#include <boost/range/algorithm.hpp>
#include "utils/error_injection.hh"
#include "readers/reversing.hh"
#include "readers/from_mutations.hh"
#include "readers/empty_v2.hh"
#include "readers/multi_range.hh"
namespace replica {

3
row_cache.cc
View File

@@ -20,6 +20,9 @@
#include "dirty_memory_manager.hh"
#include "cache_flat_mutation_reader.hh"
#include "real_dirty_memory_accounter.hh"
#include "readers/empty.hh"
#include "readers/forwardable.hh"
#include "readers/nonforwardable.hh"
namespace cache {

3
sstables/kl/reader.hh
View File

@@ -8,8 +8,9 @@
#pragma once
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "sstables/progress_monitor.hh"
#include <seastar/core/io_priority_class.hh>
namespace sstables {
namespace kl {

4
sstables/mx/reader.hh
View File

@@ -8,8 +8,10 @@
#pragma once
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader.hh"
#include "sstables/progress_monitor.hh"
#include <seastar/core/io_priority_class.hh>
namespace sstables {
namespace mx {

2
sstables/sstable_mutation_reader.hh
View File

@@ -22,7 +22,7 @@
#include "clustering_ranges_walker.hh"
#include "binary_search.hh"
#include "../dht/i_partitioner.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "sstables/mx/partition_reversing_data_source.hh"
namespace sstables {

2
sstables/sstable_set.cc
View File

@@ -21,6 +21,8 @@
#include "sstable_set_impl.hh"
#include "replica/database.hh"
#include "readers/from_mutations.hh"
#include "readers/empty_v2.hh"
namespace sstables {

5
sstables/sstable_set.hh
View File

@@ -8,12 +8,13 @@
#pragma once
#include "flat_mutation_reader.hh"
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "sstables/progress_monitor.hh"
#include "shared_sstable.hh"
#include "dht/i_partitioner.hh"
#include <seastar/core/shared_ptr.hh>
#include <seastar/core/io_priority_class.hh>
#include <vector>
namespace utils {

2
sstables/sstables.cc
View File

@@ -73,6 +73,8 @@
#include "utils/bit_cast.hh"
#include "utils/cached_file.hh"
#include "tombstone_gc.hh"
#include "readers/reversing.hh"
#include "readers/forwardable.hh"
thread_local disk_error_signal_type sstable_read_error;
thread_local disk_error_signal_type sstable_write_error;

1
streaming/stream_session.cc
View File

@@ -36,6 +36,7 @@
#include "mutation_source_metadata.hh"
#include "streaming/stream_mutation_fragments_cmd.hh"
#include "consumer.hh"
#include "readers/generating.hh"
namespace streaming {

2
streaming/stream_transfer_task.cc
View File

@@ -16,7 +16,7 @@
#include "streaming/stream_reason.hh"
#include "streaming/stream_mutation_fragments_cmd.hh"
#include "mutation_reader.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "mutation_fragment_stream_validator.hh"
#include "frozen_mutation.hh"
#include "mutation.hh"

13
test/boost/flat_mutation_reader_test.cc
View File

@@ -15,7 +15,13 @@
#include "mutation.hh"
#include "mutation_fragment.hh"
#include "test/lib/mutation_source_test.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/reversing.hh"
#include "readers/forwardable.hh"
#include "readers/delegating.hh"
#include "readers/multi_range.hh"
#include "readers/from_mutations.hh"
#include "readers/from_fragments.hh"
#include "mutation_reader.hh"
#include "schema_builder.hh"
#include "replica/memtable.hh"
@@ -32,6 +38,9 @@
#include "test/lib/random_schema.hh"
#include <boost/range/adaptor/map.hpp>
#include "readers/from_mutations_v2.hh"
#include "readers/from_fragments_v2.hh"
#include "readers/forwardable_v2.hh"
struct mock_consumer {
struct result {
@@ -560,7 +569,7 @@ void test_flat_stream(schema_ptr s, std::vector<mutation> muts, reversed_partiti
return fmr.consume_in_thread(std::move(fsc));
} else {
if (reversed) {
return with_closeable(make_reversing_reader(make_flat_mutation_reader<delegating_reader>(fmr), query::max_result_size(size_t(1) << 20)),
return with_closeable(make_reversing_reader(make_delegating_reader(fmr), query::max_result_size(size_t(1) << 20)),
[fsc = std::move(fsc)] (flat_mutation_reader& reverse_reader) mutable {
return reverse_reader.consume(std::move(fsc));
}).get0();

1
test/boost/frozen_mutation_test.cc
View File

@@ -20,6 +20,7 @@
#include "test/lib/mutation_source_test.hh"
#include <seastar/core/thread.hh>
#include "readers/from_mutations.hh"
static schema_builder new_table() {
return { "some_keyspace", "some_table" };

2
test/boost/memtable_test.cc
View File

@@ -24,7 +24,7 @@
#include "test/lib/mutation_source_test.hh"
#include "test/lib/mutation_assertions.hh"
#include "test/lib/flat_mutation_reader_assertions.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "test/lib/data_model.hh"
#include "test/lib/eventually.hh"
#include "test/lib/random_utils.hh"

1
test/boost/multishard_combining_reader_as_mutation_source_test.cc
View File

@@ -27,6 +27,7 @@
#include "mutation_reader.hh"
#include "schema_registry.hh"
#include "service/priority_manager.hh"
#include "readers/forwardable_v2.hh"
// It has to be a container that does not invalidate pointers
static std::list<dummy_sharder> keep_alive_sharder;

1
test/boost/mutation_fragment_test.cc
View File

@@ -26,6 +26,7 @@
#include "test/lib/simple_schema.hh"
#include <boost/range/algorithm/transform.hpp>
#include "readers/from_mutations.hh"
// A StreamedMutationConsumer which distributes fragments randomly into several mutations.
class fragment_scatterer {

1
test/boost/mutation_query_test.cc
View File

@@ -27,6 +27,7 @@
#include <seastar/core/thread.hh>
#include "schema_builder.hh"
#include "partition_slice_builder.hh"
#include "readers/from_mutations.hh"
using namespace std::literals::chrono_literals;

6
test/boost/mutation_reader_test.cc
View File

@@ -47,6 +47,12 @@
#include "mutation_rebuilder.hh"
#include <boost/range/algorithm/sort.hpp>
#include "readers/from_mutations.hh"
#include "readers/forwardable_v2.hh"
#include "readers/forwardable.hh"
#include "readers/from_fragments_v2.hh"
#include "readers/empty.hh"
#include "readers/empty_v2.hh"
static schema_ptr make_schema() {
return schema_builder("ks", "cf")

3
test/boost/mutation_test.cc
View File

@@ -53,6 +53,9 @@
#include "types/user.hh"
#include "concrete_types.hh"
#include "mutation_rebuilder.hh"
#include "readers/from_mutations_v2.hh"
#include "readers/from_mutations.hh"
#include "readers/from_fragments_v2.hh"
using namespace std::chrono_literals;

6
test/boost/mutation_writer_test.cc
View File

@@ -16,7 +16,8 @@
#include "mutation_fragment.hh"
#include "mutation_rebuilder.hh"
#include "test/lib/mutation_source_test.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/from_mutations.hh"
#include "mutation_writer/multishard_writer.hh"
#include "mutation_writer/timestamp_based_splitting_writer.hh"
#include "mutation_writer/partition_based_splitting_writer.hh"
@@ -28,6 +29,9 @@
#include "test/lib/log.hh"
#include <boost/range/adaptor/map.hpp>
#include "readers/from_mutations_v2.hh"
#include "readers/empty_v2.hh"
#include "readers/generating_v2.hh"
using namespace mutation_writer;

2
test/boost/querier_cache_test.cc
View File

@@ -19,6 +19,8 @@
#include <seastar/util/closeable.hh>
#include <boost/range/algorithm/sort.hpp>
#include "readers/from_mutations.hh"
#include "readers/empty_v2.hh"
using namespace std::chrono_literals;

1
test/boost/reader_concurrency_semaphore_test.cc
View File

@@ -15,6 +15,7 @@
#include <seastar/testing/test_case.hh>
#include <seastar/testing/thread_test_case.hh>
#include <boost/test/unit_test.hpp>
#include "readers/empty_v2.hh"
SEASTAR_THREAD_TEST_CASE(test_reader_concurrency_semaphore_clear_inactive_reads) {
simple_schema s;

7
test/boost/row_cache_test.cc
View File

@@ -33,6 +33,11 @@
#include "test/lib/random_utils.hh"
#include <boost/range/algorithm/min_element.hpp>
#include "readers/from_mutations.hh"
#include "readers/from_mutations_v2.hh"
#include "readers/delegating.hh"
#include "readers/delegating_v2.hh"
#include "readers/empty_v2.hh"
using namespace std::chrono_literals;
@@ -1308,7 +1313,6 @@ public:
});
}
};
static std::vector<mutation> updated_ring(std::vector<mutation>& mutations) {
std::vector<mutation> result;
for (auto&& m : mutations) {
@@ -1583,7 +1587,6 @@ SEASTAR_TEST_CASE(test_cache_population_and_clear_race) {
});
}
SEASTAR_TEST_CASE(test_mvcc) {
return seastar::async([] {
auto test = [&] (const mutation& m1, const mutation& m2, bool with_active_memtable_reader) {

2
test/boost/sstable_compaction_test.cc
View File

@@ -67,6 +67,8 @@
#include "test/lib/reader_concurrency_semaphore.hh"
#include "test/lib/sstable_utils.hh"
#include "test/lib/random_utils.hh"
#include "readers/from_mutations_v2.hh"
#include "readers/from_fragments_v2.hh"
namespace fs = std::filesystem;

2
test/boost/sstable_datafile_test.cc
View File

@@ -59,6 +59,8 @@
#include "test/lib/reader_concurrency_semaphore.hh"
#include "test/lib/sstable_utils.hh"
#include "test/lib/random_utils.hh"
#include "readers/from_mutations_v2.hh"
#include "readers/from_fragments_v2.hh"
namespace fs = std::filesystem;

1
test/boost/sstable_set_test.cc
View File

@@ -13,6 +13,7 @@
#include "sstables/sstable_set.hh"
#include "sstables/sstables.hh"
#include "test/lib/simple_schema.hh"
#include "readers/from_mutations_v2.hh"
static sstables::sstable_set make_sstable_set(schema_ptr schema, lw_shared_ptr<sstable_list> all = {}, bool use_level_metadata = true) {
return sstables::sstable_set(std::make_unique<partitioned_sstable_set>(schema, std::move(all), use_level_metadata), schema);

2
test/boost/view_build_test.cc
View File

@@ -31,6 +31,8 @@
#include "test/lib/random_utils.hh"
#include "utils/ranges.hh"
#include "readers/from_mutations_v2.hh"
using namespace std::literals::chrono_literals;
schema_ptr test_table_schema() {

3
test/lib/flat_mutation_reader_assertions.hh
View File

@@ -10,7 +10,8 @@
#include <boost/test/unit_test.hpp>
#include <seastar/util/backtrace.hh>
#include "flat_mutation_reader_v2.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/flat_mutation_reader_v2.hh"
#include "mutation_assertions.hh"
#include "schema.hh"
#include "test/lib/log.hh"

2
test/lib/mutation_source_test.cc
View File

@@ -14,7 +14,7 @@
#include "counters.hh"
#include "mutation_rebuilder.hh"
#include "test/lib/simple_schema.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "test/lib/flat_mutation_reader_assertions.hh"
#include "mutation_query.hh"
#include "mutation_rebuilder.hh"

2
test/lib/normalizing_reader.hh
View File

@@ -10,7 +10,7 @@
#include "mutation_reader.hh"
#include <seastar/core/future.hh>
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
/*
* A helper class that wraps another flat_mutation_reader

1
test/manual/enormous_table_scan_test.cc
View File

@@ -28,6 +28,7 @@
#include "range.hh"
#include "sstables/sstables.hh"
#include "schema_builder.hh"
#include "readers/forwardable.hh"
class enormous_table_reader final : public flat_mutation_reader::impl {
// Reader for a table with 4.5 billion rows, all with partition key 0 and an incrementing clustering key

4
test/perf/perf_mutation_readers.cc
View File

@@ -17,7 +17,9 @@
#include "test/perf/perf.hh"
#include "mutation_reader.hh"
#include "flat_mutation_reader.hh"
#include "readers/flat_mutation_reader.hh"
#include "readers/from_mutations.hh"
#include "readers/empty_v2.hh"
#include "replica/memtable.hh"
namespace tests {