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 143107ea perf(parquet): add opt-in parsed offset index caching (#314)
143107ea is described below
commit 143107ea4db8a11f9fd5966e661b7b0ab97d66dc
Author: wangyong9999 <[email protected]>
AuthorDate: Fri Sep 18 15:27:52 2026 +0800
perf(parquet): add opt-in parsed offset index caching (#314)
---
benchmark/parquet_format_benchmark.cpp | 110 ++++++++++++++++++---
docs/source/user_guide/parquet_metadata_cache.rst | 37 ++++++-
src/paimon/format/parquet/file_reader_wrapper.cpp | 34 ++++++-
src/paimon/format/parquet/file_reader_wrapper.h | 4 +-
.../format/parquet/file_reader_wrapper_test.cpp | 46 ++++++++-
.../parquet/page_filtered_row_group_reader.cpp | 22 ++---
.../format/parquet/parquet_file_batch_reader.cpp | 10 +-
.../parquet/parquet_file_batch_reader_test.cpp | 31 ++++++
src/paimon/format/parquet/parquet_format_defs.h | 4 +
9 files changed, 264 insertions(+), 34 deletions(-)
diff --git a/benchmark/parquet_format_benchmark.cpp
b/benchmark/parquet_format_benchmark.cpp
index 3823ac3b..eff80b39 100644
--- a/benchmark/parquet_format_benchmark.cpp
+++ b/benchmark/parquet_format_benchmark.cpp
@@ -74,6 +74,8 @@
#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/utils/roaring_bitmap32.h"
+#include "parquet/file_reader.h"
+#include "parquet/page_index.h"
namespace {
@@ -551,8 +553,9 @@ Result<int64_t> WriteParquetFile(const
std::shared_ptr<FileSystem>& fs, const st
const std::shared_ptr<arrow::Schema>& schema,
const
std::vector<std::shared_ptr<arrow::Array>>& batches,
const std::map<std::string, std::string>&
options,
- const std::string& compression) {
- ParquetWriterBuilder writer_builder(schema, kWriteBatchSize, options);
+ const std::string& compression,
+ int32_t write_batch_size = kWriteBatchSize) {
+ ParquetWriterBuilder writer_builder(schema, write_batch_size, options);
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> out, fs->Create(path,
/*overwrite=*/true));
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatWriter> writer,
writer_builder.Build(out, compression));
@@ -704,8 +707,9 @@ class ReadFixture {
public:
ReadFixture(const std::string& file_name, const
std::shared_ptr<arrow::Schema>& schema,
const BatchFactory& make_batch,
- const std::map<std::string, std::string>& write_options = {}) {
- status_ = Build(file_name, schema, make_batch, write_options);
+ const std::map<std::string, std::string>& write_options = {},
+ int32_t write_batch_size = kWriteBatchSize) {
+ status_ = Build(file_name, schema, make_batch, write_options,
write_batch_size);
}
const Status& status() const {
@@ -731,17 +735,19 @@ class ReadFixture {
private:
Status Build(const std::string& file_name, const
std::shared_ptr<arrow::Schema>& schema,
const BatchFactory& make_batch,
- const std::map<std::string, std::string>& write_options) {
+ const std::map<std::string, std::string>& write_options,
+ int32_t write_batch_size) {
PAIMON_ASSIGN_OR_RAISE(env_, BenchmarkEnv::Create());
path_ = env_->PathOf(file_name);
PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<arrow::Array>>
batches,
MakeBatches(schema, make_batch, kRowsPerBatch));
std::map<std::string, std::string> options = write_options;
- options[paimon::parquet::PARQUET_PAGE_SIZE] =
std::to_string(kPageSizeBytes);
- options[paimon::parquet::PARQUET_WRITE_MAX_ROW_GROUP_LENGTH] =
- std::to_string(kRowGroupLength);
- PAIMON_ASSIGN_OR_RAISE(file_bytes_, WriteParquetFile(env_->fs(),
path_, schema, batches,
- options,
kDefaultCompression));
+ options.emplace(paimon::parquet::PARQUET_PAGE_SIZE,
std::to_string(kPageSizeBytes));
+ options.emplace(paimon::parquet::PARQUET_WRITE_MAX_ROW_GROUP_LENGTH,
+ std::to_string(kRowGroupLength));
+ PAIMON_ASSIGN_OR_RAISE(file_bytes_,
+ WriteParquetFile(env_->fs(), path_, schema,
batches, options,
+ kDefaultCompression,
write_batch_size));
return Status::OK();
}
@@ -751,12 +757,17 @@ class ReadFixture {
Status status_;
};
+std::map<std::string, std::unique_ptr<ReadFixture>>& ReadFixtures() {
+ static std::map<std::string, std::unique_ptr<ReadFixture>> fixtures;
+ return fixtures;
+}
+
// Built on first use, so a case excluded by --benchmark_filter never pays to
write its file.
// google/benchmark may report from a different thread than it registered on,
hence the lock.
const ReadFixture& GetFixture(const std::string& key,
const
std::function<std::unique_ptr<ReadFixture>()>& build) {
static std::mutex mutex;
- static std::map<std::string, std::unique_ptr<ReadFixture>> fixtures;
+ auto& fixtures = ReadFixtures();
std::lock_guard<std::mutex> guard(mutex);
std::unique_ptr<ReadFixture>& fixture = fixtures[key];
if (!fixture) {
@@ -1229,6 +1240,71 @@ void RunReadBenchmark(::benchmark::State& state, const
ReadFixture& fixture,
state.counters["batches"] =
::benchmark::Counter(static_cast<double>(stats.batches));
}
+// Sparse lookup into a large row group of 1 KiB values. Page size controls
how much
+// OffsetIndex parsing each lookup repeats; full scan is the
decoding-dominated control.
+void BM_ParquetRead_OffsetIndexCache(::benchmark::State& state) {
+ const int64_t page_size = state.range(0);
+ const bool cache = state.range(1) != 0;
+ const bool full_scan = state.range(2) != 0;
+ const std::string key = "offset_index_" + std::to_string(page_size);
+ const auto& fixture = GetFixture(key, [key, page_size] {
+ return std::make_unique<ReadFixture>(
+ key + ".parquet", StringSchema(),
+ [](const std::shared_ptr<arrow::Schema>& schema, int64_t offset,
+ int64_t rows) -> Result<std::shared_ptr<arrow::Array>> {
+ arrow::StringBuilder builder;
+ for (int64_t row = offset; row < offset + rows; ++row) {
+ std::string value = std::to_string(row);
+ uint64_t random = static_cast<uint64_t>(row) + 1;
+ while (value.size() < 1024) {
+ random ^= random << 13;
+ random ^= random >> 7;
+ random ^= random << 17;
+ value.push_back('a' + random % 26);
+ }
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append(value));
+ }
+ std::shared_ptr<arrow::Array> values;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&values));
+ return MakeStructArray(schema->fields(), {values});
+ },
+ std::map<std::string, std::string>{
+ {paimon::parquet::PARQUET_PAGE_SIZE,
std::to_string(page_size)},
+ {paimon::parquet::PARQUET_WRITE_MAX_ROW_GROUP_LENGTH,
std::to_string(kRowsPerFile)},
+ {paimon::parquet::PARQUET_ENABLE_DICTIONARY, "false"}},
+ /*write_batch_size=*/16);
+ });
+ if (FailBenchmark(state, fixture.status())) {
+ return;
+ }
+ auto page_count = [&]() -> Result<int64_t> {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input,
+ fixture.fs()->Open(fixture.path()));
+ auto stream = std::make_shared<ArrowInputStreamAdapter>(input,
fixture.file_bytes(),
+
fixture.arrow_pool());
+ auto reader = ::parquet::ParquetFileReader::Open(stream);
+ return reader->GetPageIndexReader()
+ ->RowGroup(0)
+ ->GetOffsetIndex(0)
+ ->page_locations()
+ .size();
+ }();
+ if (FailBenchmark(state, page_count.status())) {
+ return;
+ }
+ state.counters["pages"] =
::benchmark::Counter(static_cast<double>(page_count.value()));
+ std::optional<RoaringBitmap32> bitmap;
+ if (!full_scan) {
+ bitmap.emplace();
+ bitmap->Add(kRowsPerFile / 2);
+ }
+ RunReadBenchmark(
+ state, fixture, StringSchema(), nullptr, bitmap,
+ {{paimon::parquet::PARQUET_READ_ENABLE_OFFSET_INDEX_CACHE, cache ?
"true" : "false"},
+ {paimon::parquet::PARQUET_READ_BITMAP_ROW_RANGE_REFINING_STRATEGY,
"trim"}},
+ kReadBatchSize, full_scan ? RowExpectation{} : RowExpectation{1, 1});
+}
+
void BM_ParquetRead_FullScan(::benchmark::State& state) {
RunReadBenchmark(state, FlatFixture(), FlatSchema(), /*predicate=*/nullptr,
/*selection_bitmap=*/std::nullopt, /*options=*/{},
kReadBatchSize);
@@ -1496,6 +1572,16 @@ BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, zstd,
"zstd")
->Unit(benchmark::kMillisecond)
->UseRealTime();
+BENCHMARK(BM_ParquetRead_OffsetIndexCache)
+ ->ArgNames({"page_bytes", "cache", "full_scan"})
+ ->Args({16384, 0, 0})
+ ->Args({16384, 1, 0})
+ ->Args({1048576, 0, 0})
+ ->Args({1048576, 1, 0})
+ ->Args({16384, 0, 1})
+ ->Args({16384, 1, 1})
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
BENCHMARK(BM_ParquetRead_FullScan)->Unit(benchmark::kMillisecond)->UseRealTime();
BENCHMARK_CAPTURE(BM_ParquetRead_Projection, id, "id")
->Unit(benchmark::kMillisecond)
@@ -1590,6 +1676,8 @@ int main(int argc, char** argv) {
return 1;
}
::benchmark::RunSpecifiedBenchmarks();
+ // Delete temporary files before filesystem singletons (including IOHook)
are destroyed.
+ ReadFixtures().clear();
::benchmark::Shutdown();
return g_failed.load() ? 1 : 0;
}
diff --git a/docs/source/user_guide/parquet_metadata_cache.rst
b/docs/source/user_guide/parquet_metadata_cache.rst
index 804f8432..514f1b64 100644
--- a/docs/source/user_guide/parquet_metadata_cache.rst
+++ b/docs/source/user_guide/parquet_metadata_cache.rst
@@ -37,8 +37,9 @@ This optimization is useful when the same Parquet files are
opened repeatedly
in the same process, for example repeated ``get`` or ``scan`` requests over the
same snapshot. On a cache hit, the read path avoids reading the Parquet footer
bytes from the filesystem again. Paimon C++ still parses the cached footer
bytes
-into ``parquet::FileMetaData`` for each reader open. Data pages, page indexes,
-and column chunks are still read from the file as usual.
+into ``parquet::FileMetaData`` for each reader open. ColumnIndex and
OffsetIndex
+bytes also use ``CacheKind::DATA_FILE_FOOTER`` with their actual positions and
+lengths. Data pages and column chunks are not stored in this metadata cache.
Configuration
-------------
@@ -100,6 +101,38 @@ Passing ``nullptr`` or omitting ``WithCache()`` leaves
Parquet metadata caching
disabled. If a file URI cannot be obtained, Paimon C++ also bypasses the cache
and opens the Parquet file normally.
+Reader-Local Page Indexes
+-------------------------
+
+Set ``parquet.read.enable-offset-index-cache=true`` to reuse parsed OffsetIndex
+objects during bitmap trimming, page-range planning and filtered decoding
within
+one file reader. The option defaults to ``false``; it does not require
+``WithCache()``. Enable it only after measuring the CPU/memory trade-off for
the
+workload, especially when projecting many columns.
+
+These objects are not stored in the caller's shared ``Cache``. They are
released
+with their owning row-group index reader; the existing limit of 1,024 retained
+row-group readers is a count limit, not a parsed-index byte budget. Restricted
+predicate index readers remain separate so their column hints do not restrict
+later projected-column reads.
+
+Parsed page locations require approximately
+``retained row groups x accessed columns x pages per column x
sizeof(PageLocation)``
+bytes, plus vector/map/object overhead, in addition to serialized index
buffers.
+For example, with a 24-byte ``PageLocation``, 1,024 retained row groups, 10
accessed
+columns and 1,000 pages per column require about 234 MiB for page locations
alone.
+There is no fixed byte upper bound; memory scales with the file's index sizes.
+
+This is independent of ``ReadAheadCache`` and its per-file ``FileBlockCache``,
+which reuse bytes rather than parsed objects. That block cache survives resets
+of the prefetch plan, but does not provide reuse across independent file-cache
+lifetimes. This optimization introduces no data-cache option or shared data
cache.
+
+Filesystem or read-ahead byte caching does not replace this optimization: Arrow
+deserializes the cached index bytes again on every ``GetOffsetIndex`` call.
+With the option disabled, Paimon retains the existing byte-buffer reuse without
+retaining parsed OffsetIndex objects.
+
Future Optimizations
--------------------
diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp
b/src/paimon/format/parquet/file_reader_wrapper.cpp
index 19c35310..56128328 100644
--- a/src/paimon/format/parquet/file_reader_wrapper.cpp
+++ b/src/paimon/format/parquet/file_reader_wrapper.cpp
@@ -45,6 +45,33 @@ namespace paimon::parquet {
namespace {
+// Planning and decoding repeatedly request the same offsets. Keep parsed
indexes
+// with the row-group reader, not in the cross-reader byte cache.
+class CachedRowGroupPageIndexReader : public
::parquet::RowGroupPageIndexReader {
+ public:
+ explicit CachedRowGroupPageIndexReader(
+ std::shared_ptr<::parquet::RowGroupPageIndexReader> reader)
+ : reader_(std::move(reader)) {}
+
+ std::shared_ptr<::parquet::ColumnIndex> GetColumnIndex(int32_t i) override
{
+ return reader_->GetColumnIndex(i);
+ }
+
+ std::shared_ptr<::parquet::OffsetIndex> GetOffsetIndex(int32_t i) override
{
+ auto found = offset_indexes_.find(i);
+ if (found != offset_indexes_.end()) {
+ return found->second;
+ }
+ auto index = reader_->GetOffsetIndex(i);
+ offset_indexes_.emplace(i, index);
+ return index;
+ }
+
+ private:
+ std::shared_ptr<::parquet::RowGroupPageIndexReader> reader_;
+ std::map<int32_t, std::shared_ptr<::parquet::OffsetIndex>> offset_indexes_;
+};
+
// Merge overlapping or adjacent ReadRanges into a minimal set of
non-overlapping ranges.
// PreBufferRanges requires non-overlapping ranges, so this is necessary when
combining
// ranges from multiple sources (page-level ranges, column chunk ranges, etc.).
@@ -87,7 +114,7 @@ std::vector<::arrow::io::ReadRange> MergeOverlappingRanges(
Result<std::unique_ptr<FileReaderWrapper>> FileReaderWrapper::Create(
std::unique_ptr<::parquet::arrow::FileReader>&& file_reader, int64_t
batch_size,
- std::shared_ptr<::arrow::MemoryPool> pool) {
+ bool enable_offset_index_cache, std::shared_ptr<::arrow::MemoryPool> pool)
{
try {
if (file_reader == nullptr) {
return Status::Invalid("file reader wrapper create failed. file
reader is nullptr");
@@ -111,6 +138,7 @@ Result<std::unique_ptr<FileReaderWrapper>>
FileReaderWrapper::Create(
arrow::internal::Iota(file_reader->parquet_reader()->metadata()->num_columns());
auto file_reader_wrapper = std::unique_ptr<FileReaderWrapper>(new
FileReaderWrapper(
std::move(file_reader), all_row_group_ranges, num_rows,
batch_size, pool));
+ file_reader_wrapper->enable_offset_index_cache_ =
enable_offset_index_cache;
std::vector<TargetRowGroup> all_target_row_groups;
for (int32_t i = 0; i < file_reader_wrapper->GetNumberOfRowGroups();
i++) {
all_target_row_groups.emplace_back(/*rg_index=*/i,
/*is_partially_matched=*/false,
@@ -341,6 +369,10 @@ std::shared_ptr<::parquet::RowGroupPageIndexReader>
FileReaderWrapper::GetRowGro
auto page_index_reader = GetPageIndexReader();
if (page_index_reader) {
row_group_page_index_reader =
page_index_reader->RowGroup(row_group_index);
+ if (row_group_page_index_reader && enable_offset_index_cache_) {
+ row_group_page_index_reader =
+
std::make_shared<CachedRowGroupPageIndexReader>(row_group_page_index_reader);
+ }
}
// To avoid OOM, limit the number of row group page index readers cached
in memory.
diff --git a/src/paimon/format/parquet/file_reader_wrapper.h
b/src/paimon/format/parquet/file_reader_wrapper.h
index 02e6ae5b..e9edca16 100644
--- a/src/paimon/format/parquet/file_reader_wrapper.h
+++ b/src/paimon/format/parquet/file_reader_wrapper.h
@@ -60,7 +60,7 @@ class FileReaderWrapper {
static Result<std::unique_ptr<FileReaderWrapper>> Create(
std::unique_ptr<::parquet::arrow::FileReader>&& reader, int64_t
batch_size,
- std::shared_ptr<arrow::MemoryPool> pool);
+ bool enable_offset_index_cache, std::shared_ptr<arrow::MemoryPool>
pool);
/// Seek to the specified row number.
/// @param row_number The row to seek to (must be at a row group boundary).
@@ -228,6 +228,8 @@ class FileReaderWrapper {
// Track pre-buffered ranges so we can wait on destruction
std::vector<::arrow::io::ReadRange> prebuffered_ranges_;
+ bool enable_offset_index_cache_ = false;
+
// Arrow caches the file-level PageIndexReader, but RowGroup() creates a
new reader each time.
// Keep one reader per row group so its page-index buffers are shared by
all read stages.
std::map<int32_t, std::shared_ptr<::parquet::RowGroupPageIndexReader>>
diff --git a/src/paimon/format/parquet/file_reader_wrapper_test.cpp
b/src/paimon/format/parquet/file_reader_wrapper_test.cpp
index e2335107..74b6bb8e 100644
--- a/src/paimon/format/parquet/file_reader_wrapper_test.cpp
+++ b/src/paimon/format/parquet/file_reader_wrapper_test.cpp
@@ -187,13 +187,16 @@ class FileReaderWrapperTest : public ::testing::Test {
}
Result<std::unique_ptr<FileReaderWrapper>> PrepareReaderWrapper(
- const std::string& file_path, int64_t wrapper_batch_size = 0) {
+ const std::string& file_path, int64_t wrapper_batch_size = 0,
+ bool enable_offset_index_cache = false) {
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> in,
fs_->Open(file_path));
- return PrepareReaderWrapperOnStream(std::move(in), wrapper_batch_size);
+ return PrepareReaderWrapperOnStream(std::move(in), wrapper_batch_size,
+ enable_offset_index_cache);
}
Result<std::unique_ptr<FileReaderWrapper>> PrepareReaderWrapperOnStream(
- std::shared_ptr<InputStream> in, int64_t wrapper_batch_size = 0) {
+ std::shared_ptr<InputStream> in, int64_t wrapper_batch_size = 0,
+ bool enable_offset_index_cache = false) {
PAIMON_ASSIGN_OR_RAISE(int64_t file_length, in->Length());
auto input_stream =
std::make_unique<ArrowInputStreamAdapter>(std::move(in),
file_length, arrow_pool_);
@@ -212,7 +215,8 @@ class FileReaderWrapperTest : public ::testing::Test {
PAIMON_RETURN_NOT_OK_FROM_ARROW(file_reader_builder.memory_pool(arrow_pool_.get())
->properties(arrow_reader_props)
->Build(&file_reader));
- return FileReaderWrapper::Create(std::move(file_reader),
wrapper_batch_size, arrow_pool_);
+ return FileReaderWrapper::Create(std::move(file_reader),
wrapper_batch_size,
+ enable_offset_index_cache,
arrow_pool_);
}
void PrepareParquetFile(const std::string& file_path, int32_t row_count,
@@ -274,6 +278,7 @@ TEST_F(FileReaderWrapperTest, EmptyFile) {
TEST_F(FileReaderWrapperTest, NullFileReader) {
ASSERT_NOK_WITH_MSG(FileReaderWrapper::Create(nullptr,
/*batch_size=*/0,
+
/*enable_offset_index_cache=*/false,
/*pool=*/arrow_pool_),
"file reader wrapper create failed. file reader is
nullptr");
}
@@ -311,6 +316,39 @@ TEST_F(FileReaderWrapperTest,
PredicateReadsOnlyItsPageIndexesAndKeepsPayloadRea
ASSERT_FALSE(end);
}
+TEST_F(FileReaderWrapperTest, ReusesParsedOffsetIndexesWithinFileReader) {
+ std::string file_path = PathUtil::JoinPath(dir_->Str(),
"parsed-index.parquet");
+ PrepareParquetFile(file_path, /*row_count=*/2000,
/*enable_page_index=*/true);
+ ASSERT_OK_AND_ASSIGN(auto uncached_reader,
PrepareReaderWrapper(file_path));
+ auto uncached = uncached_reader->GetRowGroupPageIndexReader(0);
+ auto first_uncached = uncached->GetOffsetIndex(0);
+ ASSERT_NE(first_uncached, uncached->GetOffsetIndex(0));
+ ASSERT_OK_AND_ASSIGN(auto reader, PrepareReaderWrapper(file_path, 0,
true));
+ std::weak_ptr<::parquet::OffsetIndex> released_index;
+ {
+ auto first = reader->GetRowGroupPageIndexReader(0);
+ auto second = reader->GetRowGroupPageIndexReader(1);
+ ASSERT_TRUE(first);
+ ASSERT_TRUE(second);
+ ASSERT_NE(first, second);
+ for (int32_t col = 0; col < 3; ++col) {
+ auto offset = first->GetOffsetIndex(col);
+ auto column = first->GetColumnIndex(col);
+ ASSERT_TRUE(offset);
+ ASSERT_TRUE(column);
+ ASSERT_EQ(offset, first->GetOffsetIndex(col));
+ ASSERT_NE(offset, second->GetOffsetIndex(col));
+ released_index = offset;
+ }
+ ASSERT_THROW(first->GetOffsetIndex(-1), ::parquet::ParquetException);
+ ASSERT_THROW(first->GetColumnIndex(100), ::parquet::ParquetException);
+ ASSERT_TRUE(first->GetOffsetIndex(0));
+ }
+ ASSERT_FALSE(released_index.expired());
+ reader.reset();
+ ASSERT_TRUE(released_index.expired());
+}
+
TEST_F(FileReaderWrapperTest, Simple) {
std::string file_path = PathUtil::JoinPath(dir_->Str(), "test.parquet");
PrepareParquetFile(file_path, /*row_count=*/5500);
diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp
b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp
index 78f756fa..b712e54d 100644
--- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp
+++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp
@@ -302,10 +302,10 @@ Result<std::shared_ptr<arrow::ChunkedArray>>
PageFilteredRowGroupReader::ReadFil
int64_t row_group_row_count, ::parquet::arrow::FileReader*
arrow_file_reader) {
// Factory: set a direct data page read plan on every leaf (per-leaf
OffsetIndex).
// The plan lets Arrow jump over unselected page headers as well as page
bodies.
- auto factory =
- [row_group_index, &rg_page_index_reader, &row_ranges,
row_group_row_count](
- int col_idx,
- ::parquet::ParquetFileReader* reader) ->
::parquet::arrow::FileColumnIterator* {
+ std::unordered_set<int> direct_read_columns;
+ auto factory = [row_group_index, &rg_page_index_reader, &row_ranges,
row_group_row_count,
+ &direct_read_columns](int col_idx,
::parquet::ParquetFileReader* reader)
+ -> ::parquet::arrow::FileColumnIterator* {
bool has_data_page_read_plan = false;
int64_t first_data_page_offset = 0;
std::vector<::parquet::DataPageReadPlanEntry> data_pages;
@@ -320,6 +320,7 @@ Result<std::shared_ptr<arrow::ChunkedArray>>
PageFilteredRowGroupReader::ReadFil
first_data_page_offset = plan->first_data_page_offset;
data_pages = std::move(plan->data_pages);
has_data_page_read_plan = true;
+ direct_read_columns.insert(col_idx);
}
}
}
@@ -359,15 +360,12 @@ Result<std::shared_ptr<arrow::ChunkedArray>>
PageFilteredRowGroupReader::ReadFil
auto row_group_metadata =
arrow_file_reader->parquet_reader()->metadata()->RowGroup(row_group_index);
column_chunk = row_group_metadata->ColumnChunk(col_idx);
- if (rg_page_index_reader) {
+ if (direct_read_columns.count(col_idx)) {
auto offset_index =
rg_page_index_reader->GetOffsetIndex(col_idx);
- if (offset_index && MakeDataPageReadPlan(row_ranges,
offset_index, *column_chunk,
- row_group_row_count))
{
- auto [compressed, total] =
- ComputeCompressedRowRanges(row_ranges, offset_index,
row_group_row_count);
- effective_ranges = std::move(compressed);
- effective_total = total;
- }
+ auto [compressed, total] =
+ ComputeCompressedRowRanges(row_ranges, offset_index,
row_group_row_count);
+ effective_ranges = std::move(compressed);
+ effective_total = total;
}
}
diff --git a/src/paimon/format/parquet/parquet_file_batch_reader.cpp
b/src/paimon/format/parquet/parquet_file_batch_reader.cpp
index 76f0df13..faa18c3e 100644
--- a/src/paimon/format/parquet/parquet_file_batch_reader.cpp
+++ b/src/paimon/format/parquet/parquet_file_batch_reader.cpp
@@ -321,9 +321,13 @@ Result<std::unique_ptr<ParquetFileBatchReader>>
ParquetFileBatchReader::Create(
PAIMON_RETURN_NOT_OK_FROM_ARROW(file_reader_builder.memory_pool(pool.get())
->properties(arrow_reader_properties)
->Build(&file_reader));
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileReaderWrapper> reader,
-
FileReaderWrapper::Create(std::move(file_reader),
-
static_cast<int64_t>(batch_size), pool));
+ PAIMON_ASSIGN_OR_RAISE(bool enable_offset_index_cache,
+ OptionsUtils::GetValueFromMap<bool>(
+ options,
PARQUET_READ_ENABLE_OFFSET_INDEX_CACHE, false));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FileReaderWrapper> reader,
+ FileReaderWrapper::Create(std::move(file_reader),
static_cast<int64_t>(batch_size),
+ enable_offset_index_cache, pool));
// Arrow silently ignores set_read_dictionary for leaves it cannot
read as dictionaries,
// so take the columns it really emits that way from the schema it
just derived.
std::set<std::string> dictionary_fields;
diff --git a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
index 3c1875ae..f25e0527 100644
--- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
+++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp
@@ -427,6 +427,37 @@ TEST_F(ParquetFileBatchReaderTest,
TestCachedFooterKeepsAllocatorAliveUntilEvict
ASSERT_TRUE(cached_pool.expired());
}
+TEST_F(ParquetFileBatchReaderTest, TestOffsetIndexCacheOption) {
+ WriteArray(file_path_, struct_array_, schema_, 1, false, 3, 1);
+ auto projection = arrow::schema({schema_->GetFieldByName("f4")});
+ auto predicate = PredicateBuilder::Equal(0, "f4", FieldType::INT,
Literal(300002));
+ for (const std::string setting : {"", "false", "true", "invalid"}) {
+ SCOPED_TRACE(setting);
+ std::map<std::string, std::string> options;
+ if (!setting.empty()) {
+ options[PARQUET_READ_ENABLE_OFFSET_INDEX_CACHE] = setting;
+ }
+ ParquetReaderBuilder builder(options, 10);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input,
fs_->Open(file_path_));
+ if (setting == "invalid") {
+ ASSERT_NOK(builder.Build(input));
+ continue;
+ }
+ ASSERT_OK_AND_ASSIGN(auto reader, builder.Build(input));
+ ArrowSchema c_schema;
+ ASSERT_TRUE(arrow::ExportSchema(*projection, &c_schema).ok());
+ ASSERT_OK(reader->SetReadSchema(&c_schema, predicate, std::nullopt));
+ ASSERT_OK_AND_ASSIGN(auto result,
+
paimon::test::ReadResultCollector::CollectResult(reader.get()));
+ ASSERT_EQ(1, result->length());
+ auto rows =
std::dynamic_pointer_cast<arrow::StructArray>(result->chunk(0));
+ ASSERT_TRUE(rows);
+ auto values =
std::dynamic_pointer_cast<arrow::Int32Array>(rows->field(0));
+ ASSERT_TRUE(values);
+ ASSERT_EQ(300002, values->Value(0));
+ }
+}
+
TEST_F(ParquetFileBatchReaderTest, TestPointReadReusesFooterAndPageIndexes) {
WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1,
/*enable_dictionary=*/false, /*max_row_group_length=*/3,
diff --git a/src/paimon/format/parquet/parquet_format_defs.h
b/src/paimon/format/parquet/parquet_format_defs.h
index 6bc811e5..f45603fc 100644
--- a/src/paimon/format/parquet/parquet_format_defs.h
+++ b/src/paimon/format/parquet/parquet_format_defs.h
@@ -99,6 +99,10 @@ static inline const char
PARQUET_READ_PREDICATE_NODE_COUNT_LIMIT[] =
static inline const char PARQUET_READ_ENABLE_PAGE_INDEX_FILTER[] =
"parquet.read.enable-page-index-filter";
+// Retain parsed OffsetIndex objects within a reader. Disabled by default.
+static inline const char PARQUET_READ_ENABLE_OFFSET_INDEX_CACHE[] =
+ "parquet.read.enable-offset-index-cache";
+
// Default is true.
static inline const char PARQUET_READ_ENABLE_PRE_BUFFER[] =
"parquet.read.enable-pre-buffer";
