lxy-9602 commented on code in PR #286:
URL: https://github.com/apache/paimon-cpp/pull/286#discussion_r3966774747
##########
include/paimon/reader/batch_reader.h:
##########
@@ -81,6 +81,23 @@ class PAIMON_EXPORT BatchReader {
/// bitmap. Noted that the returned bitmap has at least one valid row id.
virtual Result<ReadBatchWithBitmap> NextBatchWithBitmap();
+ /// Starts whatever background work this reader would otherwise start on
its first read, so a
+ /// caller that knows this reader is next can pay that startup while still
consuming the
+ /// previous one.
+ ///
+ /// This is an optional hint and never changes what the reader returns:
ordering, filtering and
+ /// metrics are the same with or without it. A reader with nothing to
start, or one that is not
+ /// yet ready to start it, does nothing. Calling it before the reader is
configured (for example
+ /// before `SetReadSchema()` on a file reader), or after `Close()`, is
such a no-op.
+ ///
+ /// It reports no error on purpose: the caller may warm up a reader it
never ends up reading,
+ /// and a hint about a file nobody reads must not fail the read in
progress. An implementation
+ /// that cannot start its work leaves it to be started by the first read,
which reports the
+ /// failure itself.
+ /// @warning The call starts background work owned by this reader, so it
must be made from the
+ /// same thread that reads this reader, and not concurrently with any
other call on it.
+ virtual void Warmup() {}
Review Comment:
Consider move to file batch reader?
##########
include/paimon/read_context.h:
##########
@@ -351,6 +357,16 @@ class PAIMON_EXPORT ReadContextBuilder {
/// @return Reference to this builder for method chaining.
ReadContextBuilder& WithCacheConfig(const CacheConfig& config);
+ /// Set how far the reader prepares the next file before it is read.
+ ///
+ /// Warmup overlaps remote-storage latency with the read of the current
file. Higher levels hide
+ /// more latency but use more memory, and may warm files that a query
never reads (for example
+ /// when a LIMIT stops the scan early).
+ /// @param level The warmup level to use (default: WarmupLevel::DECODED).
+ /// @return Reference to this builder for method chaining.
+ /// @see WarmupLevel
+ ReadContextBuilder& SetWarmupLevel(WarmupLevel level);
+
Review Comment:
`Warmup` and cache seem to be tightly coupled in the current implementation.
`SetWarmupLevel(WarmupLevel::RAW)` with `SetReadAheadCacheEnabled(false)`
appears to be effectively equivalent to `SetWarmupLevel(WarmupLevel::None)`.
Please document this limitation in the comments.
##########
include/paimon/reader/file_batch_reader.h:
##########
@@ -63,6 +63,23 @@ class PAIMON_EXPORT FileBatchReader : public BatchReader {
/// Get whether or not support read precisely while bitmap pushed down.
virtual bool SupportPreciseBitmapSelection() const = 0;
+
+ /// Starts whatever background work this reader would otherwise start on
its first read, so a
+ /// caller that knows this reader is next can pay that startup while still
consuming the
+ /// previous one.
+ ///
+ /// This is an optional hint and never changes what the reader returns:
ordering, filtering and
+ /// metrics are the same with or without it. A reader with nothing to
start, or one that is not
+ /// yet ready to start it, does nothing. Calling it before the reader is
configured (for example
+ /// before `SetReadSchema()`), or after `Close()`, is such a no-op.
+ ///
+ /// It reports no error on purpose: the caller may warm up a reader it
never ends up reading,
+ /// and a hint about a file nobody reads must not fail the read in
progress. An implementation
+ /// that cannot start its work leaves it to be started by the first read,
which reports the
+ /// failure itself.
+ /// @warning The call starts background work owned by this reader, so it
must be made from the
+ /// same thread that reads this reader, and not concurrently with any
other call on it.
Review Comment:
Is this something that can be customized by plugins? If not, should we
clarify that in the comment? The behavior described by “The call starts
background work owned by this reader” seems to be fixed by the implementation.
##########
src/paimon/common/reader/prefetch_file_batch_reader_impl.h:
##########
@@ -123,10 +125,19 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
uint32_t prefetch_queue_capacity, bool
enable_adaptive_prefetch_strategy,
const std::shared_ptr<Executor>& executor, const
std::shared_ptr<ReadAheadCache>& cache,
const std::shared_ptr<PrefetchIoMetricsState>& io_metrics,
- const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool, WarmupLevel
warmup_level);
Status CleanUp();
void Workloop();
+ /// Starts the background prefetch thread if it is not running yet. The
first read does this
+ /// itself; Warmup() is the same call made earlier, so the two must not
diverge.
+ void EnsureBackgroundThread();
+ /// Initializes and warms the read-ahead cache at most once per read-range
generation, using the
+ /// first reader's PreBufferRange(). Shared by Workloop() (DECODED) and
Warmup() (RAW), so the
+ /// two never Init the cache twice. A no-op when there is no cache. Errors
are recorded via
+ /// SetReadStatus() rather than returned: a warmup hint for a file that
may never be read must
+ /// not fail an in-flight read.
+ void WarmCacheOnce();
Review Comment:
`exchange(true)` only elects one initializer; it does not make other threads
wait for `cache_->Init()` to finish. If RAW warmup and `Workloop()` race, the
losing thread may continue with an uninitialized cache. Please either rely on
the documented non-concurrent contract and remove this race claim, or
synchronize initialization completion.
##########
src/paimon/common/reader/data_file_reader_factory.cpp:
##########
@@ -77,15 +77,16 @@ Result<std::unique_ptr<FileBatchReader>>
DataFileReaderFactory::Open(
const std::shared_ptr<FileSystem>& file_system, const
std::shared_ptr<Executor>& executor,
const std::shared_ptr<MemoryPool>& pool, const
std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
if (read_options.prefetch_enabled &&
FormatSupportsPrefetch(format_identifier)) {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<PrefetchFileBatchReaderImpl>
prefetch_reader,
- PrefetchFileBatchReaderImpl::Create(
- file_path, file_size, reader_builder,
file_system,
- read_options.prefetch_max_parallel_num,
- read_options.read_batch_size,
read_options.prefetch_batch_count,
- read_options.adaptive_prefetch_strategy,
executor,
- /*initialize_read_ranges=*/false,
- read_options.read_ahead_cache_enabled,
read_options.cache_config,
- read_options.prefetch_io_metrics_enabled,
pool, arrow_pool));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<PrefetchFileBatchReaderImpl> prefetch_reader,
+ PrefetchFileBatchReaderImpl::Create(
+ file_path, file_size, reader_builder, file_system,
+ read_options.prefetch_max_parallel_num,
read_options.read_batch_size,
+ read_options.prefetch_batch_count,
read_options.adaptive_prefetch_strategy,
+ executor,
+ /*initialize_read_ranges=*/false,
read_options.read_ahead_cache_enabled,
+ read_options.cache_config,
read_options.prefetch_io_metrics_enabled, pool,
+ arrow_pool, read_options.warmup_level));
Review Comment:
Please don’t put the new parameter at the end. Keeping `pool` as the last
parameter is more consistent.
##########
src/paimon/core/io/key_value_record_reader.h:
##########
@@ -39,6 +39,13 @@ class KeyValueRecordReader {
virtual std::shared_ptr<Metrics> GetReaderMetrics() const = 0;
+ /// Starts whatever background work this reader would otherwise start on
its first read, so a
+ /// caller that knows this reader is next can pay that startup while still
consuming the
+ /// previous one. Optional and reporting no error, like
`FileBatchReader::Warmup()`: a reader
+ /// with nothing to start does nothing, and a hint about a file nobody
reads must not fail the
+ /// read in progress.
+ virtual void Warmup() {}
Review Comment:
Is it a good idea to keep an empty `{}` here? Could this cause a newly added
`KeyValueRecordReader` in the future to miss forwarding `Warmup()`?
##########
src/paimon/core/operation/merge_file_split_read.cpp:
##########
@@ -662,14 +662,46 @@ MergeFileSplitRead::CreateRecordReadersForSection(
const std::vector<SortedRun>& section, const BinaryRow& partition,
DeletionVector::Factory dv_factory, const std::shared_ptr<Predicate>&
predicate,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+ // Every run of a section is read by the same merge, so the section's
files are built as ONE
+ // batch instead of run by run, and both section read paths go through
this one build. The
+ // readers are then handed back to the run they belong to by position.
+ std::vector<std::shared_ptr<DataFileMeta>> section_files;
+ for (const SortedRun& run : section) {
+ const std::vector<std::shared_ptr<DataFileMeta>>& run_files =
run.Files();
+ section_files.insert(section_files.end(), run_files.begin(),
run_files.end());
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::unique_ptr<FileBatchReader>> raw_file_readers,
+ CreateRawFileReaders(partition, section_files, read_schema_,
predicate, dv_factory,
+ /*row_ranges=*/{}, data_file_path_factory,
+ /*extra_format_options=*/{}));
+ // A build that dropped a reader would silently shift every following file
into the wrong run.
Review Comment:
I’m not following how the refactoring of `CreateReaderForRun` is related to
the current warmup support. As far as I see, `CreateRawFileReaders` doesn’t
introduce any additional concurrency. The current changes also add more
validation, so it doesn’t seem like they would have much impact on the final
performance?
##########
src/paimon/core/io/complete_row_tracking_fields_reader_test.cpp:
##########
@@ -486,4 +486,21 @@ TEST_F(CompleteRowTrackingFieldsBatchReaderTest,
TestOnlyReadRowTrackingFields)
CheckResult(src_array, /*first_row_id=*/100, /*snapshot_id=*/4,
read_schema, target_array);
}
+// Warmup() is only a hint, but it must reach the wrapped reader: this wrapper
sits between the
+// split read and the file, and swallowing the hint here would leave the whole
stack below it cold
+// no matter which layer asked for the warmup.
+TEST_F(CompleteRowTrackingFieldsBatchReaderTest,
TestWarmupForwardsToInnerReader) {
Review Comment:
It doesn’t seem necessary to add a comment like this to every case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
