From 018c8fd922f0cf1156001403df777eca1f552bf5 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sat, 5 Sep 2026 10:02:59 -0400 Subject: [PATCH 1/8] perf(parquet): reuse immutable page-index bytes through reader cache --- .../parquet_file_batch_reader_test.cpp | 86 ++++++++++++++ .../format/parquet/parquet_input_stream.h | 108 ++++++++++++++++++ .../format/parquet/parquet_reader_builder.h | 12 +- 3 files changed, 201 insertions(+), 5 deletions(-) create mode 100644 src/paimon/format/parquet/parquet_input_stream.h 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 3e4824e34..bcd11a55f 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp @@ -343,6 +343,92 @@ TEST_F(ParquetFileBatchReaderTest, TestParquetMetadataCacheBypassesWhenGetUriFai ASSERT_EQ(0, cache->Size()); } +TEST_F(ParquetFileBatchReaderTest, TestPageIndexBytesSurviveReaderClose) { + WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1, + /*enable_dictionary=*/false, /*max_row_group_length=*/3, + /*max_page_size=*/1); + auto cache = std::make_shared(CacheKind::DATA_FILE_FOOTER, + 128 * 1024 * 1024); + ASSERT_OK_AND_ASSIGN(std::shared_ptr raw_input, fs_->Open(file_path_)); + ASSERT_OK_AND_ASSIGN(int64_t length, raw_input->Length()); + auto raw = std::make_shared(raw_input, length, pool_); + auto raw_reader = ::parquet::ParquetFileReader::Open(raw); + auto metadata = raw_reader->metadata(); + int64_t expected_index_reads = 0; + for (int32_t round = 0; round < 3; ++round) { + if (round == 2) { + cache->InvalidateAll(); + } + ASSERT_OK_AND_ASSIGN(std::shared_ptr input, fs_->Open(file_path_)); + auto stream = std::make_shared(input, length, pool_, GetDefaultPool(), + cache, file_path_); + stream->SetPageIndexRanges(*metadata); + for (int32_t rg = 0; rg < metadata->num_row_groups(); ++rg) { + auto ranges = ::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup( + *metadata->RowGroup(rg), {}); + for (const auto& range : {ranges.column_index, ranges.offset_index}) { + ASSERT_TRUE(range.has_value()); + auto expected = raw->ReadAt(range->offset, range->length); + ASSERT_TRUE(expected.ok()) << expected.status().ToString(); + auto actual = stream->ReadAt(range->offset, range->length); + ASSERT_TRUE(actual.ok()) << actual.status().ToString(); + ASSERT_TRUE(actual.ValueOrDie()->Equals(*expected.ValueOrDie())); + if (round != 1) { + ++expected_index_reads; + } + } + } + ASSERT_EQ(expected_index_reads, cache->SupplierCallCount()); + // Reader 2 serves all indexes from the shared cache without storage IO. + if (round == 1) { + ASSERT_EQ(0, stream->StorageReadBytes()->load()); + } else { + ASSERT_GT(stream->StorageReadBytes()->load(), 0); + } + // Ordinary file bytes must still use storage, even on a cache hit round. + uint64_t bytes_before = stream->StorageReadBytes()->load(); + auto magic = stream->ReadAt(0, 4); + ASSERT_TRUE(magic.ok()) << magic.status().ToString(); + ASSERT_EQ("PAR1", magic.ValueOrDie()->ToString()); + ASSERT_EQ(bytes_before + 4, stream->StorageReadBytes()->load()); + ASSERT_EQ(expected_index_reads, cache->SupplierCallCount()); + ASSERT_TRUE(stream->Close().ok()); + } +} + +TEST_F(ParquetFileBatchReaderTest, TestPointReadReusesFooterAndPageIndexes) { + WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1, + /*enable_dictionary=*/false, /*max_row_group_length=*/3, + /*max_page_size=*/1); + auto cache = std::make_shared(CacheKind::DATA_FILE_FOOTER, + 128 * 1024 * 1024); + auto projection = arrow::schema({schema_->GetFieldByName("f4"), schema_->GetFieldByName("f8")}); + auto predicate = PredicateBuilder::Equal(0, "f4", FieldType::INT, Literal(300002)); + int64_t cold_reads = 0; + std::shared_ptr cold_result; + for (int32_t round = 0; round < 2; ++round) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr input, fs_->Open(file_path_)); + ParquetReaderBuilder builder({{PARQUET_READ_ENABLE_PAGE_INDEX_FILTER, "true"}}, 10); + builder.WithCache(cache); + 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()); + if (round == 0) { + cold_reads = cache->SupplierCallCount(); + ASSERT_GT(cold_reads, 1); // Footer plus actual page-index ranges. + cold_result = result; + } else { + ASSERT_TRUE(result->Equals(cold_result)); + ASSERT_EQ(cold_reads, cache->SupplierCallCount()); + ASSERT_GT(cache->GetCount(), cold_reads); + } + } +} + TEST_F(ParquetFileBatchReaderTest, TestReadBinaryWrittenFromBinaryAndLargeBinary) { auto check_binary_read_result = [&](const std::shared_ptr& write_type, const std::string& file_name) { diff --git a/src/paimon/format/parquet/parquet_input_stream.h b/src/paimon/format/parquet/parquet_input_stream.h new file mode 100644 index 000000000..1d38fa407 --- /dev/null +++ b/src/paimon/format/parquet/parquet_input_stream.h @@ -0,0 +1,108 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed 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. + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include "paimon/cache/cache.h" +#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/memory/memory_pool.h" +#include "parquet/page_index.h" + +namespace paimon::parquet { + +// Cache immutable page-index bytes, not Arrow readers: the latter borrow their +// input stream, reader properties and decryptor from the current file reader. +class ParquetInputStream : public ArrowInputStreamAdapter { + public: + ParquetInputStream(const std::shared_ptr& input, int64_t file_size, + const std::shared_ptr& arrow_pool, + const std::shared_ptr& pool, const std::shared_ptr& cache, + const std::string& file_uri) + : ArrowInputStreamAdapter(input, file_size, arrow_pool), + pool_(pool), + cache_(cache), + file_uri_(file_uri) {} + + // Called before the stream is published to the file reader. Only index + // ranges described by its footer are eligible; data pages remain uncached. + void SetPageIndexRanges(const ::parquet::FileMetaData& metadata) { + for (int32_t rg = 0; rg < metadata.num_row_groups(); ++rg) { + auto ranges = ::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup( + *metadata.RowGroup(rg), {}); + for (const auto& range : {ranges.column_index, ranges.offset_index}) { + if (range.has_value()) { + index_ranges_.emplace(range->offset, range->length); + } + } + } + } + + using ArrowInputStreamAdapter::ReadAt; + + arrow::Result ReadAt(int64_t position, int64_t nbytes, void* out) override { + if (!cache_ || file_uri_.empty() || nbytes <= 0 || + nbytes > std::numeric_limits::max()) { + return ArrowInputStreamAdapter::ReadAt(position, nbytes, out); + } + auto range = index_ranges_.upper_bound(position); + if (range == index_ranges_.begin()) { + return ArrowInputStreamAdapter::ReadAt(position, nbytes, out); + } + --range; + if (position - range->first > range->second || + nbytes > range->second - (position - range->first)) { + return ArrowInputStreamAdapter::ReadAt(position, nbytes, out); + } + auto key = CacheKey::ForKind(file_uri_, position, static_cast(nbytes), + CacheKind::DATA_FILE_FOOTER); + auto value = cache_->Get( + key, + [this, position, + nbytes](const std::shared_ptr&) -> Result> { + MemorySegment segment = MemorySegment::AllocateHeapMemory(nbytes, pool_.get()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + int64_t size, + ArrowInputStreamAdapter::ReadAt(position, nbytes, segment.MutableData())); + if (size != nbytes) { + return Status::IOError("Short read of Parquet page index"); + } + return std::make_shared(segment, CacheCallback()); + }); + if (!value.ok()) { + return ToArrowStatus(value.status()); + } + if (!value.value() || value.value()->GetSegment().Size() != nbytes) { + return arrow::Status::IOError("Invalid Parquet page-index cache value"); + } + std::memcpy(out, value.value()->GetSegment().Data(), nbytes); + return nbytes; + } + + private: + std::shared_ptr pool_; + std::shared_ptr cache_; + std::string file_uri_; + std::map index_ranges_; +}; + +} // namespace paimon::parquet diff --git a/src/paimon/format/parquet/parquet_reader_builder.h b/src/paimon/format/parquet/parquet_reader_builder.h index e4d2ddb43..5d1d6185a 100644 --- a/src/paimon/format/parquet/parquet_reader_builder.h +++ b/src/paimon/format/parquet/parquet_reader_builder.h @@ -35,6 +35,7 @@ #include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/format/parquet/parquet_file_batch_reader.h" #include "paimon/format/parquet/parquet_format_defs.h" +#include "paimon/format/parquet/parquet_input_stream.h" #include "paimon/format/read_hints.h" #include "paimon/format/reader_builder.h" #include "paimon/memory/memory_pool.h" @@ -88,13 +89,14 @@ class ParquetReaderBuilder : public ReaderBuilder { file_uri = std::move(file_uri_result).value(); } } - auto unique_input_stream = - std::make_unique(path, file_length, arrow_pool_); - auto storage_read_bytes = unique_input_stream->StorageReadBytes(); - std::shared_ptr input_stream( - std::move(unique_input_stream)); + auto input_stream = std::make_shared( + path, file_length, arrow_pool_, pool_, cache_, file_uri); + auto storage_read_bytes = input_stream->StorageReadBytes(); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<::parquet::FileMetaData> file_metadata, GetCachedParquetMetadata(input_stream, file_uri, arrow_pool_)); + if (file_metadata) { + input_stream->SetPageIndexRanges(*file_metadata); + } return ParquetFileBatchReader::Create( std::move(input_stream), options_, batch_size_, std::move(file_metadata), std::move(storage_read_bytes), arrow_pool_, hints_); From 3ec07882c417fac324369e7c2a97574b6ef8b593 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sat, 5 Sep 2026 10:06:49 -0400 Subject: [PATCH 2/8] fix(parquet): retain the allocator with cached metadata bytes --- .../parquet_file_batch_reader_test.cpp | 34 +++++++++++++++++-- .../format/parquet/parquet_input_stream.h | 17 +++++++++- .../format/parquet/parquet_reader_builder.h | 3 +- 3 files changed, 49 insertions(+), 5 deletions(-) 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 bcd11a55f..c52ceba95 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp @@ -355,13 +355,16 @@ TEST_F(ParquetFileBatchReaderTest, TestPageIndexBytesSurviveReaderClose) { auto raw_reader = ::parquet::ParquetFileReader::Open(raw); auto metadata = raw_reader->metadata(); int64_t expected_index_reads = 0; + std::weak_ptr cached_pool; for (int32_t round = 0; round < 3; ++round) { if (round == 2) { cache->InvalidateAll(); + ASSERT_TRUE(cached_pool.expired()); } ASSERT_OK_AND_ASSIGN(std::shared_ptr input, fs_->Open(file_path_)); - auto stream = std::make_shared(input, length, pool_, GetDefaultPool(), - cache, file_path_); + std::shared_ptr query_pool = GetMemoryPool(); + auto stream = std::make_shared(input, length, pool_, query_pool, cache, + file_path_); stream->SetPageIndexRanges(*metadata); for (int32_t rg = 0; rg < metadata->num_row_groups(); ++rg) { auto ranges = ::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup( @@ -393,7 +396,34 @@ TEST_F(ParquetFileBatchReaderTest, TestPageIndexBytesSurviveReaderClose) { ASSERT_EQ(bytes_before + 4, stream->StorageReadBytes()->load()); ASSERT_EQ(expected_index_reads, cache->SupplierCallCount()); ASSERT_TRUE(stream->Close().ok()); + if (round != 1) { + cached_pool = query_pool; + } + stream.reset(); + query_pool.reset(); + ASSERT_FALSE(cached_pool.expired()); + } + cache->InvalidateAll(); + ASSERT_TRUE(cached_pool.expired()); +} + +TEST_F(ParquetFileBatchReaderTest, TestCachedFooterKeepsAllocatorAliveUntilEviction) { + WriteArray(file_path_, struct_array_, schema_, /*write_batch_size=*/1, + /*enable_dictionary=*/false, /*max_row_group_length=*/3); + auto cache = std::make_shared(CacheKind::DATA_FILE_FOOTER, + 128 * 1024 * 1024); + std::weak_ptr cached_pool; + { + std::shared_ptr query_pool = GetMemoryPool(); + cached_pool = query_pool; + ParquetReaderBuilder builder({}, 10); + builder.WithMemoryPool(query_pool)->WithCache(cache); + ASSERT_OK_AND_ASSIGN(std::shared_ptr input, fs_->Open(file_path_)); + ASSERT_OK_AND_ASSIGN(auto reader, builder.Build(input)); } + ASSERT_FALSE(cached_pool.expired()); + cache->InvalidateAll(); + ASSERT_TRUE(cached_pool.expired()); } TEST_F(ParquetFileBatchReaderTest, TestPointReadReusesFooterAndPageIndexes) { diff --git a/src/paimon/format/parquet/parquet_input_stream.h b/src/paimon/format/parquet/parquet_input_stream.h index 1d38fa407..ceb7cdca6 100644 --- a/src/paimon/format/parquet/parquet_input_stream.h +++ b/src/paimon/format/parquet/parquet_input_stream.h @@ -30,6 +30,20 @@ namespace paimon::parquet { +inline MemorySegment AllocateParquetCacheSegment(int32_t size, + const std::shared_ptr& pool) { + struct PooledBytes { + PooledBytes(int32_t size, const std::shared_ptr& memory_pool) + : pool(memory_pool), bytes(size, memory_pool.get()) {} + // Destroy bytes before releasing their allocator. + std::shared_ptr pool; + Bytes bytes; + }; + auto owner = std::make_shared(size, pool); + auto* bytes = &owner->bytes; + return MemorySegment::Wrap(std::shared_ptr(std::move(owner), bytes)); +} + // Cache immutable page-index bytes, not Arrow readers: the latter borrow their // input stream, reader properties and decryptor from the current file reader. class ParquetInputStream : public ArrowInputStreamAdapter { @@ -79,7 +93,8 @@ class ParquetInputStream : public ArrowInputStreamAdapter { key, [this, position, nbytes](const std::shared_ptr&) -> Result> { - MemorySegment segment = MemorySegment::AllocateHeapMemory(nbytes, pool_.get()); + MemorySegment segment = + AllocateParquetCacheSegment(static_cast(nbytes), pool_); PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( int64_t size, ArrowInputStreamAdapter::ReadAt(position, nbytes, segment.MutableData())); diff --git a/src/paimon/format/parquet/parquet_reader_builder.h b/src/paimon/format/parquet/parquet_reader_builder.h index 5d1d6185a..5619b9304 100644 --- a/src/paimon/format/parquet/parquet_reader_builder.h +++ b/src/paimon/format/parquet/parquet_reader_builder.h @@ -125,8 +125,7 @@ class ParquetReaderBuilder : public ReaderBuilder { PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr metadata_footer, output_stream->Finish()); - MemorySegment segment = - MemorySegment::AllocateHeapMemory(metadata_footer->size(), pool_.get()); + MemorySegment segment = AllocateParquetCacheSegment(metadata_footer->size(), pool_); std::memcpy(segment.MutableData(), metadata_footer->data(), metadata_footer->size()); return segment; } From e1378bd5db9e6055c9b21d285c9afce26969dd18 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sat, 5 Sep 2026 10:08:39 -0400 Subject: [PATCH 3/8] style: use the ASF license header --- .../format/parquet/parquet_input_stream.h | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/paimon/format/parquet/parquet_input_stream.h b/src/paimon/format/parquet/parquet_input_stream.h index ceb7cdca6..cd81634c9 100644 --- a/src/paimon/format/parquet/parquet_input_stream.h +++ b/src/paimon/format/parquet/parquet_input_stream.h @@ -1,17 +1,20 @@ /* - * Copyright 2026-present Alibaba Inc. - * - * Licensed 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 + * 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. + * 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. */ #pragma once From ff1b68cc18345529303b3061c429034ac0591004 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 06:42:23 -0400 Subject: [PATCH 4/8] fix(parquet): include metadata definitions in cached stream --- src/paimon/format/parquet/parquet_input_stream.h | 1 + src/paimon/format/parquet/parquet_reader_builder.h | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/paimon/format/parquet/parquet_input_stream.h b/src/paimon/format/parquet/parquet_input_stream.h index cd81634c9..fbf45f244 100644 --- a/src/paimon/format/parquet/parquet_input_stream.h +++ b/src/paimon/format/parquet/parquet_input_stream.h @@ -29,6 +29,7 @@ #include "paimon/common/utils/arrow/arrow_input_stream_adapter.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/memory/memory_pool.h" +#include "parquet/metadata.h" #include "parquet/page_index.h" namespace paimon::parquet { diff --git a/src/paimon/format/parquet/parquet_reader_builder.h b/src/paimon/format/parquet/parquet_reader_builder.h index 5619b9304..fad961ace 100644 --- a/src/paimon/format/parquet/parquet_reader_builder.h +++ b/src/paimon/format/parquet/parquet_reader_builder.h @@ -89,8 +89,8 @@ class ParquetReaderBuilder : public ReaderBuilder { file_uri = std::move(file_uri_result).value(); } } - auto input_stream = std::make_shared( - path, file_length, arrow_pool_, pool_, cache_, file_uri); + auto input_stream = std::make_shared(path, file_length, arrow_pool_, + pool_, cache_, file_uri); auto storage_read_bytes = input_stream->StorageReadBytes(); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<::parquet::FileMetaData> file_metadata, GetCachedParquetMetadata(input_stream, file_uri, arrow_pool_)); From 57c094f2f878e58a1923fba7339f183fafcbb5c1 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 07:29:22 -0400 Subject: [PATCH 5/8] perf(parquet): read predicate column indexes only --- .../format/parquet/file_reader_wrapper.cpp | 31 ++++++++++++++- .../parquet/file_reader_wrapper_test.cpp | 38 +++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index fb5b65741..f69b89d46 100644 --- a/src/paimon/format/parquet/file_reader_wrapper.cpp +++ b/src/paimon/format/parquet/file_reader_wrapper.cpp @@ -29,10 +29,12 @@ #include "fmt/format.h" #include "paimon/common/utils/arrow/arrow_utils.h" #include "paimon/common/utils/math.h" +#include "paimon/common/utils/scope_guard.h" #include "paimon/format/parquet/column_index_filter.h" #include "paimon/format/parquet/page_filtered_row_group_reader.h" #include "paimon/format/parquet/parquet_format_defs.h" #include "paimon/macros.h" +#include "paimon/predicate/predicate_utils.h" #include "parquet/arrow/reader.h" #include "parquet/arrow/schema.h" #include "parquet/file_reader.h" @@ -602,8 +604,35 @@ Result FileReaderWrapper::CalculateFilteredRowRanges( return RowRanges::CreateSingle(row_count); } + auto page_index_reader = GetPageIndexReader(); + if (!page_index_reader) { + return RowRanges::CreateSingle(row_count); + } + std::set field_names; + PAIMON_RETURN_NOT_OK(PredicateUtils::GetAllNames(predicate, &field_names)); + std::vector predicate_columns; + for (const auto& name : field_names) { + auto it = column_name_to_index.find(name); + if (it == column_name_to_index.end()) { + return Status::Invalid( + fmt::format("column '{}' not found in column_name_to_index", name)); + } + predicate_columns.push_back(it->second); + } + if (predicate_columns.empty()) { + return RowRanges::CreateSingle(row_count); + } + + // Arrow otherwise reads the column-index envelope of every column, including + // potentially large min/max values in payload columns not used by the predicate. + const std::vector row_groups = {row_group_index}; + ScopeGuard clear_hint([&]() { page_index_reader->WillNotNeed(row_groups); }); + page_index_reader->WillNeed(row_groups, predicate_columns, + {/*column_index=*/true, /*offset_index=*/true}); + // Keep this restricted reader separate: projected payload columns still need + // their offset indexes when the data reader is initialized later. return ColumnIndexFilter::CalculateRowRanges(predicate, - GetRowGroupPageIndexReader(row_group_index), + page_index_reader->RowGroup(row_group_index), column_name_to_index, row_count); } PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::CalculateFilteredRowRanges") diff --git a/src/paimon/format/parquet/file_reader_wrapper_test.cpp b/src/paimon/format/parquet/file_reader_wrapper_test.cpp index 41dcde193..e23351078 100644 --- a/src/paimon/format/parquet/file_reader_wrapper_test.cpp +++ b/src/paimon/format/parquet/file_reader_wrapper_test.cpp @@ -46,9 +46,14 @@ #include "paimon/fs/file_system.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" +#include "paimon/predicate/literal.h" +#include "paimon/predicate/predicate_builder.h" #include "paimon/record_batch.h" #include "paimon/testing/utils/testharness.h" #include "parquet/arrow/reader.h" +#include "parquet/file_reader.h" +#include "parquet/metadata.h" +#include "parquet/page_index.h" #include "parquet/properties.h" namespace arrow { @@ -273,6 +278,39 @@ TEST_F(FileReaderWrapperTest, NullFileReader) { "file reader wrapper create failed. file reader is nullptr"); } +TEST_F(FileReaderWrapperTest, PredicateReadsOnlyItsPageIndexesAndKeepsPayloadReadable) { + std::string file_path = PathUtil::JoinPath(dir_->Str(), "predicate-index.parquet"); + PrepareParquetFile(file_path, /*row_count=*/1000, /*enable_page_index=*/true); + ASSERT_OK_AND_ASSIGN(std::shared_ptr input, fs_->Open(file_path)); + auto tracking_input = std::make_shared(input); + ASSERT_OK_AND_ASSIGN(auto reader, PrepareReaderWrapperOnStream(tracking_input)); + auto metadata = reader->GetFileReader()->parquet_reader()->metadata(); + auto index_ranges = + ::parquet::PageIndexReader::DeterminePageIndexRangesInRowGroup(*metadata->RowGroup(0), {1}); + ASSERT_TRUE(index_ranges.column_index.has_value()); + ASSERT_TRUE(index_ranges.offset_index.has_value()); + int64_t bytes_before = tracking_input->GetPositionalReadBytes(); + auto predicate = PredicateBuilder::Equal(1, "col2", FieldType::INT, Literal(25)); + ASSERT_OK_AND_ASSIGN(auto ranges, reader->CalculateFilteredRowRanges( + 0, predicate, {{"col1", 0}, {"col2", 1}, {"col3", 2}})); + ASSERT_EQ(10, ranges.RowCount()); + ASSERT_EQ(index_ranges.column_index->length + index_ranges.offset_index->length, + tracking_input->GetPositionalReadBytes() - bytes_before); + + // The predicate's read hint must not restrict subsequent payload offset-index reads. + ASSERT_OK(reader->PrepareForReading( + {TargetRowGroup(/*rg_index=*/0, /*is_partially_matched=*/true, ranges)}, {0, 1, 2})); + ASSERT_OK_AND_ASSIGN(auto batch, reader->Next()); + ASSERT_TRUE(batch); + auto expected = PrepareArray(PrepareArrowSchema().second, /*record_batch_size=*/10, + /*offset=*/20); + auto expected_batch = arrow::RecordBatch::FromStructArray(expected); + ASSERT_TRUE(expected_batch.ok()); + ASSERT_TRUE(batch->Equals(*expected_batch.ValueOrDie())); + ASSERT_OK_AND_ASSIGN(auto end, reader->Next()); + ASSERT_FALSE(end); +} + TEST_F(FileReaderWrapperTest, Simple) { std::string file_path = PathUtil::JoinPath(dir_->Str(), "test.parquet"); PrepareParquetFile(file_path, /*row_count=*/5500); From cd245156403e8c98979709a4ee84c57e7c209333 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 08:11:50 -0400 Subject: [PATCH 6/8] perf(snapshot): avoid directory probe for a valid latest hint --- src/paimon/core/utils/snapshot_manager.cpp | 10 +++-- .../core/utils/snapshot_manager_test.cpp | 43 +++++++++++++++++++ 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/src/paimon/core/utils/snapshot_manager.cpp b/src/paimon/core/utils/snapshot_manager.cpp index 3d38a6959..068025ddb 100644 --- a/src/paimon/core/utils/snapshot_manager.cpp +++ b/src/paimon/core/utils/snapshot_manager.cpp @@ -141,10 +141,6 @@ Result> SnapshotManager::FindEarliest( Result> SnapshotManager::FindLatest( const std::string& dir, const std::string& prefix, const std::function& path_func) const { - PAIMON_ASSIGN_OR_RAISE(bool is_exist, fs_->Exists(dir)); - if (!is_exist) { - return std::optional(); - } std::optional snapshot_id = ReadHint(LATEST, dir); if (snapshot_id != std::nullopt && snapshot_id.value() > 0) { int64_t next_snapshot = snapshot_id.value() + 1; @@ -155,6 +151,12 @@ Result> SnapshotManager::FindLatest( return snapshot_id; } } + // Reading a valid hint already establishes that its directory exists. Only + // check the directory when falling back to listing snapshots. + PAIMON_ASSIGN_OR_RAISE(bool is_exist, fs_->Exists(dir)); + if (!is_exist) { + return std::optional(); + } return FindByListFiles([](int64_t lhs, int64_t rhs) -> int64_t { return std::max(lhs, rhs); }, dir, prefix); } diff --git a/src/paimon/core/utils/snapshot_manager_test.cpp b/src/paimon/core/utils/snapshot_manager_test.cpp index a4794e0e9..ffe8a7d30 100644 --- a/src/paimon/core/utils/snapshot_manager_test.cpp +++ b/src/paimon/core/utils/snapshot_manager_test.cpp @@ -20,6 +20,7 @@ #include #include +#include #include "gtest/gtest.h" #include "paimon/common/utils/path_util.h" @@ -139,6 +140,48 @@ TEST(SnapshotManagerTest, TestPathNotExist) { ASSERT_EQ(snapshot, std::nullopt); } +TEST(SnapshotManagerTest, LatestHintAvoidsDirectoryProbeAndStillFindsNewSnapshots) { + class TrackingFileSystem : public LocalFileSystem { + public: + Result Exists(const std::string& path) const override { + exists_paths.push_back(path); + return LocalFileSystem::Exists(path); + } + mutable std::vector exists_paths; + }; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto fs = std::make_shared(); + SnapshotManager mgr(fs, dir->Str()); + ASSERT_OK(fs->Mkdirs(mgr.SnapshotDirectory())); + ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(1), "{}", true)); + ASSERT_OK(mgr.CommitLatestHint(1)); + fs->exists_paths.clear(); + + ASSERT_OK_AND_ASSIGN(std::optional latest, mgr.LatestSnapshotId()); + ASSERT_EQ(latest, 1); + ASSERT_EQ(fs->exists_paths, std::vector({mgr.SnapshotPath(2)})); + + // A commit can publish its snapshot before updating the hint. + ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(2), "{}", true)); + fs->exists_paths.clear(); + ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); + ASSERT_EQ(latest, 2); + ASSERT_EQ(fs->exists_paths, + std::vector({mgr.SnapshotPath(2), mgr.SnapshotDirectory()})); + + const std::string hint_path = PathUtil::JoinPath(mgr.SnapshotDirectory(), "LATEST"); + ASSERT_OK(fs->Delete(hint_path)); + fs->exists_paths.clear(); + ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); + ASSERT_EQ(latest, 2); + ASSERT_EQ(fs->exists_paths, std::vector({mgr.SnapshotDirectory()})); + + ASSERT_OK(fs->WriteFile(hint_path, "invalid", true)); + ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); + ASSERT_EQ(latest, 2); +} + TEST(SnapshotManagerTest, TestEarlierOrEqualTimeMillisExactMatch) { std::string test_data_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09"; auto file_system = std::make_shared(); From 526576e06a6fb6f7483dc35eb864401f68a4fbc1 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Sun, 6 Sep 2026 08:14:13 -0400 Subject: [PATCH 7/8] perf(snapshot): reuse the existing listing existence check --- src/paimon/core/utils/snapshot_manager.cpp | 8 ++------ src/paimon/core/utils/snapshot_manager_test.cpp | 12 ++++++++---- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/src/paimon/core/utils/snapshot_manager.cpp b/src/paimon/core/utils/snapshot_manager.cpp index 068025ddb..9211d9665 100644 --- a/src/paimon/core/utils/snapshot_manager.cpp +++ b/src/paimon/core/utils/snapshot_manager.cpp @@ -151,12 +151,8 @@ Result> SnapshotManager::FindLatest( return snapshot_id; } } - // Reading a valid hint already establishes that its directory exists. Only - // check the directory when falling back to listing snapshots. - PAIMON_ASSIGN_OR_RAISE(bool is_exist, fs_->Exists(dir)); - if (!is_exist) { - return std::optional(); - } + // A valid hint needs no parent-directory probe. The listing fallback checks + // directory existence itself, including tables without any snapshots yet. return FindByListFiles([](int64_t lhs, int64_t rhs) -> int64_t { return std::max(lhs, rhs); }, dir, prefix); } diff --git a/src/paimon/core/utils/snapshot_manager_test.cpp b/src/paimon/core/utils/snapshot_manager_test.cpp index ffe8a7d30..ec14582c2 100644 --- a/src/paimon/core/utils/snapshot_manager_test.cpp +++ b/src/paimon/core/utils/snapshot_manager_test.cpp @@ -18,6 +18,7 @@ #include "paimon/core/utils/snapshot_manager.h" +#include #include #include #include @@ -160,22 +161,25 @@ TEST(SnapshotManagerTest, LatestHintAvoidsDirectoryProbeAndStillFindsNewSnapshot ASSERT_OK_AND_ASSIGN(std::optional latest, mgr.LatestSnapshotId()); ASSERT_EQ(latest, 1); - ASSERT_EQ(fs->exists_paths, std::vector({mgr.SnapshotPath(2)})); + ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()), + 0); + ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotPath(2)), 1); // A commit can publish its snapshot before updating the hint. ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(2), "{}", true)); fs->exists_paths.clear(); ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); ASSERT_EQ(latest, 2); - ASSERT_EQ(fs->exists_paths, - std::vector({mgr.SnapshotPath(2), mgr.SnapshotDirectory()})); + ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()), + 1); const std::string hint_path = PathUtil::JoinPath(mgr.SnapshotDirectory(), "LATEST"); ASSERT_OK(fs->Delete(hint_path)); fs->exists_paths.clear(); ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); ASSERT_EQ(latest, 2); - ASSERT_EQ(fs->exists_paths, std::vector({mgr.SnapshotDirectory()})); + ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()), + 1); ASSERT_OK(fs->WriteFile(hint_path, "invalid", true)); ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); From 3ce3aba98b69806f7f51a03e79390c48af155945 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Mon, 7 Sep 2026 22:31:31 -0400 Subject: [PATCH 8/8] test: decouple snapshot hint checks from filesystem calls --- .../core/utils/snapshot_manager_test.cpp | 23 ++----------------- 1 file changed, 2 insertions(+), 21 deletions(-) diff --git a/src/paimon/core/utils/snapshot_manager_test.cpp b/src/paimon/core/utils/snapshot_manager_test.cpp index ec14582c2..d1bc83091 100644 --- a/src/paimon/core/utils/snapshot_manager_test.cpp +++ b/src/paimon/core/utils/snapshot_manager_test.cpp @@ -18,7 +18,6 @@ #include "paimon/core/utils/snapshot_manager.h" -#include #include #include #include @@ -141,45 +140,27 @@ TEST(SnapshotManagerTest, TestPathNotExist) { ASSERT_EQ(snapshot, std::nullopt); } -TEST(SnapshotManagerTest, LatestHintAvoidsDirectoryProbeAndStillFindsNewSnapshots) { - class TrackingFileSystem : public LocalFileSystem { - public: - Result Exists(const std::string& path) const override { - exists_paths.push_back(path); - return LocalFileSystem::Exists(path); - } - mutable std::vector exists_paths; - }; +TEST(SnapshotManagerTest, LatestSnapshotWithStaleMissingOrInvalidHint) { auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - auto fs = std::make_shared(); + auto fs = std::make_shared(); SnapshotManager mgr(fs, dir->Str()); ASSERT_OK(fs->Mkdirs(mgr.SnapshotDirectory())); ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(1), "{}", true)); ASSERT_OK(mgr.CommitLatestHint(1)); - fs->exists_paths.clear(); ASSERT_OK_AND_ASSIGN(std::optional latest, mgr.LatestSnapshotId()); ASSERT_EQ(latest, 1); - ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()), - 0); - ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotPath(2)), 1); // A commit can publish its snapshot before updating the hint. ASSERT_OK(fs->WriteFile(mgr.SnapshotPath(2), "{}", true)); - fs->exists_paths.clear(); ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); ASSERT_EQ(latest, 2); - ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()), - 1); const std::string hint_path = PathUtil::JoinPath(mgr.SnapshotDirectory(), "LATEST"); ASSERT_OK(fs->Delete(hint_path)); - fs->exists_paths.clear(); ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId()); ASSERT_EQ(latest, 2); - ASSERT_EQ(std::count(fs->exists_paths.begin(), fs->exists_paths.end(), mgr.SnapshotDirectory()), - 1); ASSERT_OK(fs->WriteFile(hint_path, "invalid", true)); ASSERT_OK_AND_ASSIGN(latest, mgr.LatestSnapshotId());