This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 5d9f9645 perf(read): support warmup between data files (#286)
5d9f9645 is described below
commit 5d9f9645a659564633a9111e9492021ceaacf046
Author: Yonghao Fang <[email protected]>
AuthorDate: Thu Sep 10 13:59:43 2026 +0800
perf(read): support warmup between data files (#286)
---
CMakeLists.txt | 5 +-
benchmark/CMakeLists.txt | 10 +-
benchmark/parquet_format_benchmark_test.cpp | 21 +-
docs/source/api/read.rst | 2 +
include/paimon/read_context.h | 21 +-
include/paimon/reader/file_batch_reader.h | 20 ++
include/paimon/utils/prefetch_cache_config.h | 22 ++
src/paimon/CMakeLists.txt | 1 +
...map_shared_shredding_read_plan_factory_test.cpp | 14 +
.../common/data/shredding/shredding_file_reader.h | 4 +
.../bitmap/apply_bitmap_index_batch_reader.h | 4 +
.../apply_bitmap_index_batch_reader_test.cpp | 17 +-
.../common/reader/data_file_reader_factory.cpp | 19 +-
.../common/reader/data_file_reader_factory.h | 3 +
.../reader/data_file_reader_factory_test.cpp | 80 ++++++
.../common/reader/delegating_prefetch_reader.h | 9 +
.../late_materializing_file_batch_reader_test.cpp | 6 +-
.../reader/prefetch_file_batch_reader_impl.cpp | 98 +++++--
.../reader/prefetch_file_batch_reader_impl.h | 22 +-
.../prefetch_file_batch_reader_impl_test.cpp | 293 ++++++++++++++++++---
src/paimon/common/utils/arrow/status_utils.h | 12 +-
.../apply_deletion_vector_batch_reader.h | 4 +
.../apply_deletion_vector_batch_reader_test.cpp | 17 +-
.../core/io/complete_row_tracking_fields_reader.h | 4 +
.../complete_row_tracking_fields_reader_test.cpp | 14 +
.../core/io/concat_key_value_record_reader.h | 30 +++
.../io/concat_key_value_record_reader_test.cpp | 142 ++++++++++
src/paimon/core/io/field_mapping_reader.h | 6 +
src/paimon/core/io/field_mapping_reader_test.cpp | 23 ++
.../core/io/key_value_data_file_record_reader.cpp | 9 +
.../core/io/key_value_data_file_record_reader.h | 2 +
.../io/key_value_data_file_record_reader_test.cpp | 15 ++
.../core/io/key_value_in_memory_record_reader.h | 3 +
src/paimon/core/io/key_value_record_reader.h | 11 +
.../core/io/merged_key_value_record_reader.h | 4 +
src/paimon/core/io/vector_file_batch_reader.h | 4 +
.../core/io/vector_file_batch_reader_test.cpp | 12 +
src/paimon/core/mergetree/compact/loser_tree.cpp | 15 ++
src/paimon/core/mergetree/compact/loser_tree.h | 4 +
.../mergetree/compact/sort_merge_reader_test.cpp | 70 +++++
.../core/mergetree/merge_tree_writer_test.cpp | 4 +
src/paimon/core/mergetree/spill_reader.h | 2 +
src/paimon/core/operation/abstract_split_read.cpp | 1 +
src/paimon/core/operation/internal_read_context.h | 4 +
.../core/operation/merge_file_split_read.cpp | 4 +
src/paimon/core/operation/read_context.cpp | 14 +-
src/paimon/core/operation/read_context_test.cpp | 21 ++
src/paimon/core/table/format/format_table_read.cpp | 1 +
.../core/table/system/audit_log_system_table.cpp | 40 +--
.../core/table/system/audit_log_system_table.h | 8 +
.../table/system/read_optimized_system_table.cpp | 11 +-
.../table/system/read_optimized_system_table.h | 8 +
src/paimon/core/table/system/system_table_test.cpp | 44 ++++
src/paimon/testing/mock/mock_file_batch_reader.h | 11 +
54 files changed, 1124 insertions(+), 121 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 2fe2df8e..f7f6adb4 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -569,7 +569,10 @@ install(EXPORT PaimonTargets
NAMESPACE Paimon::
DESTINATION ${PAIMON_CMAKE_INSTALL_DIR})
-if(PAIMON_BUILD_BENCHMARKS)
+# benchmark/ also holds gtest targets that guard the assumptions the
benchmarks are built on, and
+# those have to be built whenever tests are. The subdirectory gates each
target on the option it
+# belongs to, so a tests-only build adds no benchmark executable.
+if(PAIMON_BUILD_TESTS OR PAIMON_BUILD_BENCHMARKS)
add_subdirectory(benchmark)
endif()
diff --git a/benchmark/CMakeLists.txt b/benchmark/CMakeLists.txt
index 0d4dc28c..60ac3fb3 100644
--- a/benchmark/CMakeLists.txt
+++ b/benchmark/CMakeLists.txt
@@ -89,9 +89,9 @@ if(PAIMON_BUILD_TESTS)
paimon_shared
${GTEST_LINK_TOOLCHAIN})
- # Guards the format-layer assumptions parquet_format_benchmark.cpp is
built on. The benchmark
- # itself is only compiled under PAIMON_BUILD_BENCHMARKS, which CI does not
set, so this test is
- # what keeps those assumptions covered.
+ # Guards the format-layer assumptions parquet_format_benchmark.cpp is
built on. Links the same
+ # libraries as the benchmark itself: CoreOptions resolves manifest.format,
whose default is avro,
+ # so a binary that builds CoreOptions needs every enabled format factory
registered.
add_paimon_test(parquet_format_benchmark_test
SOURCES
parquet_format_benchmark_test.cpp
@@ -100,9 +100,7 @@ if(PAIMON_BUILD_TESTS)
STATIC_LINK_LIBS
arrow
parquet
- paimon_shared
- ${PAIMON_LOCAL_FILE_SYSTEM_SHARED_LINK_LIBS}
- ${PAIMON_PARQUET_FILE_FORMAT_STATIC_LINK_LIBS}
+ ${PAIMON_BENCHMARK_STATIC_LINK_LIBS}
test_utils_static
${GTEST_LINK_TOOLCHAIN})
endif()
diff --git a/benchmark/parquet_format_benchmark_test.cpp
b/benchmark/parquet_format_benchmark_test.cpp
index 0bf8f4a7..813c9b70 100644
--- a/benchmark/parquet_format_benchmark_test.cpp
+++ b/benchmark/parquet_format_benchmark_test.cpp
@@ -19,8 +19,8 @@
// Smoke test for the assumptions parquet_format_benchmark.cpp is built on.
//
-// The benchmark is only compiled under PAIMON_BUILD_BENCHMARKS, which CI does
not set, so nothing
-// there runs in CI. Every assumption it makes about the format layer - that a
codec name is
+// The benchmark itself is only compiled under PAIMON_BUILD_BENCHMARKS, which
CI does not set, so no
+// benchmark runs there. Every assumption it makes about the format layer -
that a codec name is
// accepted, that a dictionary-encoded input array can be written, that a
nested or high-precision
// column survives a round trip, that a predicate and a selection bitmap
return the rows the
// benchmark asserts on, that the reader metrics it reports exist - is checked
here instead, at a
@@ -42,6 +42,7 @@
#include "arrow/c/helpers.h"
#include "gtest/gtest.h"
#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/path_util.h"
@@ -62,6 +63,7 @@
#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/utils/roaring_bitmap32.h"
+#include "parquet/types.h"
namespace paimon::parquet {
namespace {
@@ -281,7 +283,7 @@ TEST_F(ParquetFormatBenchmarkTest, RegisteredCodecsWrite) {
ASSERT_TRUE(builder.Finish(&ids).ok());
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch, Wrap(schema,
{ids}));
- for (const std::string& codec :
+ for (const std::string codec :
{"none", "snappy", "gzip", "brotli", "zstd", "lz4_raw",
"lz4_hadoop"}) {
const std::string path = PathOf("codec_" + codec + ".parquet");
ASSERT_OK(Write(path, schema, batch, codec)) << "codec " << codec;
@@ -291,9 +293,14 @@ TEST_F(ParquetFormatBenchmarkTest, RegisteredCodecsWrite) {
EXPECT_TRUE(result.data->Equals(*batch)) << "codec " << codec;
}
- // The name the benchmark deliberately does not register still has to be
rejected; if arrow
- // ever starts accepting it, the comment explaining its absence is stale.
- EXPECT_FALSE(Write(PathOf("codec_lz4.parquet"), schema, batch,
"lz4").ok());
+ // The name the benchmark deliberately does not register still has to be
one Parquet rejects; if
+ // arrow ever starts accepting it, the comment explaining its absence is
stale. Checked at the
+ // name-resolution layer, not by writing a file: an actual LZ4_FRAME write
reaches parquet's
+ // Thrift conversion, whose default branch is a DCHECK, so it aborts a
Debug build instead of
+ // returning an error.
+ ASSERT_OK_AND_ASSIGN(arrow::Compression::type lz4,
ArrowUtils::GetCompressionType("lz4"));
+ EXPECT_EQ(arrow::Compression::LZ4_FRAME, lz4);
+ EXPECT_FALSE(::parquet::IsCodecSupported(lz4));
}
// A dictionary-encoded input array must reach the writer intact. VARCHAR
takes arrow's direct
@@ -511,7 +518,7 @@ TEST_F(ParquetFormatBenchmarkTest, NestedRoundTrip) {
// Each nested column also has to be readable on its own, which is what
the projected nested
// cases do.
- for (const std::string& column : {"tags", "attrs"}) {
+ for (const std::string column : {"tags", "attrs"}) {
std::shared_ptr<arrow::Schema> projected =
arrow::schema({schema->GetFieldByName(column)});
ASSERT_OK_AND_ASSIGN(ReadResult projected_result, Read(path,
projected));
EXPECT_EQ(kRows, projected_result.rows) << "column " << column;
diff --git a/docs/source/api/read.rst b/docs/source/api/read.rst
index 3c437d7d..ccf21c6a 100644
--- a/docs/source/api/read.rst
+++ b/docs/source/api/read.rst
@@ -32,6 +32,8 @@ Interface
:members:
:undoc-members:
+.. doxygenenum:: paimon::WarmupLevel
+
.. doxygenclass:: paimon::ReadContext
:members:
:undoc-members:
diff --git a/include/paimon/read_context.h b/include/paimon/read_context.h
index 8d34aab5..25cf7ec5 100644
--- a/include/paimon/read_context.h
+++ b/include/paimon/read_context.h
@@ -62,7 +62,8 @@ class PAIMON_EXPORT ReadContext {
const std::shared_ptr<RealtimeContext>& realtime_context,
const std::map<std::string, std::string>& options, bool
read_ahead_cache_enabled,
const CacheConfig& cache_config, const std::shared_ptr<Cache>&
cache,
- const std::shared_ptr<FormatTable>& format_table);
+ const std::shared_ptr<FormatTable>& format_table, WarmupLevel
warmup_level);
+
~ReadContext();
const std::string& GetPath() const {
@@ -150,6 +151,10 @@ class PAIMON_EXPORT ReadContext {
return format_table_;
}
+ WarmupLevel GetWarmupLevel() const {
+ return warmup_level_;
+ }
+
/// Whether a read schema (C ArrowSchema) for nested column pruning was
provided.
bool HasReadSchema() const {
return read_schema_ != nullptr && read_schema_->release != nullptr;
@@ -190,6 +195,7 @@ class PAIMON_EXPORT ReadContext {
CacheConfig cache_config_;
std::shared_ptr<Cache> cache_;
std::shared_ptr<FormatTable> format_table_;
+ WarmupLevel warmup_level_;
// Owns schema resources and releases ArrowSchema::release in destructor.
std::unique_ptr<ArrowSchema> read_schema_;
};
@@ -351,6 +357,19 @@ 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.
+ /// @note WarmupLevel::RAW warms the read-ahead cache, so it has no effect
and behaves like
+ /// WarmupLevel::NONE when the cache is off (see
SetReadAheadCacheEnabled()).
+ /// WarmupLevel::DECODED does not depend on the cache.
+ /// @see WarmupLevel
+ ReadContextBuilder& SetWarmupLevel(WarmupLevel level);
+
/// Set the total number of batches to prefetch across all files.
///
/// This controls the memory usage and parallelism of the prefetching
mechanism.
diff --git a/include/paimon/reader/file_batch_reader.h
b/include/paimon/reader/file_batch_reader.h
index 87771be3..32f2508c 100644
--- a/include/paimon/reader/file_batch_reader.h
+++ b/include/paimon/reader/file_batch_reader.h
@@ -63,6 +63,26 @@ 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.
+ ///
+ /// What an implementation starts is up to the implementation - a reader
that prefetches on a
+ /// background thread starts that thread, one that reads synchronously may
do nothing at all.
+ /// @warning Whatever it starts, the call itself is not thread-safe: make
it from the same
+ /// thread that reads this reader, and not concurrently with any other
call on it.
+ virtual void Warmup() {}
};
} // namespace paimon
diff --git a/include/paimon/utils/prefetch_cache_config.h
b/include/paimon/utils/prefetch_cache_config.h
index bd1197a6..9f20fbf2 100644
--- a/include/paimon/utils/prefetch_cache_config.h
+++ b/include/paimon/utils/prefetch_cache_config.h
@@ -114,4 +114,26 @@ class PAIMON_EXPORT CacheConfig {
uint64_t block_cache_limit_ = 1024 * 1024;
};
+/// Controls how far a reader prepares the next file before that file is
actually read.
+///
+/// Warmup overlaps remote-storage latency with the read of the current file.
Each level takes the
+/// next file one step further along the read pipeline: `RAW` covers the
remote fetch, `DECODED`
+/// adds the decode on top of it. Higher levels hide more latency, but commit
more memory and
+/// background I/O to files that a query may end up never reading (for example
when a LIMIT stops
+/// the scan early). Callers can trade latency against memory by picking a
level.
+enum class PAIMON_EXPORT WarmupLevel {
+ /// Do not warm up. The next file's I/O starts only when it is actually
read. This is the
+ /// behavior from before warmup existed and uses no extra memory or
background threads.
+ NONE,
+ /// Fetch only the next file's raw, still-compressed bytes into memory,
and leave the decoder
+ /// alone. Overlaps the remote fetch while keeping memory lower than
`DECODED`, because no
+ /// decoded batches are materialized ahead of the read. It fetches through
the read-ahead
+ /// cache, so it falls back to `NONE` when that cache is disabled.
+ RAW,
+ /// Fetch the raw bytes and start the background decode loop as well, so
decoded batches are
+ /// ready before the file is read. Hides the most latency but uses the
most memory. This is the
+ /// default.
+ DECODED,
+};
+
} // namespace paimon
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 0402cc23..cb083e99 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -636,6 +636,7 @@ if(PAIMON_BUILD_TESTS)
common/predicate/predicate_utils_test.cpp
common/predicate/predicate_validator_test.cpp
common/reader/concat_batch_reader_test.cpp
+ common/reader/data_file_reader_factory_test.cpp
common/reader/late_materializing_file_batch_reader_test.cpp
common/reader/predicate_batch_reader_test.cpp
common/reader/prefetch_file_batch_reader_impl_test.cpp
diff --git
a/src/paimon/common/data/shredding/map_shared_shredding_read_plan_factory_test.cpp
b/src/paimon/common/data/shredding/map_shared_shredding_read_plan_factory_test.cpp
index 1763d6a2..fe15921c 100644
---
a/src/paimon/common/data/shredding/map_shared_shredding_read_plan_factory_test.cpp
+++
b/src/paimon/common/data/shredding/map_shared_shredding_read_plan_factory_test.cpp
@@ -867,4 +867,18 @@ TEST_F(MapSharedShreddingReadPlanFactoryTest,
TestReadsRealFormatFile) {
AssertChunkedArrayEquals(expected, actual);
}
+TEST(ShreddingFileReaderTest, WarmupForwardsToInnerReader) {
+ auto file_type = arrow::struct_({arrow::field("id", arrow::int32())});
+ auto mock_reader =
+ std::make_unique<MockFileBatchReader>(/*data=*/nullptr, file_type,
/*read_batch_size=*/1);
+ auto* inner_reader = mock_reader.get();
+ auto reader = std::make_unique<ShreddingFileReader>(
+ std::move(mock_reader), std::map<std::string,
std::shared_ptr<ShreddingColumnReadPlan>>(),
+ GetArrowPool(GetDefaultPool()));
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader->Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
} // namespace paimon::test
diff --git a/src/paimon/common/data/shredding/shredding_file_reader.h
b/src/paimon/common/data/shredding/shredding_file_reader.h
index d3584c7d..503fbf18 100644
--- a/src/paimon/common/data/shredding/shredding_file_reader.h
+++ b/src/paimon/common/data/shredding/shredding_file_reader.h
@@ -59,6 +59,10 @@ class ShreddingFileReader : public FileBatchReader {
bool SupportPreciseBitmapSelection() const override;
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
private:
std::shared_ptr<arrow::MemoryPool> arrow_pool_;
std::unique_ptr<FileBatchReader> reader_;
diff --git
a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
index 4894ad06..efb159ea 100644
--- a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
+++ b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h
@@ -94,6 +94,10 @@ class ApplyBitmapIndexBatchReader : public FileBatchReader {
return reader_->SupportPreciseBitmapSelection();
}
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
private:
Result<RoaringBitmap32> Filter(int32_t batch_size) const {
RoaringBitmap32 result;
diff --git
a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
index ef01bbea..85f257ca 100644
---
a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
+++
b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
@@ -100,7 +100,8 @@ class ApplyBitmapIndexBatchReaderTest : public
::testing::Test,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_,
GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED,
pool_,
+ GetArrowPool(pool_)));
} else {
file_batch_reader =
std::make_unique<MockFileBatchReader>(data, target_type_,
batch_size);
@@ -193,6 +194,20 @@ TEST_P(ApplyBitmapIndexBatchReaderTest, TestBulkData) {
std::string result_str = fmt::format("[{}]", fmt::join(bitmap_data, ","));
CheckResult(data_str, bitmap_data, result_str,
/*specified_batch_size=*/1024);
}
+
+TEST(ApplyBitmapIndexBatchReaderWarmupTest, WarmupForwardsToInnerReader) {
+ auto target_type = arrow::struct_({arrow::field("f1", arrow::int32())});
+ auto mock_reader =
+ std::make_unique<MockFileBatchReader>(/*data=*/nullptr, target_type,
/*batch_size=*/1);
+ auto* inner_reader = mock_reader.get();
+ ApplyBitmapIndexBatchReader reader(std::move(mock_reader),
+ RoaringBitmap32::From(/*values=*/{0}));
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader.Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
INSTANTIATE_TEST_SUITE_P(EnablePrefetch, ApplyBitmapIndexBatchReaderTest,
::testing::Values(false, true));
diff --git a/src/paimon/common/reader/data_file_reader_factory.cpp
b/src/paimon/common/reader/data_file_reader_factory.cpp
index 4507127f..eeb46958 100644
--- a/src/paimon/common/reader/data_file_reader_factory.cpp
+++ b/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,
+ read_options.warmup_level, pool, arrow_pool));
return
std::make_unique<DelegatingPrefetchReader>(std::move(prefetch_reader));
}
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
diff --git a/src/paimon/common/reader/data_file_reader_factory.h
b/src/paimon/common/reader/data_file_reader_factory.h
index 264ea34b..3be8f660 100644
--- a/src/paimon/common/reader/data_file_reader_factory.h
+++ b/src/paimon/common/reader/data_file_reader_factory.h
@@ -54,6 +54,9 @@ struct DataFileReadOptions {
CacheConfig cache_config;
/// Whether the prefetching reader records what its I/O costs.
bool prefetch_io_metrics_enabled = false;
+ /// How far the prefetching reader prepares a file before it is read. It
only matters when
+ /// prefetching, because that reader is the one that can start work ahead
of the read.
+ WarmupLevel warmup_level = WarmupLevel::DECODED;
};
/// Opens one data file as a `FileBatchReader`.
diff --git a/src/paimon/common/reader/data_file_reader_factory_test.cpp
b/src/paimon/common/reader/data_file_reader_factory_test.cpp
new file mode 100644
index 00000000..895baad4
--- /dev/null
+++ b/src/paimon/common/reader/data_file_reader_factory_test.cpp
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "paimon/common/reader/data_file_reader_factory.h"
+
+#include <memory>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/array/array_nested.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/reader/delegating_prefetch_reader.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/executor.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/reader/file_batch_reader.h"
+#include "paimon/testing/mock/mock_file_system.h"
+#include "paimon/testing/mock/mock_format_reader_builder.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/prefetch_cache_config.h"
+
+namespace paimon::test {
+
+/// `Open()` is the one place a production read builds the prefetching reader,
and a level that is
+/// not forwarded here fails nothing: the reader warms up however its own
default says, and the only
+/// symptom is latency and memory a query never asked for.
+TEST(DataFileReaderFactoryTest, OpenForwardsWarmupLevelToPrefetchReader) {
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+ auto mock_fs = std::make_shared<MockFileSystem>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor,
+ CreateDefaultExecutor(/*thread_count=*/1));
+
+ std::shared_ptr<arrow::DataType> data_type =
+ arrow::struct_({arrow::field("f1", arrow::int32())});
+ std::shared_ptr<arrow::Array> data =
+ arrow::StructArray::Make(
+ {arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[1, 2,
3]").ValueOrDie()},
+ data_type->fields())
+ .ValueOrDie();
+
+ for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW,
WarmupLevel::DECODED}) {
+ MockFormatReaderBuilder reader_builder(data, data_type,
/*read_batch_size=*/10);
+ DataFileReadOptions read_options;
+ read_options.read_batch_size = 10;
+ read_options.prefetch_enabled = true;
+ read_options.prefetch_max_parallel_num = 1;
+ read_options.prefetch_batch_count = 2;
+ // No file backs the mock file system, so keep the read-ahead cache
out of the read.
+ read_options.read_ahead_cache_enabled = false;
+ read_options.warmup_level = level;
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> reader,
+ DataFileReaderFactory::Open(
+ "parquet", /*file_path=*/"DUMMY",
/*file_size=*/0, &reader_builder,
+ read_options, mock_fs, executor, pool,
GetArrowPool(pool)));
+
+ // `Open()` hands back the delegating wrapper. The level has to have
arrived on the
+ // prefetching reader underneath it, because that reader's Warmup() is
what it governs.
+ auto* delegating_reader =
dynamic_cast<DelegatingPrefetchReader*>(reader.get());
+ ASSERT_NE(delegating_reader, nullptr);
+ EXPECT_EQ(level, delegating_reader->prefetch_reader_->warmup_level_);
+ }
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/reader/delegating_prefetch_reader.h
b/src/paimon/common/reader/delegating_prefetch_reader.h
index 78396e20..b91c0418 100644
--- a/src/paimon/common/reader/delegating_prefetch_reader.h
+++ b/src/paimon/common/reader/delegating_prefetch_reader.h
@@ -73,6 +73,15 @@ class DelegatingPrefetchReader : public FileBatchReader {
return GetReader()->SupportPreciseBitmapSelection();
}
+ void Warmup() override {
+ // GetReader() bypasses the prefetch layer entirely when prefetching
is off, and a
+ // bypassed reader has no background thread to start.
+ if (!prefetch_reader_->NeedPrefetch()) {
+ return;
+ }
+ prefetch_reader_->Warmup();
+ }
+
private:
inline FileBatchReader* GetReader() const {
assert(prefetch_reader_);
diff --git
a/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
b/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
index c86635f5..4346eeed 100644
--- a/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
+++ b/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
@@ -549,7 +549,7 @@ TEST_F(LateMaterializingFileBatchReaderTest,
WorksAsInnerOfPrefetchReader) {
/*prefetch_max_parallel_num=*/1, /*batch_size=*/3,
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto predicate =
PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k",
FieldType::BIGINT, Literal(4l));
::ArrowSchema c_schema;
@@ -585,7 +585,7 @@ TEST_F(LateMaterializingFileBatchReaderTest,
PrefetchInnerReentrantSetReadSchema
/*prefetch_max_parallel_num=*/1, /*batch_size=*/3,
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto full_schema = arrow::schema(full_fields_);
auto predicate1 =
@@ -648,7 +648,7 @@ TEST_F(LateMaterializingFileBatchReaderTest,
PrefetchInnerParallelReadersWithSee
/*prefetch_max_parallel_num=*/3, /*batch_size=*/3,
/*prefetch_batch_count=*/6,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto predicate =
PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k",
FieldType::BIGINT, Literal(5l));
::ArrowSchema c_schema;
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
index 51589ba3..da7ba458 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
@@ -205,7 +205,8 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
bool read_ahead_cache_enabled, const CacheConfig& cache_config, bool
enable_io_metrics,
- const std::shared_ptr<MemoryPool>& pool, const
std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
+ WarmupLevel warmup_level, const std::shared_ptr<MemoryPool>& pool,
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
if (prefetch_max_parallel_num == 0) {
return Status::Invalid("prefetch max parallel num should be greater
than 0.");
}
@@ -283,7 +284,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(new
PrefetchFileBatchReaderImpl(
readers, batch_size, prefetch_queue_capacity,
enable_adaptive_prefetch_strategy, executor,
- cache, io_metrics, arrow_pool));
+ cache, io_metrics, warmup_level, arrow_pool));
if (initialize_read_ranges) {
// normally initialize read ranges should be false, as set read schema
will refresh read
// ranges, and set read schema will always be called before read.
@@ -296,7 +297,7 @@ PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl(
const std::vector<std::shared_ptr<PrefetchFileBatchReader>>& readers,
int32_t batch_size,
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<PrefetchIoMetricsState>& io_metrics, WarmupLevel
warmup_level,
const std::shared_ptr<arrow::MemoryPool>& arrow_pool)
: readers_(std::move(readers)),
batch_size_(batch_size),
@@ -305,6 +306,7 @@ PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl(
arrow_pool_(arrow_pool),
prefetch_queue_capacity_(prefetch_queue_capacity),
enable_adaptive_prefetch_strategy_(enable_adaptive_prefetch_strategy),
+ warmup_level_(warmup_level),
prefetch_metrics_(std::make_shared<PrefetchMetricsState>()),
io_metrics_(io_metrics) {
for (size_t i = 0; i < readers_.size(); i++) {
@@ -470,6 +472,9 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
read_ranges_in_group_.clear();
current_batch_global_row_ids_.clear();
read_ranges_freshed_ = false;
+ // SetReadSchema()/RefreshReadRanges() call cache_->Reset() right after
CleanUp(), clearing the
+ // cache's initialized state, so the next read-range generation must be
allowed to Init again.
+ cache_warmed_.store(false);
clean_prefetch_queue();
prefetch_metrics_->queue_depth.store(0, kMetricsMemoryOrder);
for (size_t i = 0; i < readers_pos_.size(); i++) {
@@ -484,26 +489,10 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
void PrefetchFileBatchReaderImpl::Workloop() {
std::vector<std::future<void>> futures;
futures.resize(readers_.size());
- if (cache_) {
- auto read_ranges = readers_[0]->PreBufferRange();
- if (read_ranges.ok()) {
- std::vector<ByteRange> ranges;
- for (const auto& read_range : read_ranges.value()) {
- ranges.emplace_back(read_range.first, read_range.second);
- }
- auto s = cache_->Init(std::move(ranges));
- if (!s.ok()) {
- SetReadStatus(s);
- } else {
- // Init() only registers the ranges, so without this the first
- // cache fetch races the readers' first reads instead of
running
- // ahead of them.
- cache_->Warmup();
- }
- } else {
- SetReadStatus(read_ranges.status());
- }
- }
+ // Warm the read-ahead cache before decoding. At DECODED this is the first
warmup; at RAW
+ // Warmup() already ran it on the caller's thread and the guard makes this
a no-op, so the
+ // cache is never Init'd twice.
+ WarmCacheOnce();
while (true) {
if (!GetReadStatus().ok()) {
@@ -729,14 +718,69 @@ Status PrefetchFileBatchReaderImpl::DoReadBatch(size_t
reader_idx) {
return HandleReadResult(reader_idx, read_range,
std::move(read_batch_with_bitmap));
}
-Result<BatchReader::ReadBatchWithBitmap>
PrefetchFileBatchReaderImpl::NextBatchWithBitmap() {
- if (!read_ranges_freshed_) {
- return Status::Invalid("prefetch reader read ranges are not
initialized");
- }
+void PrefetchFileBatchReaderImpl::EnsureBackgroundThread() {
if (!background_thread_) {
background_thread_ =
std::make_unique<std::thread>(&PrefetchFileBatchReaderImpl::Workloop, this);
}
+}
+
+void PrefetchFileBatchReaderImpl::WarmCacheOnce() {
+ if (!cache_) {
+ return;
+ }
+ // Init() is not idempotent (a second call returns Invalid), so at most
one warmup may run per
+ // read-range generation. A RAW Warmup() and the Workloop() call are
ordered by the reader's own
+ // thread, which warms first and only then starts the background thread,
so the loser of this
+ // exchange never continues on a cache another thread is still
initializing.
+ if (cache_warmed_.exchange(true)) {
+ return;
+ }
+ auto read_ranges = readers_[0]->PreBufferRange();
+ if (!read_ranges.ok()) {
+ SetReadStatus(read_ranges.status());
+ return;
+ }
+ std::vector<ByteRange> ranges;
+ for (const auto& read_range : read_ranges.value()) {
+ ranges.emplace_back(read_range.first, read_range.second);
+ }
+ Status s = cache_->Init(std::move(ranges));
+ if (!s.ok()) {
+ SetReadStatus(s);
+ return;
+ }
+ // Init() only registers the ranges, so without this the first cache fetch
races the readers'
+ // first reads instead of running ahead of them.
+ cache_->Warmup();
+}
+
+void PrefetchFileBatchReaderImpl::Warmup() {
+ if (warmup_level_ == WarmupLevel::NONE) {
+ return;
+ }
+ // Ranges that are not set mean the reader is either not configured yet or
already cleaned up,
+ // and CleanUp() leaves no background thread behind, so there is nothing
to warm up in either
+ // state. NextBatchWithBitmap still rejects the former, so a genuinely
unprepared read is
+ // reported there rather than swallowed here.
+ if (!read_ranges_freshed_) {
+ return;
+ }
+ if (warmup_level_ == WarmupLevel::RAW) {
+ // Prefetch the raw bytes only, without starting the decoder. The
first real read starts the
+ // background thread, whose Workloop() finds the cache already warmed
and skips re-Init.
+ WarmCacheOnce();
+ return;
+ }
+ // DECODED: start the background decode loop now so decoded batches are
ready before the read.
+ EnsureBackgroundThread();
+}
+
+Result<BatchReader::ReadBatchWithBitmap>
PrefetchFileBatchReaderImpl::NextBatchWithBitmap() {
+ if (!read_ranges_freshed_) {
+ return Status::Invalid("prefetch reader read ranges are not
initialized");
+ }
+ EnsureBackgroundThread();
const auto wait_start = std::chrono::steady_clock::now();
while (true) {
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
b/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
index 37bf2f89..b0b0c6f7 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
@@ -68,7 +68,7 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
uint32_t prefetch_max_parallel_num, int32_t batch_size, uint32_t
prefetch_batch_count,
bool enable_adaptive_prefetch_strategy, const
std::shared_ptr<Executor>& executor,
bool initialize_read_ranges, bool read_ahead_cache_enabled, const
CacheConfig& cache_config,
- bool enable_io_metrics, const std::shared_ptr<MemoryPool>& pool,
+ bool enable_io_metrics, WarmupLevel warmup_level, const
std::shared_ptr<MemoryPool>& pool,
const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
~PrefetchFileBatchReaderImpl() override;
@@ -103,6 +103,8 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
Status RefreshReadRanges();
+ void Warmup() override;
+
inline PrefetchFileBatchReader* GetFirstReader() const {
return readers_[0].get();
}
@@ -122,11 +124,23 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
const std::vector<std::shared_ptr<PrefetchFileBatchReader>>& readers,
int32_t batch_size,
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<PrefetchIoMetricsState>& io_metrics, WarmupLevel
warmup_level,
const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
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. The flag only elects the caller that
initializes, it does
+ /// not make a second caller wait: the two calls are ordered rather than
concurrent, because a
+ /// RAW warmup runs on the reader's own thread before that same thread
starts the background
+ /// thread that runs Workloop(). 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();
void SetReadStatus(const Status& status);
Status GetReadStatus() const;
Result<bool> IsEofRange(const std::pair<uint64_t, uint64_t>& read_range)
const;
@@ -179,8 +193,12 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
std::vector<uint64_t> current_batch_global_row_ids_;
bool need_prefetch_ = false;
bool read_ranges_freshed_ = false;
+ // Guards the one-shot read-ahead cache Init/Warmup so it runs at most
once per read-range
+ // generation. Reset by CleanUp() alongside read_ranges_freshed_.
+ std::atomic<bool> cache_warmed_{false};
const uint32_t prefetch_queue_capacity_;
const bool enable_adaptive_prefetch_strategy_;
+ const WarmupLevel warmup_level_;
int32_t parallel_num_;
std::shared_ptr<PrefetchMetricsState> prefetch_metrics_;
std::shared_ptr<PrefetchIoMetricsState> io_metrics_;
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
index 5e0c40d4..27375241 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
@@ -25,6 +25,7 @@
#include "arrow/compute/api.h"
#include "arrow/ipc/api.h"
#include "gtest/gtest.h"
+#include "paimon/common/reader/delegating_prefetch_reader.h"
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/path_util.h"
@@ -253,7 +254,8 @@ class PrefetchFileBatchReaderImplTest : public
::testing::Test,
const std::string& file_format_str, const arrow::Schema* read_schema,
const std::shared_ptr<Predicate>& predicate,
const std::optional<RoaringBitmap32>& selection_bitmap, int32_t
batch_size,
- int32_t prefetch_max_parallel_num, bool read_ahead_cache_enabled)
const {
+ int32_t prefetch_max_parallel_num, bool read_ahead_cache_enabled,
+ WarmupLevel warmup_level) const {
EXPECT_OK_AND_ASSIGN(std::unique_ptr<FileFormat> file_format,
FileFormatFactory::Get(file_format_str, {}));
EXPECT_OK_AND_ASSIGN(auto reader_builder,
file_format->CreateReaderBuilder(batch_size));
@@ -269,7 +271,7 @@ class PrefetchFileBatchReaderImplTest : public
::testing::Test,
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false, read_ahead_cache_enabled,
CacheConfig(),
- /*enable_io_metrics=*/true, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/true, warmup_level, pool_,
GetArrowPool(pool_)));
std::unique_ptr<ArrowSchema> c_schema =
std::make_unique<ArrowSchema>();
auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get());
EXPECT_TRUE(arrow_status.ok());
@@ -335,6 +337,17 @@ CollectResultAndRowIds(FileBatchReader* reader) {
return std::make_pair(result_array, row_ids);
}
+/// Whether Warmup() started the background decode loop.
+///
+/// The levels differ only in what Warmup() itself starts, and metrics cannot
show that: the first
+/// NextBatch starts the same loop lazily, so once a read is under way every
level reports produced
+/// batches, while right after Warmup() a DECODED warmup may not have finished
its first batch yet.
+/// Whether the thread exists is settled the moment Warmup() returns, so it
separates the levels
+/// without racing the background work.
+bool DecodeThreadStarted(const PrefetchFileBatchReaderImpl* reader) {
+ return reader->background_thread_ != nullptr;
+}
+
std::vector<TestParam> PrepareTestParam() {
std::vector<TestParam> values = {TestParam{"parquet",
/*read_ahead_cache_enabled=*/true},
TestParam{"parquet",
/*read_ahead_cache_enabled=*/false}};
@@ -360,7 +373,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestSimple) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
if (prefetch_max_parallel_num == 1) {
ASSERT_NOK(
reader->GetReaderMetrics()->GetCounter(PrefetchIoMetrics::READ_LATENCY_COUNT));
@@ -388,7 +401,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestReadWithLimits)
{
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/true, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/true, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
// simulate read limits, only read 8 batches
for (int32_t i = 0; i < 8; i++) {
ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap
batch_with_bitmap,
@@ -442,7 +455,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestReadWithoutInitializeReadRanges) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
// simulate read limits, only read 8 batches
ASSERT_NOK_WITH_MSG(reader->NextBatchWithBitmap(),
"prefetch reader read ranges are not initialized");
@@ -460,7 +473,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestFailedIoMetrics) {
/*prefetch_max_parallel_num=*/1, /*batch_size=*/10,
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- /*enable_io_metrics=*/true, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/true, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
ASSERT_NOK_WITH_MSG(reader->NextBatchWithBitmap(), "injected synchronous
read failure");
std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
@@ -549,7 +562,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, RefreshReadRanges) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
ASSERT_OK(prefetch_reader->RefreshReadRanges());
std::vector<std::pair<uint64_t, uint64_t>> read_ranges_0 = {{0, 30}, {90,
101}};
@@ -580,7 +593,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
RefreshReadRangesDisablePrefetchByAdapti
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/true, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
ASSERT_FALSE(reader->NeedPrefetch());
std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
@@ -593,6 +606,69 @@ TEST_F(PrefetchFileBatchReaderImplTest,
RefreshReadRangesDisablePrefetchByAdapti
ASSERT_EQ(adaptive_disabled_count, 1);
}
+// DelegatingPrefetchReader is the layer that decides whether a read goes
through the prefetch
+// machinery or straight to the first inner reader, so its Warmup() has to
follow the same decision.
+// With prefetching on it must forward, otherwise a DECODED warmup silently
degrades to a cold
+// read.
+TEST_F(PrefetchFileBatchReaderImplTest,
DelegatingPrefetchReaderWarmupForwardsWhenPrefetchEnabled) {
+ auto data_array = PrepareArray(100);
+ int32_t batch_size = 10;
+ int32_t prefetch_max_parallel_num = 1;
+ MockFormatReaderBuilder reader_builder(data_array, data_type_, batch_size);
+ ASSERT_OK_AND_ASSIGN(
+ auto reader,
+ PrefetchFileBatchReaderImpl::Create(
+ /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder,
mock_fs_,
+ prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
+ /*enable_adaptive_prefetch_strategy=*/false, executor_,
+ /*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
+ auto* prefetch_reader = reader.get();
+ ASSERT_TRUE(prefetch_reader->NeedPrefetch());
+
+ DelegatingPrefetchReader delegating_reader(std::move(reader));
+ delegating_reader.Warmup();
+ ASSERT_TRUE(DecodeThreadStarted(prefetch_reader));
+
+ ASSERT_OK_AND_ASSIGN(auto array,
ReadResultCollector::CollectResult(&delegating_reader));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
+ ASSERT_TRUE(expected_array->Equals(array));
+}
+
+// With prefetching off the delegating layer hands the consumer readers_[0]
directly, so forwarding
+// the warmup would start a Workloop that decodes readers_[0] on a background
thread while the
+// consumer reads that same reader on its own thread. Warmup() must stay a
no-op in that state and
+// the bypassed read must still return every row.
+TEST_F(PrefetchFileBatchReaderImplTest,
DelegatingPrefetchReaderWarmupSkippedWhenPrefetchBypassed) {
+ auto data_array = PrepareArray(200);
+ int32_t batch_size = 10;
+ int32_t prefetch_max_parallel_num = 1;
+ ControlledMockFormatReaderBuilder reader_builder(data_array, data_type_,
batch_size,
+ /*read_ranges=*/{{0,
100}},
+ /*need_prefetch=*/true,
+
/*set_read_ranges_statuses=*/{});
+ ASSERT_OK_AND_ASSIGN(
+ auto reader,
+ PrefetchFileBatchReaderImpl::Create(
+ /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder,
mock_fs_,
+ prefetch_max_parallel_num, batch_size,
+ /*prefetch_batch_count=*/2,
+ /*enable_adaptive_prefetch_strategy=*/true, executor_,
+ /*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
+ auto* prefetch_reader = reader.get();
+ ASSERT_FALSE(prefetch_reader->NeedPrefetch());
+
+ DelegatingPrefetchReader delegating_reader(std::move(reader));
+ delegating_reader.Warmup();
+ ASSERT_FALSE(DecodeThreadStarted(prefetch_reader));
+
+ ASSERT_OK_AND_ASSIGN(auto array,
ReadResultCollector::CollectResult(&delegating_reader));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
+ ASSERT_TRUE(expected_array->Equals(array));
+ ASSERT_FALSE(DecodeThreadStarted(prefetch_reader));
+}
+
TEST_F(PrefetchFileBatchReaderImplTest, SetReadRanges) {
auto data_array = PrepareArray(400);
int32_t batch_size = 30;
@@ -605,7 +681,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, SetReadRanges) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
ASSERT_FALSE(prefetch_reader->need_prefetch_);
prefetch_reader->need_prefetch_ = true;
@@ -649,7 +725,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
SetReadRangesReturnErrorWhenPushDownFail
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->need_prefetch_ = true;
@@ -670,13 +746,13 @@ TEST_F(PrefetchFileBatchReaderImplTest,
WorkloopSetReadStatusWhenCacheInitFailed
invalid_cache_config.SetPreBufferLimit(128 * 1024);
ASSERT_OK_AND_ASSIGN(
- auto reader,
- PrefetchFileBatchReaderImpl::Create(
- /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder,
mock_fs_,
- prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
- /*enable_adaptive_prefetch_strategy=*/false, executor_,
- /*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- invalid_cache_config, /*enable_io_metrics=*/false, pool_,
GetArrowPool(pool_)));
+ auto reader, PrefetchFileBatchReaderImpl::Create(
+ /*data_file_path=*/"", /*data_file_size=*/0,
&reader_builder, mock_fs_,
+ prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
+ /*enable_adaptive_prefetch_strategy=*/false,
executor_,
+ /*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
+ invalid_cache_config, /*enable_io_metrics=*/false,
WarmupLevel::DECODED,
+ pool_, GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->Workloop();
@@ -697,7 +773,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
DoReadBatchReturnOkWhenShutdown) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->is_shutdown_ = true;
@@ -716,7 +792,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
DoReadBatchReturnOkWhenNoCurrentReadRang
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->read_ranges_in_group_ = {{}};
@@ -735,7 +811,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestReadWithLargeBatchSize) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
auto row_ids = array_and_row_ids.second;
@@ -756,7 +832,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestPartialReaderSuccessRead) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
for (int32_t i = 0; i < prefetch_max_parallel_num; i++) {
dynamic_cast<MockFileBatchReader*>(prefetch_reader->readers_[i].get())
@@ -802,7 +878,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestAllReaderFailedWithIOError) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
for (int32_t i = 0; i < prefetch_max_parallel_num; i++) {
@@ -841,7 +917,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestPrefetchWithEmptyData) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
auto row_ids = array_and_row_ids.second;
@@ -861,7 +937,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestCallNextBatchAfterReadingEof) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
auto row_ids = array_and_row_ids.second;
@@ -905,7 +981,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestCreateReaderWithoutNextBatch) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
}
TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
@@ -920,7 +996,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
/*prefetch_max_parallel_num=*/0, batch_size, 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -928,7 +1004,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, /*batch_size=*/-1,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
}
{
ASSERT_NOK_WITH_MSG(
@@ -937,7 +1013,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)),
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)),
"data file size should not be negative");
}
{
@@ -946,8 +1022,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false,
/*executor=*/nullptr, /*initialize_read_ranges=*/true,
- /*read_ahead_cache_enabled=*/true, CacheConfig(),
/*enable_io_metrics=*/false, pool_,
- GetArrowPool(pool_)));
+ /*read_ahead_cache_enabled=*/true, CacheConfig(),
/*enable_io_metrics=*/false,
+ WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -955,7 +1031,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -963,7 +1039,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
/*fs=*/nullptr, prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
}
{
ASSERT_OK_AND_ASSIGN(
@@ -973,7 +1049,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool_, GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_,
GetArrowPool(pool_)));
ASSERT_NOK_WITH_MSG(reader->SeekToRow(/*row_number=*/101),
"not support seek to row for prefetch reader");
}
@@ -999,7 +1075,7 @@ TEST_P(PrefetchFileBatchReaderImplTest,
TestPrefetchWithPredicatePushdownWithCom
auto reader = PreparePrefetchReader(file_format, schema.get(), predicate,
/*selection_bitmap=*/std::nullopt,
/*batch_size=*/batch_size,
/*prefetch_max_parallel_num=*/3,
- read_ahead_cache_enabled);
+ read_ahead_cache_enabled,
WarmupLevel::DECODED);
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
arrow::ArrayVector expected_array_vector;
@@ -1045,7 +1121,7 @@ TEST_P(PrefetchFileBatchReaderImplTest,
auto reader = PreparePrefetchReader(file_format, schema.get(), predicate,
/*selection_bitmap=*/std::nullopt,
/*batch_size=*/batch_size,
/*prefetch_max_parallel_num=*/3,
- read_ahead_cache_enabled);
+ read_ahead_cache_enabled,
WarmupLevel::DECODED);
ASSERT_OK(reader->RefreshReadRanges());
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
@@ -1077,7 +1153,8 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestPrefetchWithBitmap) {
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
CacheConfig(),
- /*enable_io_metrics=*/false, pool_,
GetArrowPool(pool_)));
+ /*enable_io_metrics=*/false,
WarmupLevel::DECODED, pool_,
+ GetArrowPool(pool_)));
ASSERT_OK_AND_ASSIGN(auto result_chunk_array,
ReadResultCollector::CollectResult(std::move(reader)));
@@ -1108,7 +1185,7 @@ TEST_P(PrefetchFileBatchReaderImplTest, TestRowMapping) {
auto reader = PreparePrefetchReader(file_format, schema.get(), predicate,
/*selection_bitmap=*/std::nullopt,
/*batch_size=*/10,
/*prefetch_max_parallel_num=*/3,
- read_ahead_cache_enabled);
+ read_ahead_cache_enabled,
WarmupLevel::DECODED);
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> batch,
paimon::test::ReadResultCollector::CollectResultOneBatch(reader.get()));
@@ -1158,4 +1235,146 @@ TEST_P(PrefetchFileBatchReaderImplTest, TestRowMapping)
{
}
}
+// WarmupLevel::NONE makes Warmup() a complete no-op: it neither starts the
background decode thread
+// nor warms the read-ahead cache. Reading still returns every row, starting
cold on the first
+// NextBatch, which lazily starts the background loop.
+TEST_P(PrefetchFileBatchReaderImplTest, TestWarmupLevelNone) {
+ auto [file_format, read_ahead_cache_enabled] = GetParam();
+ auto data_array = PrepareArray(90);
+ PrepareTestData(file_format, data_array, /*stripe_row_count=*/30,
/*row_index_stride=*/10);
+ auto schema = arrow::schema(fields_);
+ auto reader = PreparePrefetchReader(file_format, schema.get(),
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt,
/*batch_size=*/10,
+ /*prefetch_max_parallel_num=*/3,
read_ahead_cache_enabled,
+ WarmupLevel::NONE);
+
+ reader->Warmup();
+
+ // Warmup() did nothing: it started no decode thread, decoded no batch and
the cache issued no
+ // prefetch IO.
+ ASSERT_FALSE(DecodeThreadStarted(reader.get()));
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t produced_batches,
+
metrics->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+ ASSERT_EQ(produced_batches, 0);
+ if (read_ahead_cache_enabled) {
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
+
metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(io_count, 0);
+ }
+
+ ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
+ ASSERT_TRUE(expected_array->Equals(array_and_row_ids.first));
+}
+
+// WarmupLevel::RAW fetches the file's raw bytes into memory but does NOT
start the decoder, so no
+// batch is produced until the first real read. It overlaps the remote fetch
while keeping memory
+// lower than DECODED.
+TEST_P(PrefetchFileBatchReaderImplTest, TestWarmupLevelRaw) {
+ auto [file_format, read_ahead_cache_enabled] = GetParam();
+ auto data_array = PrepareArray(90);
+ PrepareTestData(file_format, data_array, /*stripe_row_count=*/30,
/*row_index_stride=*/10);
+ auto schema = arrow::schema(fields_);
+ auto reader = PreparePrefetchReader(file_format, schema.get(),
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt,
/*batch_size=*/10,
+ /*prefetch_max_parallel_num=*/3,
read_ahead_cache_enabled,
+ WarmupLevel::RAW);
+
+ reader->Warmup();
+ // A second Warmup() must be a no-op: the one-shot guard prevents a
duplicate cache Init(),
+ // which would otherwise fail with "Cache has already been initialized".
+ reader->Warmup();
+
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ // The decoder never started, so no decode thread exists and no batch has
been produced yet.
+ ASSERT_FALSE(DecodeThreadStarted(reader.get()));
+ if (read_ahead_cache_enabled) {
+ // The cache was warmed synchronously on the caller's thread, so
prefetch IO was issued.
+ ASSERT_OK_AND_ASSIGN(uint64_t io_count,
+
metrics->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_GT(io_count, 0);
+ }
+ ASSERT_OK_AND_ASSIGN(uint64_t produced_batches,
+
metrics->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+ ASSERT_EQ(produced_batches, 0);
+
+ // Reading returns every row: the background loop started by the first
NextBatch finds the
+ // cache already warmed and skips re-initializing it.
+ ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
+ ASSERT_TRUE(expected_array->Equals(array_and_row_ids.first));
+}
+
+// RefreshReadRanges() starts a new read-range generation, and it resets the
read-ahead cache right
+// after CleanUp(), so the generation begins cold. The one-shot warmup guard
has to be rearmed at
+// the same time: otherwise every generation after the first skips its warmup
and reads cold while
+// the guard still claims the cache is warm.
+TEST_P(PrefetchFileBatchReaderImplTest,
TestWarmupCacheRearmsForNewReadRangeGeneration) {
+ auto [file_format, read_ahead_cache_enabled] = GetParam();
+ if (!read_ahead_cache_enabled) {
+ GTEST_SKIP() << "there is no read-ahead cache to rearm";
+ }
+ auto data_array = PrepareArray(90);
+ PrepareTestData(file_format, data_array, /*stripe_row_count=*/30,
/*row_index_stride=*/10);
+ auto schema = arrow::schema(fields_);
+ auto reader = PreparePrefetchReader(file_format, schema.get(),
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt,
/*batch_size=*/10,
+ /*prefetch_max_parallel_num=*/3,
read_ahead_cache_enabled,
+ WarmupLevel::RAW);
+
+ // GetReaderMetrics() returns a snapshot, so it is re-read at every step.
+ ASSERT_OK(reader->RefreshReadRanges());
+ reader->Warmup();
+ ASSERT_OK_AND_ASSIGN(uint64_t first_io_count,
+
reader->GetReaderMetrics()->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_GT(first_io_count, 0);
+
+ ASSERT_OK(reader->RefreshReadRanges());
+ ASSERT_OK_AND_ASSIGN(uint64_t reset_io_count,
+
reader->GetReaderMetrics()->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_EQ(reset_io_count, 0);
+
+ // The new generation must be warmed again, which a stale guard would
silently skip.
+ reader->Warmup();
+ ASSERT_OK_AND_ASSIGN(uint64_t second_io_count,
+
reader->GetReaderMetrics()->GetCounter(ReadAheadCacheMetrics::IO_COUNT));
+ ASSERT_GT(second_io_count, 0);
+
+ ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
+ ASSERT_TRUE(expected_array->Equals(array_and_row_ids.first));
+}
+
+// WarmupLevel::DECODED (the default) starts the background decode loop during
Warmup(). Reading
+// returns every row and reports produced batches, matching the behavior from
before warmup levels
+// existed.
+TEST_P(PrefetchFileBatchReaderImplTest, TestWarmupLevelDecoded) {
+ auto [file_format, read_ahead_cache_enabled] = GetParam();
+ auto data_array = PrepareArray(90);
+ PrepareTestData(file_format, data_array, /*stripe_row_count=*/30,
/*row_index_stride=*/10);
+ auto schema = arrow::schema(fields_);
+ auto reader = PreparePrefetchReader(file_format, schema.get(),
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt,
/*batch_size=*/10,
+ /*prefetch_max_parallel_num=*/3,
read_ahead_cache_enabled,
+ WarmupLevel::DECODED);
+
+ reader->Warmup();
+
+ // The discriminating assertion: DECODED is the only level that starts the
decode loop inside
+ // Warmup() itself, and it must have done so before the first read.
Checking produced batches
+ // instead cannot tell DECODED from NONE, because the first NextBatch
starts the same loop
+ // lazily and every level ends up with produced_batches > 0 once the read
is over.
+ ASSERT_TRUE(DecodeThreadStarted(reader.get()));
+
+ ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
+ auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
+ ASSERT_TRUE(expected_array->Equals(array_and_row_ids.first));
+
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t produced_batches,
+
metrics->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+ ASSERT_GT(produced_batches, 0);
+}
+
} // namespace paimon::test
diff --git a/src/paimon/common/utils/arrow/status_utils.h
b/src/paimon/common/utils/arrow/status_utils.h
index a0b505f6..13f9a1cd 100644
--- a/src/paimon/common/utils/arrow/status_utils.h
+++ b/src/paimon/common/utils/arrow/status_utils.h
@@ -87,18 +87,20 @@ inline Status ToPaimonStatus(const arrow::Status& status) {
}
}
+// The conversion is spelled with its full name because these macros are also
used outside of
+// namespace paimon, where an unqualified name would not be found.
#define PAIMON_RETURN_NOT_OK_FROM_ARROW(ARROW_STATUS) \
do { \
arrow::Status __s = (ARROW_STATUS); \
if (PAIMON_UNLIKELY(!(__s).ok())) { \
- return ToPaimonStatus(__s); \
+ return ::paimon::ToPaimonStatus(__s); \
} \
} while (false)
-#define PAIMON_ASSIGN_OR_RAISE_IMPL_FROM_ARROW(result_name, lhs, rexpr)
\
- auto&& result_name = (rexpr);
\
- PAIMON_RETURN_IF_(!(result_name).ok(),
ToPaimonStatus((result_name).status()), \
- PAIMON_STRINGIFY(rexpr));
\
+#define PAIMON_ASSIGN_OR_RAISE_IMPL_FROM_ARROW(result_name, lhs, rexpr)
\
+ auto&& result_name = (rexpr);
\
+ PAIMON_RETURN_IF_(!(result_name).ok(),
::paimon::ToPaimonStatus((result_name).status()), \
+ PAIMON_STRINGIFY(rexpr));
\
lhs = std::move(result_name).ValueUnsafe()
#define PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(lhs, rexpr) \
diff --git
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h
index 1f43468c..c428a079 100644
--- a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h
+++ b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h
@@ -96,6 +96,10 @@ class ApplyDeletionVectorBatchReader : public
FileBatchReader {
return reader_->SupportPreciseBitmapSelection();
}
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
private:
Result<RoaringBitmap32> Filter(int32_t batch_size) const {
RoaringBitmap32 is_valid;
diff --git
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
index 5f4ef074..2e7b29b9 100644
---
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
+++
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
@@ -90,7 +90,7 @@ class ApplyDeletionVectorBatchReaderTest : public
::testing::Test,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- /*enable_io_metrics=*/false, pool, arrow_pool));
+ /*enable_io_metrics=*/false, WarmupLevel::DECODED,
pool, arrow_pool));
} else {
file_batch_reader =
std::make_unique<MockFileBatchReader>(data, target_type_,
batch_size);
@@ -166,6 +166,21 @@ TEST_P(ApplyDeletionVectorBatchReaderTest, TestSimple2) {
CheckResult(data_str, dv_data, "[10, 11, 12, 13, 14, 15, 16]");
}
}
+
+TEST(ApplyDeletionVectorBatchReaderWarmupTest, WarmupForwardsToInnerReader) {
+ auto target_type = arrow::struct_({arrow::field("f1", arrow::int32())});
+ auto mock_reader =
+ std::make_unique<MockFileBatchReader>(/*data=*/nullptr, target_type,
/*batch_size=*/1);
+ auto* inner_reader = mock_reader.get();
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+ auto deletion_vector =
DeletionVector::FromPrimitiveArray(/*is_deleted=*/{}, pool.get());
+ ApplyDeletionVectorBatchReader reader(std::move(mock_reader),
std::move(deletion_vector));
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader.Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
INSTANTIATE_TEST_SUITE_P(EnablePrefetch, ApplyDeletionVectorBatchReaderTest,
::testing::Values(false, true));
} // namespace paimon::test
diff --git a/src/paimon/core/io/complete_row_tracking_fields_reader.h
b/src/paimon/core/io/complete_row_tracking_fields_reader.h
index 21ec2211..894b5a92 100644
--- a/src/paimon/core/io/complete_row_tracking_fields_reader.h
+++ b/src/paimon/core/io/complete_row_tracking_fields_reader.h
@@ -77,6 +77,10 @@ class CompleteRowTrackingFieldsBatchReader : public
FileBatchReader {
return reader_->SupportPreciseBitmapSelection();
}
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
private:
Status ConvertRowTrackingField(int64_t array_length, int64_t init_value,
const
std::function<Result<int64_t>(int32_t)>& convert_func,
diff --git a/src/paimon/core/io/complete_row_tracking_fields_reader_test.cpp
b/src/paimon/core/io/complete_row_tracking_fields_reader_test.cpp
index 806080b1..04f05c30 100644
--- a/src/paimon/core/io/complete_row_tracking_fields_reader_test.cpp
+++ b/src/paimon/core/io/complete_row_tracking_fields_reader_test.cpp
@@ -486,4 +486,18 @@ TEST_F(CompleteRowTrackingFieldsBatchReaderTest,
TestOnlyReadRowTrackingFields)
CheckResult(src_array, /*first_row_id=*/100, /*snapshot_id=*/4,
read_schema, target_array);
}
+TEST_F(CompleteRowTrackingFieldsBatchReaderTest,
TestWarmupForwardsToInnerReader) {
+ auto file_batch_reader = std::make_unique<MockFileBatchReader>(
+ /*data=*/nullptr, arrow::struct_({arrow::field("f0", arrow::int32())}),
+ /*read_batch_size=*/1);
+ auto* inner_reader = file_batch_reader.get();
+ auto reader = std::make_unique<CompleteRowTrackingFieldsBatchReader>(
+ std::move(file_batch_reader), /*first_row_id=*/10, /*snapshot_id=*/1,
+ /*file_field_names=*/std::nullopt, GetArrowPool(pool_));
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader->Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/io/concat_key_value_record_reader.h
b/src/paimon/core/io/concat_key_value_record_reader.h
index 99318a45..b3b6ddc2 100644
--- a/src/paimon/core/io/concat_key_value_record_reader.h
+++ b/src/paimon/core/io/concat_key_value_record_reader.h
@@ -18,6 +18,7 @@
#pragma once
+#include <algorithm>
#include <cstddef>
#include <memory>
#include <utility>
@@ -41,6 +42,11 @@ class ConcatKeyValueRecordReader : public
KeyValueRecordReader {
Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch()
override {
while (current_ < readers_.size()) {
+ // Lookahead: the next files' first reads are paid while this file
is still being
+ // consumed, instead of serially after its EOF. The files here are
read strictly one
+ // after another, and a filtered read of a keyed table yields
about one batch per file,
+ // so without this every file costs a full round trip nobody
overlaps with.
+ WarmupRange(current_, 1 + kWarmupLookahead);
auto& current_reader = readers_[current_];
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<KeyValueRecordReader::Iterator> iterator,
current_reader->NextBatch());
@@ -65,7 +71,31 @@ class ConcatKeyValueRecordReader : public
KeyValueRecordReader {
return MetricsImpl::CollectReadMetrics(readers_);
}
+ /// Forwards to the child a read would touch next, so a Concat nested in
another one is warmed
+ /// through instead of swallowing the call. Only that one child, without
the lookahead: the
+ /// caller warming this reader is not consuming it yet, and the lookahead
is started by the
+ /// first read anyway, so a section's runs do not each hold two warm files
before the merge has
+ /// consumed anything.
+ void Warmup() override {
+ WarmupRange(current_, /*count=*/1);
+ }
+
private:
+ /// How many files ahead of the one being consumed get their first read
started. Each warm file
+ /// holds a prefetch queue of its own, so this trades memory for overlap,
and one file ahead is
+ /// what a strictly sequential consumer can actually use.
+ static constexpr size_t kWarmupLookahead = 1;
+
+ /// Warms up \p count readers starting at \p idx, stopping at the end of
the list. Idempotent,
+ /// so calling it on every batch costs one virtual call plus one pointer
test per already-warm
+ /// reader.
+ void WarmupRange(size_t idx, size_t count) {
+ const size_t end = std::min(idx + count, readers_.size());
+ for (size_t i = idx; i < end; i++) {
+ readers_[i]->Warmup();
+ }
+ }
+
// KeyValue rows may outlive the active child and still reference buffers
allocated by it.
std::vector<std::unique_ptr<KeyValueRecordReader>> readers_;
size_t current_{0};
diff --git a/src/paimon/core/io/concat_key_value_record_reader_test.cpp
b/src/paimon/core/io/concat_key_value_record_reader_test.cpp
index b9da3cfd..345df470 100644
--- a/src/paimon/core/io/concat_key_value_record_reader_test.cpp
+++ b/src/paimon/core/io/concat_key_value_record_reader_test.cpp
@@ -206,4 +206,146 @@ TEST_F(ConcatKeyValueRecordReaderTest, TestEmptyReader) {
CheckResult({src_array2, src_array1}, expected, key_schema, value_schema);
}
+namespace {
+/// Counts Warmup() calls and forwards everything else, so a test can assert
how far ahead the
+/// lookahead reaches.
+class CountingWarmupReader : public KeyValueRecordReader {
+ public:
+ explicit CountingWarmupReader(std::unique_ptr<KeyValueRecordReader>&&
inner)
+ : inner_(std::move(inner)) {}
+
+ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch()
override {
+ return inner_->NextBatch();
+ }
+
+ std::shared_ptr<Metrics> GetReaderMetrics() const override {
+ return inner_->GetReaderMetrics();
+ }
+
+ void Warmup() override {
+ warmups_++;
+ inner_->Warmup();
+ }
+
+ void Close() override {
+ inner_->Close();
+ }
+
+ int warmups() const {
+ return warmups_;
+ }
+
+ private:
+ std::unique_ptr<KeyValueRecordReader> inner_;
+ int warmups_ = 0;
+};
+} // namespace
+
+TEST_F(ConcatKeyValueRecordReaderTest, TestWarmupLooksOneReaderAhead) {
+ arrow::FieldVector fields = {arrow::field("_SEQUENCE_NUMBER",
arrow::int64()),
+ arrow::field("_VALUE_KIND", arrow::int8()),
+ arrow::field("k0", arrow::int32()),
+ arrow::field("k1", arrow::int32()),
+ arrow::field("v0", arrow::int32()),
+ arrow::field("v1", arrow::int32()),
+ arrow::field("v2", arrow::int32())};
+ std::shared_ptr<arrow::Schema> key_schema =
+ arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
+ std::shared_ptr<arrow::Schema> value_schema =
+ arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4],
fields[5], fields[6]}));
+ std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
+ auto make_array = [&src_type](const std::string& json) {
+ return std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(src_type,
json).ValueOrDie());
+ };
+ const std::vector<std::shared_ptr<arrow::StructArray>> src_array_vec = {
+ make_array(R"([[0, 0, 1, 1, 10, 20, 30], [0, 0, 1, 2, 11, 21, 31]])"),
+ make_array(R"([[0, 0, 2, 1, 12, 22, 32], [0, 0, 2, 2, 13, 23, 33]])"),
+ make_array(R"([[0, 0, 3, 1, 14, 24, 34], [0, 0, 3, 2, 15, 25, 35]])")};
+
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+ std::vector<CountingWarmupReader*> counters;
+ // The counters point into what the returned Concat owns, so they are only
valid while it is.
+ auto build = [&](std::vector<CountingWarmupReader*>* out) {
+ out->clear();
+ std::vector<std::unique_ptr<KeyValueRecordReader>> record_readers;
+ for (const auto& src_array : src_array_vec) {
+ auto file_batch_reader = std::make_unique<MockFileBatchReader>(
+ src_array, src_array->type(), /*batch_size=*/100);
+ auto record_reader =
std::make_unique<MockKeyValueDataFileRecordReader>(
+ std::move(file_batch_reader), key_schema, value_schema,
/*level=*/0, pool);
+ auto counting =
std::make_unique<CountingWarmupReader>(std::move(record_reader));
+ out->push_back(counting.get());
+ record_readers.push_back(std::move(counting));
+ }
+ return
std::make_unique<ConcatKeyValueRecordReader>(std::move(record_readers));
+ };
+
+ {
+ std::unique_ptr<ConcatKeyValueRecordReader> concat = build(&counters);
+ // Nothing is started before a read asks for it.
+ EXPECT_EQ(counters[0]->warmups(), 0);
+ EXPECT_EQ(counters[1]->warmups(), 0);
+ EXPECT_EQ(counters[2]->warmups(), 0);
+
+ // The first batch warms the file it reads and the one after it, and
no further: a deeper
+ // lookahead would hold more files open at once than the prefetch
executor has threads for.
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<KeyValueRecordReader::Iterator>
iterator,
+ concat->NextBatch());
+ ASSERT_NE(iterator, nullptr);
+ EXPECT_EQ(counters[0]->warmups(), 1);
+ EXPECT_EQ(counters[1]->warmups(), 1);
+ EXPECT_EQ(counters[2]->warmups(), 0);
+
+ // Rows may reference buffers the child allocated, so drop them before
closing it.
+ iterator.reset();
+ // A caller that only warms this Concat is not consuming it yet, so
the call reaches the
+ // child a read would touch next and stops there: it must not add the
lookahead on top of
+ // what the read above already started.
+ concat->Warmup();
+ EXPECT_EQ(counters[0]->warmups(), 2);
+ EXPECT_EQ(counters[1]->warmups(), 1);
+ EXPECT_EQ(counters[2]->warmups(), 0);
+ // Idempotent: warming an already-warm reader is not an error.
+ concat->Warmup();
+ concat->Close();
+ }
+
+ {
+ // Warmup() before the first read reaches only the first file, so a
merge that warms k runs
+ // starts k files and not 2k.
+ std::unique_ptr<ConcatKeyValueRecordReader> concat = build(&counters);
+ concat->Warmup();
+ EXPECT_EQ(counters[0]->warmups(), 1);
+ EXPECT_EQ(counters[1]->warmups(), 0);
+ EXPECT_EQ(counters[2]->warmups(), 0);
+ concat->Close();
+ }
+
+ {
+ std::unique_ptr<ConcatKeyValueRecordReader> concat = build(&counters);
+ ASSERT_OK_AND_ASSIGN(
+ auto results,
+ (ReadResultCollector::CollectKeyValueResult<
+ ConcatKeyValueRecordReader,
KeyValueRecordReader::Iterator>(concat.get())));
+ std::vector<KeyValue> expected = KeyValueChecker::GenerateKeyValues(
+ /*seq_vec=*/{0, 0, 0, 0, 0, 0},
+ /*key_vec=*/{{1, 1}, {1, 2}, {2, 1}, {2, 2}, {3, 1}, {3, 2}},
+ /*value_vec=*/
+ {{1, 1, 10, 20, 30},
+ {1, 2, 11, 21, 31},
+ {2, 1, 12, 22, 32},
+ {2, 2, 13, 23, 33},
+ {3, 1, 14, 24, 34},
+ {3, 2, 15, 25, 35}},
+ pool);
+ // Warming up changes when a file is opened, never what it reads.
+ KeyValueChecker::CheckResult(expected, results,
key_schema->num_fields(),
+ value_schema->num_fields());
+ for (const CountingWarmupReader* counter : counters) {
+ EXPECT_GE(counter->warmups(), 1);
+ }
+ }
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/io/field_mapping_reader.h
b/src/paimon/core/io/field_mapping_reader.h
index c0057cf2..fa976b2f 100644
--- a/src/paimon/core/io/field_mapping_reader.h
+++ b/src/paimon/core/io/field_mapping_reader.h
@@ -91,6 +91,12 @@ class FieldMappingReader : public FileBatchReader {
return reader_->SupportPreciseBitmapSelection();
}
+ /// This is the outermost wrapper of every data file's reader stack, so
swallowing the call here
+ /// would leave the whole stack cold no matter who asked for the warmup.
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
private:
FieldMappingReader(int32_t field_count, std::unique_ptr<FileBatchReader>&&
reader,
const BinaryRow& partition,
std::unique_ptr<FieldMapping>&& mapping,
diff --git a/src/paimon/core/io/field_mapping_reader_test.cpp
b/src/paimon/core/io/field_mapping_reader_test.cpp
index 3e73294f..3756475f 100644
--- a/src/paimon/core/io/field_mapping_reader_test.cpp
+++ b/src/paimon/core/io/field_mapping_reader_test.cpp
@@ -916,4 +916,27 @@ TEST_F(FieldMappingReaderTest,
TestCreateFailFastOnInvalidMapSelectedKeysMetadat
/*skip_map_selected_keys_filter_field_ids=*/{},
GetArrowPool(pool_)),
"Duplicate selected key 'a'");
}
+
+TEST_F(FieldMappingReaderTest, TestWarmupForwardsToInnerReader) {
+ std::vector<DataField> data_fields = {DataField(0, arrow::field("a",
arrow::int32()))};
+ auto data_schema = DataField::ConvertDataFieldsToArrowSchema(data_fields);
+ ASSERT_OK_AND_ASSIGN(auto mapping_builder,
+ FieldMappingBuilder::Create(data_schema,
/*partition_keys=*/{},
+ /*predicate=*/nullptr));
+ ASSERT_OK_AND_ASSIGN(auto mapping,
mapping_builder->CreateFieldMapping(data_fields));
+ auto mock = std::make_unique<MockFileBatchReader>(
+ /*data=*/nullptr, arrow::struct_(data_schema->fields()),
/*read_batch_size=*/1);
+ auto* inner_reader = mock.get();
+
+ ASSERT_OK_AND_ASSIGN(auto reader,
+ FieldMappingReader::Create(data_schema->num_fields(),
std::move(mock),
+ BinaryRow::EmptyRow(),
std::move(mapping),
+
/*skip_map_selected_keys_filter_field_ids=*/{},
+ GetArrowPool(pool_)));
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader->Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/io/key_value_data_file_record_reader.cpp
b/src/paimon/core/io/key_value_data_file_record_reader.cpp
index 49a216e9..af6e3f32 100644
--- a/src/paimon/core/io/key_value_data_file_record_reader.cpp
+++ b/src/paimon/core/io/key_value_data_file_record_reader.cpp
@@ -162,6 +162,15 @@ Result<std::unique_ptr<KeyValueRecordReader::Iterator>>
KeyValueDataFileRecordRe
return std::make_unique<KeyValueDataFileRecordReader::Iterator>(this);
}
+void KeyValueDataFileRecordReader::Warmup() {
+ // reader_ is typed as BatchReader, but Warmup() lives on FileBatchReader.
This is the single
+ // boundary where the two hierarchies meet; every FileBatchReader
decorator below here forwards
+ // Warmup() through its own FileBatchReader-typed inner reader without any
cast.
+ if (auto* file_reader = dynamic_cast<FileBatchReader*>(reader_.get())) {
+ file_reader->Warmup();
+ }
+}
+
void KeyValueDataFileRecordReader::Reset() {
file_reader_ = nullptr;
selection_bitmap_ = RoaringBitmap32();
diff --git a/src/paimon/core/io/key_value_data_file_record_reader.h
b/src/paimon/core/io/key_value_data_file_record_reader.h
index d968dd14..1216c034 100644
--- a/src/paimon/core/io/key_value_data_file_record_reader.h
+++ b/src/paimon/core/io/key_value_data_file_record_reader.h
@@ -73,6 +73,8 @@ class KeyValueDataFileRecordReader : public
KeyValueRecordReader {
return reader_->GetReaderMetrics();
}
+ void Warmup() override;
+
void Close() override {
Reset();
reader_->Close();
diff --git a/src/paimon/core/io/key_value_data_file_record_reader_test.cpp
b/src/paimon/core/io/key_value_data_file_record_reader_test.cpp
index d2077759..84689f23 100644
--- a/src/paimon/core/io/key_value_data_file_record_reader_test.cpp
+++ b/src/paimon/core/io/key_value_data_file_record_reader_test.cpp
@@ -456,4 +456,19 @@ TEST_F(KeyValueDataFileRecordReaderTest,
TestKeyFieldAfterValueField) {
}
}
+TEST_F(KeyValueDataFileRecordReaderTest, TestWarmupForwardsToInnerReader) {
+ auto file_batch_reader = std::make_unique<MockFileBatchReader>(
+ /*data=*/nullptr, arrow::struct_({arrow::field("_SEQUENCE_NUMBER",
arrow::int64())}),
+ /*read_batch_size=*/1);
+ auto* inner_reader = file_batch_reader.get();
+ auto key_schema = arrow::schema({arrow::field("k0", arrow::int32())});
+ auto value_schema = arrow::schema({arrow::field("v0", arrow::int32())});
+ KeyValueDataFileRecordReader reader(std::move(file_batch_reader),
key_schema, value_schema,
+ /*level=*/0, pool_);
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader.Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/io/key_value_in_memory_record_reader.h
b/src/paimon/core/io/key_value_in_memory_record_reader.h
index 8ffd0c7b..9e5d35e2 100644
--- a/src/paimon/core/io/key_value_in_memory_record_reader.h
+++ b/src/paimon/core/io/key_value_in_memory_record_reader.h
@@ -73,6 +73,9 @@ class KeyValueInMemoryRecordReader : public
KeyValueRecordReader {
return std::make_shared<MetricsImpl>();
}
+ /// Nothing to warm up: the records are already in memory and no file is
opened by a read.
+ void Warmup() override {}
+
void Close() override;
private:
diff --git a/src/paimon/core/io/key_value_record_reader.h
b/src/paimon/core/io/key_value_record_reader.h
index efd2f3b4..4e452e0d 100644
--- a/src/paimon/core/io/key_value_record_reader.h
+++ b/src/paimon/core/io/key_value_record_reader.h
@@ -39,6 +39,17 @@ 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. Reports no error, like `FileBatchReader::Warmup()`: a
hint about a file
+ /// nobody reads must not fail the read in progress.
+ ///
+ /// Pure virtual although a reader with nothing to start has nothing to do
here: a wrapping
+ /// reader that forgot to forward the call would silently leave the file
readers below it cold,
+ /// which no test of the read result can catch, so every implementation is
made to say whether
+ /// it forwards or is a no-op.
+ virtual void Warmup() = 0;
+
virtual void Close() = 0;
};
} // namespace paimon
diff --git a/src/paimon/core/io/merged_key_value_record_reader.h
b/src/paimon/core/io/merged_key_value_record_reader.h
index a1b7aa5e..45dff2f9 100644
--- a/src/paimon/core/io/merged_key_value_record_reader.h
+++ b/src/paimon/core/io/merged_key_value_record_reader.h
@@ -63,6 +63,10 @@ class MergedKeyValueRecordReader : public
KeyValueRecordReader {
std::shared_ptr<Metrics> GetReaderMetrics() const override;
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
void Close() override;
private:
diff --git a/src/paimon/core/io/vector_file_batch_reader.h
b/src/paimon/core/io/vector_file_batch_reader.h
index 29b064d8..3e7dc418 100644
--- a/src/paimon/core/io/vector_file_batch_reader.h
+++ b/src/paimon/core/io/vector_file_batch_reader.h
@@ -74,6 +74,10 @@ class VectorFileBatchReader : public FileBatchReader {
return reader_->SupportPreciseBitmapSelection();
}
+ void Warmup() override {
+ reader_->Warmup();
+ }
+
private:
Result<ReadBatch> ConvertBatch(ReadBatch&& batch) const;
diff --git a/src/paimon/core/io/vector_file_batch_reader_test.cpp
b/src/paimon/core/io/vector_file_batch_reader_test.cpp
index 560ea5b2..007b35e1 100644
--- a/src/paimon/core/io/vector_file_batch_reader_test.cpp
+++ b/src/paimon/core/io/vector_file_batch_reader_test.cpp
@@ -83,6 +83,18 @@ TEST(VectorFileBatchReaderTest, ConvertSchemaAndNextBatch) {
ASSERT_TRUE(BatchReader::IsEofBatch(batch));
}
+TEST(VectorFileBatchReaderTest, WarmupForwardsToInnerReader) {
+ auto physical_type = AsStructType(arrow::struct_({arrow::field("id",
arrow::int32())}));
+ auto mock_reader =
+ std::make_unique<MockFileBatchReader>(/*data=*/nullptr, physical_type,
/*batch_size=*/1);
+ MockFileBatchReader* inner_reader = mock_reader.get();
+ VectorFileBatchReader reader(std::move(mock_reader),
GetArrowPool(GetDefaultPool()));
+
+ ASSERT_EQ(0, inner_reader->GetWarmupCount());
+ reader.Warmup();
+ ASSERT_EQ(1, inner_reader->GetWarmupCount());
+}
+
TEST(VectorFileBatchReaderTest, KeepFixedSizeListFileSchema) {
auto logical_type = AsStructType(arrow::struct_({
arrow::field("id", arrow::int32()),
diff --git a/src/paimon/core/mergetree/compact/loser_tree.cpp
b/src/paimon/core/mergetree/compact/loser_tree.cpp
index e9321190..ad48903a 100644
--- a/src/paimon/core/mergetree/compact/loser_tree.cpp
+++ b/src/paimon/core/mergetree/compact/loser_tree.cpp
@@ -36,8 +36,23 @@
LoserTree::LoserTree(std::vector<std::unique_ptr<KeyValueRecordReader>>&& reader
}
}
+void LoserTree::WarmupLeaves() {
+ // Warmed in the same order the loop below consumes them, so the leaf it
blocks on first is the
+ // one whose read was started first. Each leaf only starts the file it is
about to read: a
+ // deeper lookahead here would put two warm files per run in flight before
the merge has
+ // consumed anything, and the run itself starts its next file once it is
being read.
+ for (int32_t i = size_ - 1; i >= 0; i--) {
+ leaves_[i].reader->Warmup();
+ }
+}
+
Status LoserTree::InitializeIfNeeded() {
if (!initialized_) {
+ // Every leaf's first read below blocks on its own file, and the
leaves of a section are
+ // independent sorted runs, so without warming them the section pays
those read latencies
+ // one after another. Warming starts them together, and the loop then
collects reads that
+ // are already in flight.
+ WarmupLeaves();
std::fill(tree_.begin(), tree_.end(), -1);
for (int32_t i = size_ - 1; i >= 0; i--) {
PAIMON_RETURN_NOT_OK(leaves_[i].AdvanceIfAvailable());
diff --git a/src/paimon/core/mergetree/compact/loser_tree.h
b/src/paimon/core/mergetree/compact/loser_tree.h
index 50b3bab5..65425848 100644
--- a/src/paimon/core/mergetree/compact/loser_tree.h
+++ b/src/paimon/core/mergetree/compact/loser_tree.h
@@ -154,6 +154,10 @@ class LoserTree {
};
private:
+ /// Starts the first read of the leaves InitializeIfNeeded is about to
block on, so the sorted
+ /// runs of a section overlap their read latencies instead of serializing
them.
+ void WarmupLeaves();
+
int32_t size_;
bool initialized_;
// KeyValue rows may be consumed asynchronously and still reference
buffers allocated by the
diff --git a/src/paimon/core/mergetree/compact/sort_merge_reader_test.cpp
b/src/paimon/core/mergetree/compact/sort_merge_reader_test.cpp
index dabee55c..41db77ba 100644
--- a/src/paimon/core/mergetree/compact/sort_merge_reader_test.cpp
+++ b/src/paimon/core/mergetree/compact/sort_merge_reader_test.cpp
@@ -30,6 +30,7 @@
#include "arrow/array/array_nested.h"
#include "arrow/ipc/json_simple.h"
#include "gtest/gtest.h"
+#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/fields_comparator.h"
#include "paimon/core/core_options.h"
@@ -39,6 +40,7 @@
#include "paimon/core/key_value.h"
#include "paimon/core/mergetree/compact/aggregate/aggregate_merge_function.h"
#include "paimon/core/mergetree/compact/deduplicate_merge_function.h"
+#include "paimon/core/mergetree/compact/loser_tree.h"
#include "paimon/core/mergetree/compact/reducer_merge_function_wrapper.h"
#include "paimon/core/mergetree/compact/sort_merge_reader_with_loser_tree.h"
#include "paimon/core/mergetree/compact/sort_merge_reader_with_min_heap.h"
@@ -900,4 +902,72 @@ TEST_F(SortMergeReaderTest, TestRawSortNoMergeWithMinHeap)
{
value_schema, expected,
/*need_merge=*/false);
}
+namespace {
+/// Records the order in which a loser tree touches its leaves, so a test can
tell a tree that
+/// warms every leaf up front from one that warms each leaf only as it reaches
it. NextBatch()
+/// returns no iterator, which puts every leaf at EOF immediately and keeps
InitializeIfNeeded()
+/// the only thing under test.
+class LeafTouchRecordingReader : public KeyValueRecordReader {
+ public:
+ LeafTouchRecordingReader(int32_t leaf_index, std::vector<std::string>*
events)
+ : leaf_index_(leaf_index), events_(events) {}
+
+ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch()
override {
+ events_->push_back("read" + std::to_string(leaf_index_));
+ return std::unique_ptr<KeyValueRecordReader::Iterator>();
+ }
+
+ std::shared_ptr<Metrics> GetReaderMetrics() const override {
+ return std::make_shared<MetricsImpl>();
+ }
+
+ void Warmup() override {
+ events_->push_back("warm" + std::to_string(leaf_index_));
+ }
+
+ void Close() override {}
+
+ private:
+ int32_t leaf_index_;
+ std::vector<std::string>* events_;
+};
+} // namespace
+
+// Every leaf of a section is an independent sorted run whose first read
blocks on its own file, so
+// those latencies only overlap when all leaves are warmed before the first
one is advanced. The
+// merge then collects reads that are already in flight instead of paying them
one after another.
+TEST_F(SortMergeReaderTest, TestLoserTreeWarmsAllLeavesBeforeAdvancingAny) {
+ std::vector<std::string> events;
+ std::vector<std::unique_ptr<KeyValueRecordReader>> readers;
+ constexpr int32_t kLeafCount = 3;
+ for (int32_t i = 0; i < kLeafCount; i++) {
+ readers.push_back(std::make_unique<LeafTouchRecordingReader>(i,
&events));
+ }
+ // An exhausted leaf compares as the smallest, which is how the merge
reader's own comparators
+ // treat a run that has nothing left.
+ auto exhausted_last_comparator = [](const std::optional<KeyValue>& lhs,
+ const std::optional<KeyValue>& rhs) ->
int32_t {
+ if (lhs == std::nullopt) {
+ return -1;
+ }
+ if (rhs == std::nullopt) {
+ return 1;
+ }
+ return 0;
+ };
+ LoserTree loser_tree(std::move(readers), exhausted_last_comparator,
exhausted_last_comparator);
+
+ ASSERT_OK(loser_tree.InitializeIfNeeded());
+
+ // Warmup visits the leaves in the order the advancing loop consumes them,
so the leaf the tree
+ // blocks on first is the one whose read was started first.
+ const std::vector<std::string> expected_events = {"warm2", "warm1",
"warm0",
+ "read2", "read1",
"read0"};
+ ASSERT_EQ(expected_events, events);
+
+ // Initialization is one-shot: a second call must neither warm nor read
again.
+ ASSERT_OK(loser_tree.InitializeIfNeeded());
+ ASSERT_EQ(expected_events, events);
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp
b/src/paimon/core/mergetree/merge_tree_writer_test.cpp
index f4cbaeaf..8a152e2f 100644
--- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp
+++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp
@@ -85,6 +85,10 @@ class TrackingKeyValueRecordReader : public
KeyValueRecordReader {
return inner_reader_->GetReaderMetrics();
}
+ void Warmup() override {
+ inner_reader_->Warmup();
+ }
+
void Close() override {
if (closed_flag_ != nullptr) {
*closed_flag_ = true;
diff --git a/src/paimon/core/mergetree/spill_reader.h
b/src/paimon/core/mergetree/spill_reader.h
index 6dba6fd7..af05d271 100644
--- a/src/paimon/core/mergetree/spill_reader.h
+++ b/src/paimon/core/mergetree/spill_reader.h
@@ -61,6 +61,8 @@ class SpillReader : public KeyValueRecordReader {
Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch()
override;
std::shared_ptr<Metrics> GetReaderMetrics() const override;
+ /// Nothing to warm up: the spill file is read directly, without a
prefetching reader below.
+ void Warmup() override {}
void Close() override;
private:
diff --git a/src/paimon/core/operation/abstract_split_read.cpp
b/src/paimon/core/operation/abstract_split_read.cpp
index 0a33c7e7..f1540843 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -138,6 +138,7 @@ DataFileReadOptions
AbstractSplitRead::BuildDataFileReadOptions() const {
read_options.read_ahead_cache_enabled = context_->ReadAheadCacheEnabled();
read_options.cache_config = context_->GetCacheConfig();
read_options.prefetch_io_metrics_enabled =
options_.PrefetchIoMetricsEnabled();
+ read_options.warmup_level = context_->GetWarmupLevel();
return read_options;
}
diff --git a/src/paimon/core/operation/internal_read_context.h
b/src/paimon/core/operation/internal_read_context.h
index cadef31c..ddbbb49f 100644
--- a/src/paimon/core/operation/internal_read_context.h
+++ b/src/paimon/core/operation/internal_read_context.h
@@ -111,6 +111,10 @@ class InternalReadContext {
return read_context_->GetCacheConfig();
}
+ WarmupLevel GetWarmupLevel() const {
+ return read_context_->GetWarmupLevel();
+ }
+
/// Create a new InternalReadContext with a different read schema.
/// Useful for creating a context with a minimal column set for
specialized reads.
/// All other settings (predicate, options, table_schema, etc.) are
inherited
diff --git a/src/paimon/core/operation/merge_file_split_read.cpp
b/src/paimon/core/operation/merge_file_split_read.cpp
index 570059ab..746d4d3a 100644
--- a/src/paimon/core/operation/merge_file_split_read.cpp
+++ b/src/paimon/core/operation/merge_file_split_read.cpp
@@ -116,6 +116,10 @@ class SortMergeKeyValueRecordReader : public
KeyValueRecordReader {
reader_->Close();
}
+ /// A no-op: SortMergeReader has no warmup verb to forward to, so a
section is warmed from
+ /// inside its own merge (LoserTree warms its leaves) and not from the
section before it.
+ void Warmup() override {}
+
std::shared_ptr<Metrics> GetReaderMetrics() const override {
return reader_->GetReaderMetrics();
}
diff --git a/src/paimon/core/operation/read_context.cpp
b/src/paimon/core/operation/read_context.cpp
index 807a507b..7545c543 100644
--- a/src/paimon/core/operation/read_context.cpp
+++ b/src/paimon/core/operation/read_context.cpp
@@ -45,7 +45,7 @@ ReadContext::ReadContext(
const std::shared_ptr<RealtimeContext>& realtime_context,
const std::map<std::string, std::string>& options, bool
read_ahead_cache_enabled,
const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
- const std::shared_ptr<FormatTable>& format_table)
+ const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level)
: path_(path),
branch_(branch),
read_field_names_(read_field_names),
@@ -68,7 +68,8 @@ ReadContext::ReadContext(
read_ahead_cache_enabled_(read_ahead_cache_enabled),
cache_config_(cache_config),
cache_(cache),
- format_table_(format_table) {}
+ format_table_(format_table),
+ warmup_level_(warmup_level) {}
ReadContext::~ReadContext() {
if (read_schema_ && read_schema_->release) {
@@ -114,6 +115,7 @@ class ReadContextBuilder::Impl {
realtime_context_.reset();
cache_config_ = CacheConfig();
cache_.reset();
+ warmup_level_ = WarmupLevel::DECODED;
}
private:
@@ -144,6 +146,7 @@ class ReadContextBuilder::Impl {
bool read_ahead_cache_enabled_ = true;
CacheConfig cache_config_;
std::shared_ptr<Cache> cache_;
+ WarmupLevel warmup_level_ = WarmupLevel::DECODED;
};
ReadContextBuilder::ReadContextBuilder(const std::string& path)
@@ -279,6 +282,11 @@ ReadContextBuilder&
ReadContextBuilder::SetReadAheadCacheEnabled(bool enabled) {
return *this;
}
+ReadContextBuilder& ReadContextBuilder::SetWarmupLevel(WarmupLevel level) {
+ impl_->warmup_level_ = level;
+ return *this;
+}
+
ReadContextBuilder& ReadContextBuilder::WithCacheConfig(const CacheConfig&
cache_config) {
impl_->cache_config_ = cache_config;
return *this;
@@ -350,7 +358,7 @@ Result<std::unique_ptr<ReadContext>>
ReadContextBuilder::Finish() {
impl_->row_to_batch_thread_number_, impl_->table_schema_,
impl_->memory_pool_,
impl_->executor_, impl_->specific_file_system_,
impl_->fs_scheme_to_identifier_map_,
impl_->realtime_context_, impl_->options_,
impl_->read_ahead_cache_enabled_,
- impl_->cache_config_, impl_->cache_, impl_->format_table_);
+ impl_->cache_config_, impl_->cache_, impl_->format_table_,
impl_->warmup_level_);
if (impl_->read_schema_ && impl_->read_schema_->release) {
ctx->SetReadSchema(std::move(impl_->read_schema_));
}
diff --git a/src/paimon/core/operation/read_context_test.cpp
b/src/paimon/core/operation/read_context_test.cpp
index 3d362352..572fd5a5 100644
--- a/src/paimon/core/operation/read_context_test.cpp
+++ b/src/paimon/core/operation/read_context_test.cpp
@@ -46,6 +46,7 @@ TEST(ReadContextTest, TestDefaultValue) {
ASSERT_FALSE(ctx->EnablePredicateFilter());
ASSERT_FALSE(ctx->EnablePrefetch());
ASSERT_TRUE(ctx->ReadAheadCacheEnabled());
+ ASSERT_EQ(WarmupLevel::DECODED, ctx->GetWarmupLevel());
ASSERT_EQ(600, ctx->GetPrefetchBatchCount());
ASSERT_EQ(3, ctx->GetPrefetchMaxParallelNum());
ASSERT_FALSE(ctx->EnableMultiThreadRowToBatch());
@@ -73,6 +74,7 @@ TEST(ReadContextTest, TestSetContent) {
builder.EnablePredicateFilter(true);
builder.EnablePrefetch(true);
builder.SetReadAheadCacheEnabled(false);
+ builder.SetWarmupLevel(WarmupLevel::RAW);
builder.SetPrefetchBatchCount(1200);
builder.SetPrefetchMaxParallelNum(6);
builder.EnableMultiThreadRowToBatch(true);
@@ -98,6 +100,7 @@ TEST(ReadContextTest, TestSetContent) {
ASSERT_TRUE(ctx->EnablePredicateFilter());
ASSERT_TRUE(ctx->EnablePrefetch());
ASSERT_FALSE(ctx->ReadAheadCacheEnabled());
+ ASSERT_EQ(WarmupLevel::RAW, ctx->GetWarmupLevel());
ASSERT_EQ(1200, ctx->GetPrefetchBatchCount());
ASSERT_EQ(6, ctx->GetPrefetchMaxParallelNum());
ASSERT_TRUE(ctx->EnableMultiThreadRowToBatch());
@@ -117,6 +120,24 @@ TEST(ReadContextTest, TestSetContent) {
ASSERT_TRUE(ctx->GetCache());
}
+TEST(ReadContextTest, TestSetWarmupLevel) {
+ for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW,
WarmupLevel::DECODED}) {
+ ReadContextBuilder builder("table_root_path");
+ // The setter hands back the builder so it chains like every other
setter on it.
+ ASSERT_EQ(&builder, &builder.SetWarmupLevel(level));
+ ASSERT_OK_AND_ASSIGN(auto ctx, builder.Finish());
+ ASSERT_EQ(level, ctx->GetWarmupLevel());
+ }
+
+ // Finish() resets the builder, so reusing one must not carry the previous
warmup level over.
+ ReadContextBuilder builder("table_root_path");
+ builder.SetWarmupLevel(WarmupLevel::NONE);
+ ASSERT_OK_AND_ASSIGN(auto first_ctx, builder.Finish());
+ ASSERT_EQ(WarmupLevel::NONE, first_ctx->GetWarmupLevel());
+ ASSERT_OK_AND_ASSIGN(auto second_ctx, builder.Finish());
+ ASSERT_EQ(WarmupLevel::DECODED, second_ctx->GetWarmupLevel());
+}
+
TEST(ReadContextTest, TestSetOptionsOverridesAddedOptions) {
ReadContextBuilder builder("table_root_path");
builder.AddOption("old", "value");
diff --git a/src/paimon/core/table/format/format_table_read.cpp
b/src/paimon/core/table/format/format_table_read.cpp
index 86b1a880..cda8cbfe 100644
--- a/src/paimon/core/table/format/format_table_read.cpp
+++ b/src/paimon/core/table/format/format_table_read.cpp
@@ -337,6 +337,7 @@ Result<std::unique_ptr<FormatTableRead>>
FormatTableRead::Create(
read_options.prefetch_batch_count = read_context->GetPrefetchBatchCount();
read_options.read_ahead_cache_enabled =
read_context->ReadAheadCacheEnabled();
read_options.cache_config = read_context->GetCacheConfig();
+ read_options.warmup_level = read_context->GetWarmupLevel();
return CreateInternal(table, projection, read_context->GetMemoryPool(),
read_context->GetPredicate(),
read_context->EnablePredicateFilter(),
diff --git a/src/paimon/core/table/system/audit_log_system_table.cpp
b/src/paimon/core/table/system/audit_log_system_table.cpp
index 3b5f1c82..8de9742e 100644
--- a/src/paimon/core/table/system/audit_log_system_table.cpp
+++ b/src/paimon/core/table/system/audit_log_system_table.cpp
@@ -425,11 +425,31 @@ Result<std::unique_ptr<TableRead>>
AuditLogSystemTable::NewChangelogRead(
return Status::NotImplemented(Name(), " system table predicate
pushdown is not supported");
}
- ReadContextBuilder builder(table_path_);
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> base_read_schema,
BaseReadSchema());
using StringMap = std::map<std::string, std::string>;
PAIMON_ASSIGN_OR_RAISE(StringMap read_options, ReadOptions());
PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options,
CoreOptions::FromMap(read_options));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> data_context,
+ CreateDataReadContext(context, read_options));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableRead> data_read,
+ TableRead::Create(std::move(data_context)));
+ auto* key_value_read = dynamic_cast<KeyValueTableRead*>(data_read.get());
+ if (!key_value_read) {
+ return Status::Invalid("audit_log system table requires key-value
table read");
+ }
+ key_value_read->ForceKeepDelete(true);
+ bool include_sequence_number =
core_options.TableReadSequenceNumberEnabled();
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> output_schema,
ArrowSchema());
+ return std::make_unique<ChangelogTableRead>(std::move(data_read),
std::move(output_schema),
+ include_sequence_number,
std::move(converter),
+ context->GetMemoryPool());
+}
+
+Result<std::unique_ptr<ReadContext>>
AuditLogSystemTable::CreateDataReadContext(
+ const std::shared_ptr<ReadContext>& context,
+ const std::map<std::string, std::string>& read_options) const {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> base_read_schema,
BaseReadSchema());
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options,
CoreOptions::FromMap(read_options));
+ ReadContextBuilder builder(table_path_);
builder.SetOptions(read_options)
.SetReadFieldNames(base_read_schema->field_names())
.WithBranch(core_options.GetBranch())
@@ -444,21 +464,9 @@ Result<std::unique_ptr<TableRead>>
AuditLogSystemTable::NewChangelogRead(
.SetRowToBatchThreadNumber(context->GetRowToBatchThreadNumber())
.SetReadAheadCacheEnabled(context->ReadAheadCacheEnabled())
.WithCacheConfig(context->GetCacheConfig())
+ .SetWarmupLevel(context->GetWarmupLevel())
.WithCache(context->GetCache());
-
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> data_context,
builder.Finish());
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableRead> data_read,
- TableRead::Create(std::move(data_context)));
- auto* key_value_read = dynamic_cast<KeyValueTableRead*>(data_read.get());
- if (!key_value_read) {
- return Status::Invalid("audit_log system table requires key-value
table read");
- }
- key_value_read->ForceKeepDelete(true);
- bool include_sequence_number =
core_options.TableReadSequenceNumberEnabled();
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> output_schema,
ArrowSchema());
- return std::make_unique<ChangelogTableRead>(std::move(data_read),
std::move(output_schema),
- include_sequence_number,
std::move(converter),
- context->GetMemoryPool());
+ return builder.Finish();
}
Result<std::shared_ptr<arrow::Schema>> AuditLogSystemTable::BaseReadSchema()
const {
diff --git a/src/paimon/core/table/system/audit_log_system_table.h
b/src/paimon/core/table/system/audit_log_system_table.h
index e21e287c..780f5fa7 100644
--- a/src/paimon/core/table/system/audit_log_system_table.h
+++ b/src/paimon/core/table/system/audit_log_system_table.h
@@ -69,6 +69,14 @@ class AuditLogSystemTable : public SystemTable {
Result<std::unique_ptr<TableRead>> NewChangelogRead(
const std::shared_ptr<ReadContext>& context,
std::shared_ptr<const ChangelogBatchConverter> converter) const;
+ /// Build the context for the data table underneath this changelog view.
+ ///
+ /// The builder starts from the defaults, so every setting the caller
configured has to be
+ /// copied across explicitly: one that is not copied silently reverts to
its default for
+ /// `$audit_log` and `$binlog`, and neither end reports it.
+ Result<std::unique_ptr<ReadContext>> CreateDataReadContext(
+ const std::shared_ptr<ReadContext>& context,
+ const std::map<std::string, std::string>& read_options) const;
Result<std::shared_ptr<arrow::Schema>> BaseReadSchema() const;
Result<std::map<std::string, std::string>> ReadOptions() const;
diff --git a/src/paimon/core/table/system/read_optimized_system_table.cpp
b/src/paimon/core/table/system/read_optimized_system_table.cpp
index d7bfa091..8d569644 100644
--- a/src/paimon/core/table/system/read_optimized_system_table.cpp
+++ b/src/paimon/core/table/system/read_optimized_system_table.cpp
@@ -90,7 +90,7 @@ Result<std::unique_ptr<TableScan>>
ReadOptimizedSystemTable::NewScan(
return TableScan::Create(std::move(base_context));
}
-Result<std::unique_ptr<TableRead>> ReadOptimizedSystemTable::NewRead(
+Result<std::unique_ptr<ReadContext>>
ReadOptimizedSystemTable::CreateDataReadContext(
const std::shared_ptr<ReadContext>& context) const {
auto options = options_;
std::string branch = context->GetBranch();
@@ -115,6 +115,7 @@ Result<std::unique_ptr<TableRead>>
ReadOptimizedSystemTable::NewRead(
.WithFileSystemSchemeToIdentifierMap(context->GetFileSystemSchemeToIdentifierMap())
.SetReadAheadCacheEnabled(context->ReadAheadCacheEnabled())
.WithCacheConfig(context->GetCacheConfig())
+ .SetWarmupLevel(context->GetWarmupLevel())
.WithCache(context->GetCache())
.SetReadFieldNames(context->GetReadFieldNames())
.SetReadFieldIds(context->GetReadFieldIds());
@@ -128,7 +129,13 @@ Result<std::unique_ptr<TableRead>>
ReadOptimizedSystemTable::NewRead(
if (context->GetSpecificTableSchema().has_value()) {
builder.SetTableSchema(context->GetSpecificTableSchema().value());
}
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> base_context,
builder.Finish());
+ return builder.Finish();
+}
+
+Result<std::unique_ptr<TableRead>> ReadOptimizedSystemTable::NewRead(
+ const std::shared_ptr<ReadContext>& context) const {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReadContext> base_context,
+ CreateDataReadContext(context));
return TableRead::Create(std::move(base_context));
}
diff --git a/src/paimon/core/table/system/read_optimized_system_table.h
b/src/paimon/core/table/system/read_optimized_system_table.h
index 14fadbfd..3123fe03 100644
--- a/src/paimon/core/table/system/read_optimized_system_table.h
+++ b/src/paimon/core/table/system/read_optimized_system_table.h
@@ -44,6 +44,14 @@ class ReadOptimizedSystemTable : public SystemTable {
const std::shared_ptr<ReadContext>& context) const override;
private:
+ /// Build the context for the data table underneath this read-optimized
view.
+ ///
+ /// The builder starts from the defaults, so every setting the caller
configured has to be
+ /// copied across explicitly: one that is not copied silently reverts to
its default for `$ro`,
+ /// and neither end reports it.
+ Result<std::unique_ptr<ReadContext>> CreateDataReadContext(
+ const std::shared_ptr<ReadContext>& context) const;
+
std::map<std::string, std::string> ReadOptimizedOptions() const;
std::string table_path_;
diff --git a/src/paimon/core/table/system/system_table_test.cpp
b/src/paimon/core/table/system/system_table_test.cpp
index 41ac2d8b..00a538f3 100644
--- a/src/paimon/core/table/system/system_table_test.cpp
+++ b/src/paimon/core/table/system/system_table_test.cpp
@@ -40,6 +40,7 @@
#include "paimon/fs/file_system_factory.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/metrics.h"
+#include "paimon/read_context.h"
#include "paimon/reader/batch_reader.h"
#include "paimon/result.h"
#include "paimon/status.h"
@@ -231,6 +232,49 @@ TEST(SystemTableTest,
TestReadOptimizedSystemTablePathParsing) {
ASSERT_EQ(parsed->system_table_name, ReadOptimizedSystemTable::kName);
}
+// A system table builds a fresh ReadContext for the data table underneath it,
starting from the
+// defaults, so a setting that is not copied across silently reverts: a caller
that asked for NONE
+// or RAW would get DECODED back. `$ro` has its own builder chain and
`$audit_log` and `$binlog`
+// share one, so all three are pinned here.
+TEST(SystemTableTest, TestNewReadPropagatesWarmupLevel) {
+ std::map<std::string, std::string> options = {{Options::FILE_SYSTEM,
"local"},
+ {Options::FILE_FORMAT,
"orc"}};
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+ CreateTableSchemaForTest(options));
+
+ AuditLogSystemTable audit_log(/*fs=*/nullptr, "/tmp/table", table_schema,
options);
+ BinlogSystemTable binlog(/*fs=*/nullptr, "/tmp/table", table_schema,
options);
+ ReadOptimizedSystemTable read_optimized("/tmp/table", table_schema,
options);
+
+ for (WarmupLevel level : {WarmupLevel::NONE, WarmupLevel::RAW,
WarmupLevel::DECODED}) {
+ ReadContextBuilder builder("/tmp/table");
+ builder.SetOptions(options).SetWarmupLevel(level);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext>
caller_unique_context, builder.Finish());
+ std::shared_ptr<ReadContext>
caller_context(std::move(caller_unique_context));
+ ASSERT_EQ(level, caller_context->GetWarmupLevel());
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> ro_context,
+
read_optimized.CreateDataReadContext(caller_context));
+ EXPECT_EQ(level, ro_context->GetWarmupLevel()) << "$ro dropped the
caller's WarmupLevel";
+
+ // The changelog chain is checked where the context is built rather
than on the read: the
+ // ChangelogTableRead that wraps it is local to
audit_log_system_table.cpp, so a test cannot
+ // name the type to reach the read underneath it. EXPECT rather than
ASSERT, because the
+ // three chains are independent and one broken chain must not hide
another.
+ ASSERT_OK_AND_ASSIGN(auto audit_log_options, audit_log.ReadOptions());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> audit_log_context,
+ audit_log.CreateDataReadContext(caller_context,
audit_log_options));
+ EXPECT_EQ(level, audit_log_context->GetWarmupLevel())
+ << "$audit_log dropped the caller's WarmupLevel";
+
+ ASSERT_OK_AND_ASSIGN(auto binlog_options, binlog.ReadOptions());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> binlog_context,
+ binlog.CreateDataReadContext(caller_context,
binlog_options));
+ EXPECT_EQ(level, binlog_context->GetWarmupLevel())
+ << "$binlog dropped the caller's WarmupLevel";
+ }
+}
+
TEST(SystemTableTest,
TestGlobalSystemTableWithoutCatalogReturnsNotImplemented) {
ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", "/tmp", {}));
std::shared_ptr<FileSystem> shared_fs(std::move(fs));
diff --git a/src/paimon/testing/mock/mock_file_batch_reader.h
b/src/paimon/testing/mock/mock_file_batch_reader.h
index 866fc636..e017d19f 100644
--- a/src/paimon/testing/mock/mock_file_batch_reader.h
+++ b/src/paimon/testing/mock/mock_file_batch_reader.h
@@ -201,10 +201,20 @@ class MockFileBatchReader : public
PrefetchFileBatchReader {
}
void Close() override {}
+ /// Counts the hint instead of acting on it: this mock has no background
work to start, so the
+ /// count is what lets a test tell a wrapper that forwards Warmup() from
one that swallows it.
+ void Warmup() override {
+ warmup_count_++;
+ }
+
std::vector<std::pair<uint64_t, uint64_t>> GetReadRanges() const {
return read_ranges_;
}
+ int32_t GetWarmupCount() const {
+ return warmup_count_;
+ }
+
bool SupportPreciseBitmapSelection() const override {
return false;
}
@@ -266,6 +276,7 @@ class MockFileBatchReader : public PrefetchFileBatchReader {
Status next_batch_status_;
bool enable_randomize_batch_size_ = true;
std::vector<std::pair<uint64_t, uint64_t>> read_ranges_;
+ int32_t warmup_count_ = 0;
std::mt19937 random_engine_{std::random_device{}()}; //
NOLINT(whitespace/braces)
};
