From 8944a254dd2f2cd70e490245f764e1c1fc32950d Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Wed, 8 Jul 2026 10:58:01 +0800 Subject: [PATCH 1/9] Fix interfaces.h hunk: correct count + trailing context --- cmake_modules/arrow.diff | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff index ae7752097..70d33c7b5 100644 --- a/cmake_modules/arrow.diff +++ b/cmake_modules/arrow.diff @@ -431,9 +431,10 @@ diff --git a/cpp/cmake_modules/BuildUtils.cmake b/cpp/cmake_modules/BuildUtils.c diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h --- a/cpp/src/arrow/io/interfaces.h +++ b/cpp/src/arrow/io/interfaces.h -@@ -211,7 +211,7 @@ +@@ -210,5 +210,5 @@ /// \brief Advance or skip stream indicated number of bytes /// \param[in] nbytes the number to move forward /// \return Status - Status Advance(int64_t nbytes); + virtual Status Advance(int64_t nbytes); + \ No newline at end of file From e0a7b12581a25c3e237dd431a021349f218c9e6e Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Wed, 8 Jul 2026 11:27:57 +0800 Subject: [PATCH 2/9] feat: support page index filter for nested column type --- .../format/parquet/file_reader_wrapper.cpp | 29 ++-- .../page_filtered_row_group_reader.cpp | 164 ++++++++++-------- .../parquet/page_filtered_row_group_reader.h | 31 ++-- .../page_filtered_row_group_reader_test.cpp | 16 +- .../parquet/parquet_file_batch_reader.cpp | 9 +- 5 files changed, 132 insertions(+), 117 deletions(-) diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index e7d6bf606..66e2e11e4 100644 --- a/src/paimon/format/parquet/file_reader_wrapper.cpp +++ b/src/paimon/format/parquet/file_reader_wrapper.cpp @@ -29,6 +29,7 @@ #include "paimon/format/parquet/parquet_format_defs.h" #include "paimon/macros.h" #include "parquet/arrow/reader.h" +#include "parquet/arrow/schema.h" #include "parquet/file_reader.h" #include "parquet/metadata.h" #include "parquet/page_index.h" @@ -234,8 +235,9 @@ Result> FileReaderWrapper::NextPageFiltered( PAIMON_ASSIGN_OR_RAISE( current_page_filtered_reader_, PageFilteredRowGroupReader::ReadFilteredRowGroup( - file_reader_->parquet_reader(), target_rg, target_column_indices_, - page_filtered_read_schema_, file_reader_->properties().cache_options(), + file_reader_.get(), file_reader_->parquet_reader(), target_rg, + target_column_indices_, page_filtered_read_schema_, + file_reader_->properties().cache_options(), pre_buffered, page_ranges, max_chunksize, pool_)); current_filtered_row_ranges_ = target_rg.row_ranges; current_filtered_rg_start_ = all_row_group_ranges_[rg_id].first; @@ -343,20 +345,17 @@ Status FileReaderWrapper::BuildPageFilteredSchema(const std::vector& co if (page_filtered_read_schema_) { return Status::OK(); } - std::shared_ptr schema; - PAIMON_RETURN_NOT_OK_FROM_ARROW(file_reader_->GetSchema(&schema)); - auto parquet_schema = file_reader_->parquet_reader()->metadata()->schema(); + // Use SchemaManifest to build schema with proper nested field structure. + // GetFieldIndices maps leaf column indices to top-level field indices, + // correctly handling nested types (List, Struct, Map). + const auto& manifest = file_reader_->manifest(); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::vector field_indices, + manifest.GetFieldIndices( + std::vector(column_indices.begin(), column_indices.end()))); std::vector> fields; - for (int32_t col_idx : column_indices) { - const std::string& col_name = parquet_schema->Column(col_idx)->name(); - auto field = schema->GetFieldByName(col_name); - if (!field) { - return Status::Invalid(fmt::format( - "PrepareForReading: Parquet column {} ('{}') has no matching Arrow field in " - "file schema", - col_idx, col_name)); - } - fields.push_back(field); + for (int field_idx : field_indices) { + fields.push_back(manifest.schema_fields[field_idx].field); } page_filtered_read_schema_ = arrow::schema(fields); return Status::OK(); 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 9c87438b8..9d9e41981 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp @@ -27,7 +27,9 @@ #include "arrow/util/future.h" #include "fmt/format.h" #include "paimon/common/utils/arrow/status_utils.h" +#include "parquet/arrow/reader.h" #include "parquet/arrow/reader_internal.h" +#include "parquet/arrow/schema.h" #include "parquet/metadata.h" #include "parquet/schema.h" @@ -123,87 +125,37 @@ std::pair PageFilteredRowGroupReader::ComputeCompressedRowRa } Status PageFilteredRowGroupReader::ExecuteSkipReadPattern( - const std::shared_ptr<::parquet::internal::RecordReader>& record_reader, - const RowRanges& ranges, int64_t total_row_count, int32_t row_group_index, - int32_t column_index) { + ::parquet::arrow::ColumnReader* column_reader, const RowRanges& ranges, + int64_t total_row_count, int32_t row_group_index, int32_t field_index) { int64_t current_row = 0; for (const auto& range : ranges.GetRanges()) { if (range.from > current_row) { int64_t to_skip = range.from - current_row; - int64_t skipped = record_reader->SkipRecords(to_skip); + int64_t skipped = column_reader->SkipRecords(to_skip); if (skipped != to_skip) { return Status::Invalid(fmt::format( "PageFilteredRowGroupReader: expected to skip {} records but skipped {} " - "(row_group={}, column={})", - to_skip, skipped, row_group_index, column_index)); + "(row_group={}, field={})", + to_skip, skipped, row_group_index, field_index)); } current_row = range.from; } int64_t to_read = range.Count(); - int64_t read = record_reader->ReadRecords(to_read); + int64_t read = column_reader->ReadRecords(to_read); if (read != to_read) { - return Status::Invalid( - fmt::format("PageFilteredRowGroupReader: expected to read {} records but read {} " - "(row_group={}, column={}, range=[{},{}])", - to_read, read, row_group_index, column_index, range.from, range.to)); + return Status::Invalid(fmt::format( + "PageFilteredRowGroupReader: expected to read {} records but read {} " + "(row_group={}, field={}, range=[{},{}])", + to_read, read, row_group_index, field_index, range.from, range.to)); } current_row += to_read; } if (current_row < total_row_count) { - record_reader->SkipRecords(total_row_count - current_row); + column_reader->SkipRecords(total_row_count - current_row); } return Status::OK(); } -Result> PageFilteredRowGroupReader::ReadFilteredColumn( - const std::shared_ptr<::parquet::RowGroupReader>& row_group_reader, - ::parquet::ParquetFileReader* parquet_reader, - const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - int32_t row_group_index, int32_t column_index, const RowRanges& row_ranges, - const std::shared_ptr& field, int64_t row_group_row_count, - std::shared_ptr<::arrow::MemoryPool> pool) { - auto file_metadata = parquet_reader->metadata(); - const auto* col_descriptor = file_metadata->schema()->Column(column_index); - - // Try to get OffsetIndex for I/O-level page skipping - RowRanges effective_ranges = row_ranges; - int64_t effective_row_count = row_group_row_count; - - std::shared_ptr<::parquet::OffsetIndex> offset_index; - if (rg_page_index_reader) { - offset_index = rg_page_index_reader->GetOffsetIndex(column_index); - } - - auto page_reader = row_group_reader->GetColumnPageReader(column_index); - - if (offset_index) { - // Set data_page_filter for I/O-level page skipping - page_reader->set_data_page_filter( - MakePageFilter(row_ranges, offset_index, row_group_row_count)); - // Compute compressed RowRanges for the decode-level skip/read pattern - auto [compressed_ranges, compressed_total] = - ComputeCompressedRowRanges(row_ranges, offset_index, row_group_row_count); - effective_ranges = std::move(compressed_ranges); - effective_row_count = compressed_total; - } - - // Create RecordReader - ::parquet::internal::LevelInfo leaf_info = - ::parquet::internal::LevelInfo::ComputeLevelInfo(col_descriptor); - auto record_reader = - ::parquet::internal::RecordReader::Make(col_descriptor, leaf_info, pool.get()); - record_reader->SetPageReader(std::move(page_reader)); - - PAIMON_RETURN_NOT_OK(ExecuteSkipReadPattern( - record_reader, effective_ranges, effective_row_count, row_group_index, column_index)); - - std::shared_ptr chunked_array; - PAIMON_RETURN_NOT_OK_FROM_ARROW(::parquet::arrow::TransferColumnData( - record_reader.get(), field, col_descriptor, pool.get(), &chunked_array)); - - return chunked_array; -} - Status PageFilteredRowGroupReader::WaitForPreBuffer( ::parquet::ParquetFileReader* parquet_reader, int32_t row_group_index, const std::vector& column_indices, const ::arrow::io::CacheOptions& cache_options, @@ -228,7 +180,72 @@ Status PageFilteredRowGroupReader::WaitForPreBuffer( return Status::OK(); } +Result> PageFilteredRowGroupReader::ReadFilteredField( + ::parquet::arrow::FileReader* arrow_file_reader, + const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, + int32_t row_group_index, int32_t field_index, + const std::vector& column_indices, const RowRanges& row_ranges, + int64_t row_group_row_count, std::shared_ptr<::arrow::MemoryPool> pool) { + const auto& manifest = arrow_file_reader->manifest(); + const auto& schema_field = manifest.schema_fields[field_index]; + bool is_flat = schema_field.is_leaf(); + + // For flat columns: compute compressed RowRanges if OffsetIndex is available. + // data_page_filter + compressed_ranges enable I/O-level page skipping. + // For nested fields: use original row_ranges, decode-level skipping only. + RowRanges effective_ranges = row_ranges; + int64_t effective_total = row_group_row_count; + if (is_flat && rg_page_index_reader) { + auto offset_index = rg_page_index_reader->GetOffsetIndex(schema_field.column_index); + if (offset_index) { + auto [compressed, total] = + ComputeCompressedRowRanges(row_ranges, offset_index, row_group_row_count); + effective_ranges = std::move(compressed); + effective_total = total; + } + } + + // Factory: set data_page_filter only for flat columns with OffsetIndex + auto factory = [row_group_index, is_flat, &rg_page_index_reader, &row_ranges, + row_group_row_count](int col_idx, ::parquet::ParquetFileReader* reader) + -> ::parquet::arrow::FileColumnIterator* { + auto* iter = new ::parquet::arrow::FileColumnIterator(col_idx, reader, {row_group_index}); + if (is_flat && rg_page_index_reader) { + auto offset_index = rg_page_index_reader->GetOffsetIndex(col_idx); + if (offset_index) { + iter->set_data_page_filter( + MakePageFilter(row_ranges, offset_index, row_group_row_count)); + } + } + return iter; + }; + + // Build reader tree with leaf column filtering + std::unique_ptr<::parquet::arrow::ColumnReader> column_reader; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow_file_reader->GetColumn( + field_index, column_indices, factory, &column_reader)); + + if (!column_reader) { + return std::shared_ptr(); + } + + // Phase 1: Prepare for reading + PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->BeginRead(effective_total)); + + // Phase 2: Execute skip/read pattern + PAIMON_RETURN_NOT_OK(ExecuteSkipReadPattern(column_reader.get(), effective_ranges, + effective_total, row_group_index, field_index)); + + // Phase 3: Build the Arrow array (TransferColumnData for leaves + assemble for nested) + std::shared_ptr chunked_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW( + column_reader->BuildArray(effective_total, &chunked_array)); + + return chunked_array; +} + Result> PageFilteredRowGroupReader::ReadFilteredRowGroup( + ::parquet::arrow::FileReader* arrow_file_reader, ::parquet::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, const std::vector& column_indices, const std::shared_ptr& arrow_schema, const ::arrow::io::CacheOptions& cache_options, bool pre_buffered, @@ -248,7 +265,6 @@ Result> PageFilteredRowGroupReader::Re PAIMON_RETURN_NOT_OK(WaitForPreBuffer(parquet_reader, row_group_index, column_indices, cache_options, pre_buffered, page_ranges, pool)); - auto row_group_reader = parquet_reader->RowGroup(row_group_index); auto rg_metadata = parquet_reader->metadata()->RowGroup(row_group_index); int64_t row_group_row_count = rg_metadata->num_rows(); @@ -260,23 +276,27 @@ Result> PageFilteredRowGroupReader::Re rg_page_index_reader = page_index_reader->RowGroup(row_group_index); } - // Read each column with page filtering + // Group leaf column indices by top-level field using SchemaManifest + const auto& manifest = arrow_file_reader->manifest(); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::vector field_indices, + manifest.GetFieldIndices(std::vector(column_indices.begin(), column_indices.end()))); + + // Read each field with page filtering (unified path for flat and nested) std::vector> columns; - columns.reserve(column_indices.size()); + columns.reserve(field_indices.size()); - for (size_t i = 0; i < column_indices.size(); ++i) { + for (int field_idx : field_indices) { PAIMON_ASSIGN_OR_RAISE( std::shared_ptr chunked_array, - ReadFilteredColumn(row_group_reader, parquet_reader, rg_page_index_reader, - row_group_index, column_indices[i], row_ranges, - arrow_schema->field(static_cast(i)), row_group_row_count, - pool)); + ReadFilteredField(arrow_file_reader, rg_page_index_reader, row_group_index, + field_idx, column_indices, row_ranges, row_group_row_count, pool)); - if (chunked_array->length() != expected_rows) { + if (chunked_array && chunked_array->length() != expected_rows) { return Status::Invalid(fmt::format( - "PageFilteredRowGroupReader: column {} produced {} rows but expected {} " + "PageFilteredRowGroupReader: field {} produced {} rows but expected {} " "(row_group={})", - column_indices[i], chunked_array->length(), expected_rows, row_group_index)); + field_idx, chunked_array->length(), expected_rows, row_group_index)); } columns.push_back(std::move(chunked_array)); diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader.h b/src/paimon/format/parquet/page_filtered_row_group_reader.h index 5092bb5ca..c0ee79731 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.h +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.h @@ -28,6 +28,7 @@ #include "arrow/type.h" #include "paimon/format/parquet/row_ranges.h" #include "paimon/result.h" +#include "parquet/arrow/reader.h" #include "parquet/column_reader.h" #include "parquet/file_reader.h" #include "parquet/page_index.h" @@ -44,6 +45,7 @@ class PageFilteredRowGroupReader { ~PageFilteredRowGroupReader() = delete; /// Read a row group with page-level filtering. + /// @param arrow_file_reader The Arrow FileReader for ColumnReader tree creation /// @param parquet_reader The underlying ParquetFileReader /// @param target_row_group Target row group with index and row ranges /// @param column_indices Leaf column indices to read @@ -56,6 +58,7 @@ class PageFilteredRowGroupReader { /// @param max_chunksize Per-batch row cap for the returned reader. /// @return A RecordBatchReader streaming the filtered rows. static Result> ReadFilteredRowGroup( + ::parquet::arrow::FileReader* arrow_file_reader, ::parquet::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, const std::vector& column_indices, const std::shared_ptr& arrow_schema, @@ -86,30 +89,30 @@ class PageFilteredRowGroupReader { const std::vector<::arrow::io::ReadRange>& page_ranges, std::shared_ptr<::arrow::MemoryPool> pool); - /// Execute the skip/read pattern on a RecordReader based on RowRanges. - static Status ExecuteSkipReadPattern( - const std::shared_ptr<::parquet::internal::RecordReader>& record_reader, - const RowRanges& ranges, int64_t total_row_count, int32_t row_group_index, - int32_t column_index); + /// Execute the skip/read pattern on a ColumnReader based on RowRanges. + static Status ExecuteSkipReadPattern(::parquet::arrow::ColumnReader* column_reader, + const RowRanges& ranges, int64_t total_row_count, + int32_t row_group_index, int32_t field_index); /// Create a data_page_filter callback for a column based on RowRanges + OffsetIndex. static std::function MakePageFilter( const RowRanges& row_ranges, const std::shared_ptr<::parquet::OffsetIndex>& offset_index, int64_t row_group_row_count); - /// Read a single column using skip/read pattern driven by RowRanges. - static Result> ReadFilteredColumn( - const std::shared_ptr<::parquet::RowGroupReader>& row_group_reader, - ::parquet::ParquetFileReader* parquet_reader, - const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - int32_t row_group_index, int32_t column_index, const RowRanges& row_ranges, - const std::shared_ptr& field, int64_t row_group_row_count, - std::shared_ptr<::arrow::MemoryPool> pool); - /// Compute compressed RowRanges after data_page_filter skips non-matching pages. static std::pair ComputeCompressedRowRanges( const RowRanges& original_ranges, const std::shared_ptr<::parquet::OffsetIndex>& offset_index, int64_t row_group_row_count); + + /// Read a field (flat or nested) using ColumnReader tree. + /// For flat columns: sets data_page_filter + uses compressed RowRanges (I/O + decode skipping). + /// For nested fields: decode-level skipping only via SkipRecords/ReadRecords. + static Result> ReadFilteredField( + ::parquet::arrow::FileReader* arrow_file_reader, + const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, + int32_t row_group_index, int32_t field_index, + const std::vector& column_indices, const RowRanges& row_ranges, + int64_t row_group_row_count, std::shared_ptr<::arrow::MemoryPool> pool); }; } // namespace paimon::parquet diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp index d6bb36ceb..44ee8d28e 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp @@ -932,10 +932,10 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnRowGroupFilter) { // Should get rows 50-99 = 50 rows ASSERT_TRUE(result); - ASSERT_EQ(50, result->length()); + ASSERT_EQ(30, result->length()); // Build expected result: rows 50-99 from the original data - auto expected = data->Slice(50, 50); + auto expected = data->Slice(70 , 30); ASSERT_TRUE(expected->Equals(result->chunk(0))); } @@ -1052,10 +1052,10 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedListColumnRowGroupFilter) { ReadWithPredicateImpl(file_name, read_schema, predicate, &result); ASSERT_TRUE(result); - ASSERT_EQ(50, result->length()); + ASSERT_EQ(30, result->length()); // Build expected result: rows 50-99 from the original data - auto expected = data->Slice(50, 50); + auto expected = data->Slice(70, 30); ASSERT_TRUE(expected->Equals(result->chunk(0))); } @@ -1080,10 +1080,10 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedMapColumnRowGroupFilter) { ReadWithPredicateImpl(file_name, read_schema, predicate, &result); ASSERT_TRUE(result); - ASSERT_EQ(50, result->length()); + ASSERT_EQ(30, result->length()); // Build expected result: rows 50-99 from the original data - auto expected = data->Slice(50, 50); + auto expected = data->Slice(70, 30); ASSERT_TRUE(expected->Equals(result->chunk(0))); } @@ -1137,10 +1137,10 @@ TEST_F(PageFilteredRowGroupReaderTest, MultipleAdjacentNestedColumns) { ReadWithPredicateImpl(file_name, read_schema, predicate, &result); ASSERT_TRUE(result); - ASSERT_EQ(50, result->length()); + ASSERT_EQ(30, result->length()); // Build expected result: rows 50-99 from the original data - auto expected = data->Slice(50, 50); + auto expected = data->Slice(70, 30); ASSERT_TRUE(expected->Equals(result->chunk(0))); } diff --git a/src/paimon/format/parquet/parquet_file_batch_reader.cpp b/src/paimon/format/parquet/parquet_file_batch_reader.cpp index 43f4c64c8..11edb868d 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader.cpp @@ -134,13 +134,6 @@ Status ParquetFileBatchReader::SetReadSchema( arrow::ImportSchema(schema)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr file_schema, reader_->GetSchema()); - bool has_nested_field = false; - for (const auto& field : read_schema->fields()) { - if (ArrowSchemaValidator::IsNestedType(field->type())) { - has_nested_field = true; - break; - } - } // Recursively match read_schema against file_schema by field names. // STRUCT supports sub-field projection; LIST/MAP require exact type match. @@ -178,7 +171,7 @@ Status ParquetFileBatchReader::SetReadSchema( DEFAULT_PARQUET_READ_ENABLE_PAGE_INDEX_FILTER)); // walkaround: page index filter does not support nested fields for now, skip page index // filter if there is any nested field in the schema - if (enable_page_index_filter && !has_nested_field) { + if (enable_page_index_filter) { // Build column name to index map for page-level filtering. // For leaf columns, indices[0] is the correct leaf column index in Parquet. // For nested types (struct/list/map), FlattenSchema produces multiple leaf indices, From 45058ce27a703a3432a888aad759b7bb847f380d Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Wed, 8 Jul 2026 14:11:07 +0800 Subject: [PATCH 3/9] test: add test cases --- .../page_filtered_row_group_reader_test.cpp | 217 ++++++++++-------- 1 file changed, 125 insertions(+), 92 deletions(-) diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp index 44ee8d28e..68a6abede 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp @@ -44,6 +44,7 @@ #include "paimon/status.h" #include "paimon/testing/utils/read_result_collector.h" #include "paimon/testing/utils/testharness.h" +#include "paimon/utils/roaring_bitmap32.h" #include "parquet/arrow/reader.h" #include "parquet/file_reader.h" #include "parquet/properties.h" @@ -873,34 +874,32 @@ TEST_F(PageFilteredRowGroupReaderTest, ComputePageRangesWithDictionaryEncoding) auto partial_concat = arrow::Concatenate(result_partial->chunks()).ValueOrDie(); ASSERT_TRUE(partial_concat->Equals(expected_struct)); } -/// Helper: build a StructArray with a top-level int32 "id" column and a nested struct column -/// "info" containing two int32 fields: "x" and "y". -/// id[i] = i, info.x[i] = i * 100, info.y[i] = i * 100 + 1, for i in [0, N). -/// -/// Arrow schema: { id: int32, info: struct } -/// Parquet leaf columns: [id (index 0), info.x (index 1), info.y (index 2)] -static std::shared_ptr MakeNestedStructData(int32_t num_rows) { - arrow::Int32Builder id_builder, x_builder, y_builder; +/// Helper: build an Int32Array with sequential values 0..N-1. +static std::shared_ptr MakeIdColumn(int32_t num_rows) { + arrow::Int32Builder id_builder; EXPECT_TRUE(id_builder.Reserve(num_rows).ok()); + for (int32_t i = 0; i < num_rows; ++i) { + id_builder.UnsafeAppend(i); + } + return id_builder.Finish().ValueOrDie(); +} + +/// Helper: build a struct array (without id column). +/// x[i] = i * 100, y[i] = i * 100 + 1, for i in [0, N). +static std::shared_ptr MakeNestedStructData(int32_t num_rows) { + arrow::Int32Builder x_builder, y_builder; EXPECT_TRUE(x_builder.Reserve(num_rows).ok()); EXPECT_TRUE(y_builder.Reserve(num_rows).ok()); for (int32_t i = 0; i < num_rows; ++i) { - id_builder.UnsafeAppend(i); x_builder.UnsafeAppend(i * 100); y_builder.UnsafeAppend(i * 100 + 1); } - auto id_array = id_builder.Finish().ValueOrDie(); auto x_array = x_builder.Finish().ValueOrDie(); auto y_array = y_builder.Finish().ValueOrDie(); auto field_x = arrow::field("x", arrow::int32()); auto field_y = arrow::field("y", arrow::int32()); - auto inner_struct = - arrow::StructArray::Make({x_array, y_array}, {field_x, field_y}).ValueOrDie(); - - auto field_id = arrow::field("id", arrow::int32()); - auto field_info = arrow::field("info", arrow::struct_({field_x, field_y})); - return arrow::StructArray::Make({id_array, inner_struct}, {field_id, field_info}).ValueOrDie(); + return arrow::StructArray::Make({x_array, y_array}, {field_x, field_y}).ValueOrDie(); } /// Test: rowgroup-level filtering on a file with nested struct columns. @@ -916,13 +915,19 @@ static std::shared_ptr MakeNestedStructData(int32_t num_rows /// The read schema requests both "id" and "info" columns. TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnRowGroupFilter) { std::string file_name = dir_->Str() + "/nested_struct_filter.parquet"; - auto data = MakeNestedStructData(100); - WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); auto field_x = arrow::field("x", arrow::int32()); auto field_y = arrow::field("y", arrow::int32()); - auto read_schema = arrow::schema({arrow::field("id", arrow::int32()), - arrow::field("info", arrow::struct_({field_x, field_y}))}); + auto field_id = arrow::field("id", arrow::int32()); + auto field_info = arrow::field("info", arrow::struct_({field_x, field_y})); + + auto id_array = MakeIdColumn(100); + auto info_array = MakeNestedStructData(100); + auto data = arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}) + .ValueOrDie(); + WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); + + auto read_schema = arrow::schema({field_id, field_info}); auto predicate = PredicateBuilder::GreaterOrEqual( /*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70)); @@ -935,7 +940,7 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnRowGroupFilter) { ASSERT_EQ(30, result->length()); // Build expected result: rows 50-99 from the original data - auto expected = data->Slice(70 , 30); + auto expected = data->Slice(70, 30); ASSERT_TRUE(expected->Equals(result->chunk(0))); } @@ -949,13 +954,18 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnRowGroupFilter) { /// Predicate on "id": id >= 70. TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnOnlyReadIdField) { std::string file_name = dir_->Str() + "/nested_struct_only_nested.parquet"; - auto data = MakeNestedStructData(100); - WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); - auto field_id = arrow::field("id", arrow::int32()); auto field_x = arrow::field("x", arrow::int32()); auto field_y = arrow::field("y", arrow::int32()); + auto field_id = arrow::field("id", arrow::int32()); auto field_info = arrow::field("info", arrow::struct_({field_x, field_y})); + + auto id_array = MakeIdColumn(100); + auto info_array = MakeNestedStructData(100); + auto data = arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}) + .ValueOrDie(); + WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); + // Read "id" column only auto read_schema = arrow::schema({field_id}); @@ -975,19 +985,9 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnOnlyReadIdField) { ASSERT_TRUE(data->field(0)->Slice(70, 30)->Equals(result_struct->field(0))); } -/// Helper: build a StructArray with an int32 "id" column and a list "tags" column. -/// id[i] = i, tags[i] = [i*10, i*10+1], for i in [0, N). -/// -/// Arrow schema: { id: int32, tags: list } -/// Parquet leaf columns: [id (index 0), tags.item (index 1)] -static std::shared_ptr MakeListColumnData(int32_t num_rows) { - arrow::Int32Builder id_builder; - EXPECT_TRUE(id_builder.Reserve(num_rows).ok()); - for (int32_t i = 0; i < num_rows; ++i) { - id_builder.UnsafeAppend(i); - } - auto id_array = id_builder.Finish().ValueOrDie(); - +/// Helper: build a list array (without id column). +/// tags[i] = [i*10, i*10+1], for i in [0, N). +static std::shared_ptr MakeListColumnData(int32_t num_rows) { auto value_builder = std::make_shared(); arrow::ListBuilder list_builder(arrow::default_memory_pool(), value_builder); for (int32_t i = 0; i < num_rows; ++i) { @@ -995,26 +995,12 @@ static std::shared_ptr MakeListColumnData(int32_t num_rows) EXPECT_TRUE(value_builder->Append(i * 10).ok()); EXPECT_TRUE(value_builder->Append(i * 10 + 1).ok()); } - auto list_array = list_builder.Finish().ValueOrDie(); - - auto field_id = arrow::field("id", arrow::int32()); - auto field_tags = arrow::field("tags", arrow::list(arrow::field("item", arrow::int32()))); - return arrow::StructArray::Make({id_array, list_array}, {field_id, field_tags}).ValueOrDie(); + return list_builder.Finish().ValueOrDie(); } -/// Helper: build a StructArray with an int32 "id" column and a map "props" column. -/// id[i] = i, props[i] = {"k_i": i * 100}, for i in [0, N). -/// -/// Arrow schema: { id: int32, props: map } -/// Parquet leaf columns: [id (index 0), props.key (index 1), props.value (index 2)] -static std::shared_ptr MakeMapColumnData(int32_t num_rows) { - arrow::Int32Builder id_builder; - EXPECT_TRUE(id_builder.Reserve(num_rows).ok()); - for (int32_t i = 0; i < num_rows; ++i) { - id_builder.UnsafeAppend(i); - } - auto id_array = id_builder.Finish().ValueOrDie(); - +/// Helper: build a map array (without id column). +/// props[i] = {"k_i": i * 100}, for i in [0, N). +static std::shared_ptr MakeMapColumnData(int32_t num_rows) { auto key_builder = std::make_shared(); auto value_builder = std::make_shared(); arrow::MapBuilder map_builder(arrow::default_memory_pool(), key_builder, value_builder); @@ -1024,11 +1010,7 @@ static std::shared_ptr MakeMapColumnData(int32_t num_rows) { EXPECT_TRUE(key_builder->Append(key).ok()); EXPECT_TRUE(value_builder->Append(i * 100).ok()); } - auto map_array = map_builder.Finish().ValueOrDie(); - - auto field_id = arrow::field("id", arrow::int32()); - auto field_props = arrow::field("props", arrow::map(arrow::utf8(), arrow::int32())); - return arrow::StructArray::Make({id_array, map_array}, {field_id, field_props}).ValueOrDie(); + return map_builder.Finish().ValueOrDie(); } /// Test: rowgroup-level filtering on a file with a list column. @@ -1038,12 +1020,17 @@ static std::shared_ptr MakeMapColumnData(int32_t num_rows) { /// Predicate: id >= 70 → row groups 0 skipped, row groups 1 read → 50 rows expected. TEST_F(PageFilteredRowGroupReaderTest, NestedListColumnRowGroupFilter) { std::string file_name = dir_->Str() + "/nested_list_filter.parquet"; - auto data = MakeListColumnData(100); + + auto field_id = arrow::field("id", arrow::int32()); + auto field_tags = arrow::field("tags", arrow::list(arrow::field("item", arrow::int32()))); + + auto id_array = MakeIdColumn(100); + auto tags_array = MakeListColumnData(100); + auto data = arrow::StructArray::Make({id_array, tags_array}, {field_id, field_tags}) + .ValueOrDie(); WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); - auto read_schema = - arrow::schema({arrow::field("id", arrow::int32()), - arrow::field("tags", arrow::list(arrow::field("item", arrow::int32())))}); + auto read_schema = arrow::schema({field_id, field_tags}); auto predicate = PredicateBuilder::GreaterOrEqual( /*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70)); @@ -1066,12 +1053,17 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedListColumnRowGroupFilter) { /// Predicate: id >= 70 → row groups 0 skipped, row groups 1 read → 50 rows expected. TEST_F(PageFilteredRowGroupReaderTest, NestedMapColumnRowGroupFilter) { std::string file_name = dir_->Str() + "/nested_map_filter.parquet"; - auto data = MakeMapColumnData(100); + + auto field_id = arrow::field("id", arrow::int32()); + auto field_props = arrow::field("props", arrow::map(arrow::utf8(), arrow::int32())); + + auto id_array = MakeIdColumn(100); + auto props_array = MakeMapColumnData(100); + auto data = arrow::StructArray::Make({id_array, props_array}, {field_id, field_props}) + .ValueOrDie(); WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); - auto read_schema = - arrow::schema({arrow::field("id", arrow::int32()), - arrow::field("props", arrow::map(arrow::utf8(), arrow::int32()))}); + auto read_schema = arrow::schema({field_id, field_props}); auto predicate = PredicateBuilder::GreaterOrEqual( /*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70)); @@ -1095,35 +1087,16 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedMapColumnRowGroupFilter) { TEST_F(PageFilteredRowGroupReaderTest, MultipleAdjacentNestedColumns) { std::string file_name = dir_->Str() + "/multi_nested.parquet"; - // Build data with id, info (struct), tags (list) - arrow::Int32Builder id_builder, x_builder, y_builder; - ASSERT_TRUE(id_builder.Reserve(100).ok()); - ASSERT_TRUE(x_builder.Reserve(100).ok()); - ASSERT_TRUE(y_builder.Reserve(100).ok()); - auto value_builder = std::make_shared(); - arrow::ListBuilder list_builder(arrow::default_memory_pool(), value_builder); - - for (int32_t i = 0; i < 100; ++i) { - id_builder.UnsafeAppend(i); - x_builder.UnsafeAppend(i * 100); - y_builder.UnsafeAppend(i * 100 + 1); - ASSERT_TRUE(list_builder.Append().ok()); - ASSERT_TRUE(value_builder->Append(i * 10).ok()); - } - auto id_array = id_builder.Finish().ValueOrDie(); - auto x_array = x_builder.Finish().ValueOrDie(); - auto y_array = y_builder.Finish().ValueOrDie(); - auto list_array = list_builder.Finish().ValueOrDie(); - auto field_x = arrow::field("x", arrow::int32()); auto field_y = arrow::field("y", arrow::int32()); - auto inner_struct = - arrow::StructArray::Make({x_array, y_array}, {field_x, field_y}).ValueOrDie(); - auto field_id = arrow::field("id", arrow::int32()); auto field_info = arrow::field("info", arrow::struct_({field_x, field_y})); auto field_tags = arrow::field("tags", arrow::list(arrow::field("item", arrow::int32()))); - auto data = arrow::StructArray::Make({id_array, inner_struct, list_array}, + + auto id_array = MakeIdColumn(100); + auto info_array = MakeNestedStructData(100); + auto tags_array = MakeListColumnData(100); + auto data = arrow::StructArray::Make({id_array, info_array, tags_array}, {field_id, field_info, field_tags}) .ValueOrDie(); @@ -1144,4 +1117,64 @@ TEST_F(PageFilteredRowGroupReaderTest, MultipleAdjacentNestedColumns) { ASSERT_TRUE(expected->Equals(result->chunk(0))); } +/// Test: predicate pushdown with all nested column types (struct, list, map). +/// +/// Schema: { id: int32, info: struct, +/// tags: list, props: map } +/// 100 rows, 10 rows per page, 50 rows per row group → 2 row groups. +/// Predicate: id in [15, 29] or id in [80, 99] (Between is inclusive). +/// Read schema: full schema (all columns). +/// Page-level filtering (10 rows/page): +/// Between(15, 29) → pages 1-2 (rows 10-29) +/// Between(80, 99) → pages 8-9 (rows 80-99) +/// Total: 40 rows. +TEST_F(PageFilteredRowGroupReaderTest, MultipleNestedColumns) { + std::string file_name = dir_->Str() + "/multi_nested_columns.parquet"; + + auto field_x = arrow::field("x", arrow::int32()); + auto field_y = arrow::field("y", arrow::int32()); + auto field_id = arrow::field("id", arrow::int32()); + auto field_info = arrow::field("info", arrow::struct_({field_x, field_y})); + auto field_tags = arrow::field("tags", arrow::list(arrow::field("item", arrow::int32()))); + auto field_props = arrow::field("props", arrow::map(arrow::utf8(), arrow::int32())); + + // Build data with all nested column types using shared helpers + auto id_array = MakeIdColumn(100); + auto info_array = MakeNestedStructData(100); + auto tags_array = MakeListColumnData(100); + auto props_array = MakeMapColumnData(100); + auto data = arrow::StructArray::Make({id_array, info_array, tags_array, props_array}, + {field_id, field_info, field_tags, field_props}) + .ValueOrDie(); + + // Write: 10 rows per page, 50 rows per row group → 2 row groups + WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); + + // Read full schema + auto read_schema = arrow::schema({field_id, field_info, field_tags, field_props}); + + // predicate: id in [15, 29] or id in [80, 99] + ASSERT_OK_AND_ASSIGN( + auto predicate, + PredicateBuilder::Or({PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"id", + FieldType::INT, Literal(15), Literal(29)), + PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"id", + FieldType::INT, Literal(80), Literal(99))})); + + std::shared_ptr result; + ReadWithPredicateImpl(file_name, read_schema, predicate, &result, + /*batch_size=*/1024); + + // Page-level filtering (10 rows/page): + // Between(15, 29) → pages 1-2 (rows 10-29) + // Between(80, 99) → pages 8-9 (rows 80-99) + // Total: 40 rows + ASSERT_TRUE(result); + ASSERT_EQ(40, result->length()); + + auto expected = + arrow::ChunkedArray::Make({data->Slice(10, 20), data->Slice(80, 20)}).ValueOrDie(); + ASSERT_TRUE(result->Equals(expected)); +} + } // namespace paimon::parquet::test From b0aed2508ac69b9eb6447167c60b1f1af0189966 Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Wed, 8 Jul 2026 15:12:16 +0800 Subject: [PATCH 4/9] refractor --- .../format/parquet/file_reader_wrapper.cpp | 36 ++------- .../format/parquet/file_reader_wrapper.h | 8 -- .../page_filtered_row_group_reader.cpp | 73 ++++++++++++------- .../parquet/page_filtered_row_group_reader.h | 22 +++--- .../page_filtered_row_group_reader_test.cpp | 26 +++---- 5 files changed, 73 insertions(+), 92 deletions(-) diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index 66e2e11e4..555d14062 100644 --- a/src/paimon/format/parquet/file_reader_wrapper.cpp +++ b/src/paimon/format/parquet/file_reader_wrapper.cpp @@ -232,13 +232,11 @@ Result> FileReaderWrapper::NextPageFiltered( file_reader_->parquet_reader(), target_rg, target_column_indices_); bool pre_buffered = !prebuffered_ranges_.empty(); int64_t max_chunksize = batch_size_ > 0 ? batch_size_ : std::numeric_limits::max(); - PAIMON_ASSIGN_OR_RAISE( - current_page_filtered_reader_, - PageFilteredRowGroupReader::ReadFilteredRowGroup( - file_reader_.get(), file_reader_->parquet_reader(), target_rg, - target_column_indices_, page_filtered_read_schema_, - file_reader_->properties().cache_options(), - pre_buffered, page_ranges, max_chunksize, pool_)); + PAIMON_ASSIGN_OR_RAISE(current_page_filtered_reader_, + PageFilteredRowGroupReader::ReadFilteredRowGroup( + file_reader_.get(), target_rg, target_column_indices_, + file_reader_->properties().cache_options(), pre_buffered, + page_ranges, max_chunksize, pool_)); current_filtered_row_ranges_ = target_rg.row_ranges; current_filtered_rg_start_ = all_row_group_ranges_[rg_id].first; filtered_global_offset_ = 0; @@ -341,26 +339,6 @@ Status FileReaderWrapper::PrepareForReadingLazy( return Status::OK(); } -Status FileReaderWrapper::BuildPageFilteredSchema(const std::vector& column_indices) { - if (page_filtered_read_schema_) { - return Status::OK(); - } - // Use SchemaManifest to build schema with proper nested field structure. - // GetFieldIndices maps leaf column indices to top-level field indices, - // correctly handling nested types (List, Struct, Map). - const auto& manifest = file_reader_->manifest(); - PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( - std::vector field_indices, - manifest.GetFieldIndices( - std::vector(column_indices.begin(), column_indices.end()))); - std::vector> fields; - for (int field_idx : field_indices) { - fields.push_back(manifest.schema_fields[field_idx].field); - } - page_filtered_read_schema_ = arrow::schema(fields); - return Status::OK(); -} - std::vector<::arrow::io::ReadRange> FileReaderWrapper::CollectPreBufferRanges( const std::vector& column_indices) { std::vector<::arrow::io::ReadRange> ranges; @@ -409,7 +387,6 @@ Status FileReaderWrapper::PrepareForReading(const std::vector& t try { target_row_groups_ = target_row_groups; target_column_indices_ = column_indices; - page_filtered_read_schema_.reset(); // Partition into fully-matched and page-filtered row groups, skipping excluded ones. std::vector fully_matched_row_groups; @@ -425,9 +402,6 @@ Status FileReaderWrapper::PrepareForReading(const std::vector& t } bool has_partially_matched = fully_matched_row_groups.size() != active_count; - if (has_partially_matched) { - PAIMON_RETURN_NOT_OK(BuildPageFilteredSchema(column_indices)); - } WaitForPendingPreBuffer(); diff --git a/src/paimon/format/parquet/file_reader_wrapper.h b/src/paimon/format/parquet/file_reader_wrapper.h index 758ff703a..f9a09c674 100644 --- a/src/paimon/format/parquet/file_reader_wrapper.h +++ b/src/paimon/format/parquet/file_reader_wrapper.h @@ -157,9 +157,6 @@ class FileReaderWrapper { /// Read next batch from the fully-matched batch_reader_. Returns nullptr when exhausted. Result> NextFullyMatched(); - /// Build page_filtered_read_schema_ from the given column indices. No-op if already built. - Status BuildPageFilteredSchema(const std::vector& column_indices); - /// Collect all byte ranges that need pre-buffering (page-filtered + fully-matched). std::vector<::arrow::io::ReadRange> CollectPreBufferRanges( const std::vector& column_indices); @@ -193,11 +190,6 @@ class FileReaderWrapper { // Target row groups with row ranges for none page-level filtering and page-level filtering std::vector target_row_groups_; - // Arrow schema covering target_column_indices_, used when constructing the per-RG - // page-filtered reader. Cached in PrepareForReading because it's identical across - // all page-filtered RGs in a session. - std::shared_ptr page_filtered_read_schema_; - // Track pre-buffered ranges so we can wait on destruction std::vector<::arrow::io::ReadRange> prebuffered_ranges_; }; 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 9d9e41981..ff26d54bb 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp @@ -125,8 +125,8 @@ std::pair PageFilteredRowGroupReader::ComputeCompressedRowRa } Status PageFilteredRowGroupReader::ExecuteSkipReadPattern( - ::parquet::arrow::ColumnReader* column_reader, const RowRanges& ranges, - int64_t total_row_count, int32_t row_group_index, int32_t field_index) { + ::parquet::arrow::ColumnReader* column_reader, const RowRanges& ranges, int64_t total_row_count, + int32_t row_group_index, int32_t field_index) { int64_t current_row = 0; for (const auto& range : ranges.GetRanges()) { if (range.from > current_row) { @@ -143,10 +143,10 @@ Status PageFilteredRowGroupReader::ExecuteSkipReadPattern( int64_t to_read = range.Count(); int64_t read = column_reader->ReadRecords(to_read); if (read != to_read) { - return Status::Invalid(fmt::format( - "PageFilteredRowGroupReader: expected to read {} records but read {} " - "(row_group={}, field={}, range=[{},{}])", - to_read, read, row_group_index, field_index, range.from, range.to)); + return Status::Invalid( + fmt::format("PageFilteredRowGroupReader: expected to read {} records but read {} " + "(row_group={}, field={}, range=[{},{}])", + to_read, read, row_group_index, field_index, range.from, range.to)); } current_row += to_read; } @@ -183,9 +183,9 @@ Status PageFilteredRowGroupReader::WaitForPreBuffer( Result> PageFilteredRowGroupReader::ReadFilteredField( ::parquet::arrow::FileReader* arrow_file_reader, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - int32_t row_group_index, int32_t field_index, - const std::vector& column_indices, const RowRanges& row_ranges, - int64_t row_group_row_count, std::shared_ptr<::arrow::MemoryPool> pool) { + int32_t row_group_index, int32_t field_index, const std::vector& column_indices, + const RowRanges& row_ranges, int64_t row_group_row_count, + std::shared_ptr<::arrow::MemoryPool> pool) { const auto& manifest = arrow_file_reader->manifest(); const auto& schema_field = manifest.schema_fields[field_index]; bool is_flat = schema_field.is_leaf(); @@ -206,9 +206,10 @@ Result> PageFilteredRowGroupReader::ReadFil } // Factory: set data_page_filter only for flat columns with OffsetIndex - auto factory = [row_group_index, is_flat, &rg_page_index_reader, &row_ranges, - row_group_row_count](int col_idx, ::parquet::ParquetFileReader* reader) - -> ::parquet::arrow::FileColumnIterator* { + auto factory = + [row_group_index, is_flat, &rg_page_index_reader, &row_ranges, row_group_row_count]( + int col_idx, + ::parquet::ParquetFileReader* reader) -> ::parquet::arrow::FileColumnIterator* { auto* iter = new ::parquet::arrow::FileColumnIterator(col_idx, reader, {row_group_index}); if (is_flat && rg_page_index_reader) { auto offset_index = rg_page_index_reader->GetOffsetIndex(col_idx); @@ -222,8 +223,8 @@ Result> PageFilteredRowGroupReader::ReadFil // Build reader tree with leaf column filtering std::unique_ptr<::parquet::arrow::ColumnReader> column_reader; - PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow_file_reader->GetColumn( - field_index, column_indices, factory, &column_reader)); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow_file_reader->GetColumn(field_index, column_indices, factory, &column_reader)); if (!column_reader) { return std::shared_ptr(); @@ -238,19 +239,19 @@ Result> PageFilteredRowGroupReader::ReadFil // Phase 3: Build the Arrow array (TransferColumnData for leaves + assemble for nested) std::shared_ptr chunked_array; - PAIMON_RETURN_NOT_OK_FROM_ARROW( - column_reader->BuildArray(effective_total, &chunked_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->BuildArray(effective_total, &chunked_array)); return chunked_array; } Result> PageFilteredRowGroupReader::ReadFilteredRowGroup( - ::parquet::arrow::FileReader* arrow_file_reader, - ::parquet::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, - const std::vector& column_indices, const std::shared_ptr& arrow_schema, - const ::arrow::io::CacheOptions& cache_options, bool pre_buffered, - const std::vector<::arrow::io::ReadRange>& page_ranges, int64_t max_chunksize, - std::shared_ptr<::arrow::MemoryPool> pool) { + ::parquet::arrow::FileReader* arrow_file_reader, const TargetRowGroup& target_row_group, + const std::vector& column_indices, const ::arrow::io::CacheOptions& cache_options, + bool pre_buffered, const std::vector<::arrow::io::ReadRange>& page_ranges, + int64_t max_chunksize, std::shared_ptr<::arrow::MemoryPool> pool) { + auto parquet_reader = arrow_file_reader->parquet_reader(); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr arrow_schema, + BuildPageFilteredSchema(arrow_file_reader, column_indices)); const auto& row_ranges = target_row_group.row_ranges; int32_t row_group_index = target_row_group.row_group_index; @@ -289,14 +290,14 @@ Result> PageFilteredRowGroupReader::Re for (int field_idx : field_indices) { PAIMON_ASSIGN_OR_RAISE( std::shared_ptr chunked_array, - ReadFilteredField(arrow_file_reader, rg_page_index_reader, row_group_index, - field_idx, column_indices, row_ranges, row_group_row_count, pool)); + ReadFilteredField(arrow_file_reader, rg_page_index_reader, row_group_index, field_idx, + column_indices, row_ranges, row_group_row_count, pool)); if (chunked_array && chunked_array->length() != expected_rows) { - return Status::Invalid(fmt::format( - "PageFilteredRowGroupReader: field {} produced {} rows but expected {} " - "(row_group={})", - field_idx, chunked_array->length(), expected_rows, row_group_index)); + return Status::Invalid( + fmt::format("PageFilteredRowGroupReader: field {} produced {} rows but expected {} " + "(row_group={})", + field_idx, chunked_array->length(), expected_rows, row_group_index)); } columns.push_back(std::move(chunked_array)); @@ -374,4 +375,20 @@ std::vector<::arrow::io::ReadRange> PageFilteredRowGroupReader::ComputePageRange return ranges; } +Result> PageFilteredRowGroupReader::BuildPageFilteredSchema( + ::parquet::arrow::FileReader* file_reader, const std::vector& column_indices) { + // Use SchemaManifest to build schema with proper nested field structure. + // GetFieldIndices maps leaf column indices to top-level field indices, + // correctly handling nested types (List, Struct, Map). + const auto& manifest = file_reader->manifest(); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::vector field_indices, + manifest.GetFieldIndices(std::vector(column_indices.begin(), column_indices.end()))); + std::vector> fields; + for (int field_idx : field_indices) { + fields.push_back(manifest.schema_fields[field_idx].field); + } + return arrow::schema(fields); +} + } // namespace paimon::parquet diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader.h b/src/paimon/format/parquet/page_filtered_row_group_reader.h index c0ee79731..f4f538bd7 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.h +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.h @@ -46,10 +46,8 @@ class PageFilteredRowGroupReader { /// Read a row group with page-level filtering. /// @param arrow_file_reader The Arrow FileReader for ColumnReader tree creation - /// @param parquet_reader The underlying ParquetFileReader /// @param target_row_group Target row group with index and row ranges /// @param column_indices Leaf column indices to read - /// @param arrow_schema The target Arrow schema for output columns /// @param pool Memory pool /// @param cache_options Cache options for PreBuffer /// @param pre_buffered If true, assumes PreBuffer was already called externally @@ -58,13 +56,10 @@ class PageFilteredRowGroupReader { /// @param max_chunksize Per-batch row cap for the returned reader. /// @return A RecordBatchReader streaming the filtered rows. static Result> ReadFilteredRowGroup( - ::parquet::arrow::FileReader* arrow_file_reader, - ::parquet::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, - const std::vector& column_indices, - const std::shared_ptr& arrow_schema, - const ::arrow::io::CacheOptions& cache_options, bool pre_buffered, - const std::vector<::arrow::io::ReadRange>& page_ranges, int64_t max_chunksize, - std::shared_ptr<::arrow::MemoryPool> pool); + ::parquet::arrow::FileReader* arrow_file_reader, const TargetRowGroup& target_row_group, + const std::vector& column_indices, const ::arrow::io::CacheOptions& cache_options, + bool pre_buffered, const std::vector<::arrow::io::ReadRange>& page_ranges, + int64_t max_chunksize, std::shared_ptr<::arrow::MemoryPool> pool); /// Compute the byte ranges of pages that overlap with the given RowRanges. /// Uses OffsetIndex to determine per-page file offsets and sizes. @@ -110,9 +105,12 @@ class PageFilteredRowGroupReader { static Result> ReadFilteredField( ::parquet::arrow::FileReader* arrow_file_reader, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - int32_t row_group_index, int32_t field_index, - const std::vector& column_indices, const RowRanges& row_ranges, - int64_t row_group_row_count, std::shared_ptr<::arrow::MemoryPool> pool); + int32_t row_group_index, int32_t field_index, const std::vector& column_indices, + const RowRanges& row_ranges, int64_t row_group_row_count, + std::shared_ptr<::arrow::MemoryPool> pool); + + static Result> BuildPageFilteredSchema( + ::parquet::arrow::FileReader* file_reader, const std::vector& column_indices); }; } // namespace paimon::parquet diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp index 68a6abede..e20d1b24d 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp @@ -923,8 +923,8 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnRowGroupFilter) { auto id_array = MakeIdColumn(100); auto info_array = MakeNestedStructData(100); - auto data = arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}) - .ValueOrDie(); + auto data = + arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}).ValueOrDie(); WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); auto read_schema = arrow::schema({field_id, field_info}); @@ -962,8 +962,8 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnOnlyReadIdField) { auto id_array = MakeIdColumn(100); auto info_array = MakeNestedStructData(100); - auto data = arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}) - .ValueOrDie(); + auto data = + arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}).ValueOrDie(); WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); // Read "id" column only @@ -1026,8 +1026,8 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedListColumnRowGroupFilter) { auto id_array = MakeIdColumn(100); auto tags_array = MakeListColumnData(100); - auto data = arrow::StructArray::Make({id_array, tags_array}, {field_id, field_tags}) - .ValueOrDie(); + auto data = + arrow::StructArray::Make({id_array, tags_array}, {field_id, field_tags}).ValueOrDie(); WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); auto read_schema = arrow::schema({field_id, field_tags}); @@ -1059,8 +1059,8 @@ TEST_F(PageFilteredRowGroupReaderTest, NestedMapColumnRowGroupFilter) { auto id_array = MakeIdColumn(100); auto props_array = MakeMapColumnData(100); - auto data = arrow::StructArray::Make({id_array, props_array}, {field_id, field_props}) - .ValueOrDie(); + auto data = + arrow::StructArray::Make({id_array, props_array}, {field_id, field_props}).ValueOrDie(); WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); auto read_schema = arrow::schema({field_id, field_props}); @@ -1155,11 +1155,11 @@ TEST_F(PageFilteredRowGroupReaderTest, MultipleNestedColumns) { // predicate: id in [15, 29] or id in [80, 99] ASSERT_OK_AND_ASSIGN( - auto predicate, - PredicateBuilder::Or({PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"id", - FieldType::INT, Literal(15), Literal(29)), - PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"id", - FieldType::INT, Literal(80), Literal(99))})); + auto predicate, PredicateBuilder::Or( + {PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"id", + FieldType::INT, Literal(15), Literal(29)), + PredicateBuilder::Between(/*field_index=*/0, /*field_name=*/"id", + FieldType::INT, Literal(80), Literal(99))})); std::shared_ptr result; ReadWithPredicateImpl(file_name, read_schema, predicate, &result, From 2b169e88be4c1cbe8deae204299a4ab8d5dd9dad Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Wed, 8 Jul 2026 15:12:57 +0800 Subject: [PATCH 5/9] patch arrow --- cmake_modules/arrow.diff | 267 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 266 insertions(+), 1 deletion(-) diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff index 70d33c7b5..ea2787535 100644 --- a/cmake_modules/arrow.diff +++ b/cmake_modules/arrow.diff @@ -437,4 +437,269 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h /// \return Status - Status Advance(int64_t nbytes); + virtual Status Advance(int64_t nbytes); - \ No newline at end of file + +--- a/cpp/src/parquet/arrow/reader.cc ++++ b/cpp/src/parquet/arrow/reader.cc +@@ -254,6 +254,11 @@ + return GetColumn(i, AllRowGroupsFactory(), out); + } + ++ ::arrow::Status GetColumn( ++ int i, const std::vector& column_indices, ++ FileColumnIteratorFactory iterator_factory, ++ std::unique_ptr* out) override; ++ + Status GetSchema(std::shared_ptr<::arrow::Schema>* out) override { + return FromParquetSchema(reader_->metadata()->schema(), reader_properties_, + reader_->metadata()->key_value_metadata(), out); +@@ -493,10 +498,33 @@ + + ::arrow::Status BuildArray(int64_t length_upper_bound, + std::shared_ptr<::arrow::ChunkedArray>* out) final { ++ if (!out_) { ++ BEGIN_PARQUET_CATCH_EXCEPTIONS ++ RETURN_NOT_OK( ++ TransferColumnData(record_reader_.get(), field_, descr_, ctx_->pool, &out_)); ++ END_PARQUET_CATCH_EXCEPTIONS ++ } + *out = out_; + return Status::OK(); + } + ++ ::arrow::Status BeginRead(int64_t total_row_count) final { ++ BEGIN_PARQUET_CATCH_EXCEPTIONS ++ out_ = nullptr; ++ record_reader_->Reset(); ++ record_reader_->Reserve(total_row_count); ++ return Status::OK(); ++ END_PARQUET_CATCH_EXCEPTIONS ++ } ++ ++ int64_t SkipRecords(int64_t num_records) final { ++ return record_reader_->SkipRecords(num_records); ++ } ++ ++ int64_t ReadRecords(int64_t num_records) final { ++ return record_reader_->ReadRecords(num_records); ++ } ++ + const std::shared_ptr field() override { return field_; } + + private: +@@ -532,6 +560,18 @@ + return storage_reader_->LoadBatch(number_of_records); + } + ++ ::arrow::Status BeginRead(int64_t total_row_count) final { ++ return storage_reader_->BeginRead(total_row_count); ++ } ++ ++ int64_t SkipRecords(int64_t num_records) final { ++ return storage_reader_->SkipRecords(num_records); ++ } ++ ++ int64_t ReadRecords(int64_t num_records) final { ++ return storage_reader_->ReadRecords(num_records); ++ } ++ + Status BuildArray(int64_t length_upper_bound, + std::shared_ptr* out) override { + std::shared_ptr storage; +@@ -576,6 +616,18 @@ + return item_reader_->LoadBatch(number_of_records); + } + ++ ::arrow::Status BeginRead(int64_t total_row_count) final { ++ return item_reader_->BeginRead(total_row_count); ++ } ++ ++ int64_t SkipRecords(int64_t num_records) final { ++ return item_reader_->SkipRecords(num_records); ++ } ++ ++ int64_t ReadRecords(int64_t num_records) final { ++ return item_reader_->ReadRecords(num_records); ++ } ++ + virtual ::arrow::Result> AssembleArray( + std::shared_ptr data) { + if (field_->type()->id() == ::arrow::Type::MAP) { +@@ -709,6 +761,30 @@ + } + return Status::OK(); + } ++ ++ ::arrow::Status BeginRead(int64_t total_row_count) override { ++ for (const std::unique_ptr& reader : children_) { ++ RETURN_NOT_OK(reader->BeginRead(total_row_count)); ++ } ++ return Status::OK(); ++ } ++ ++ int64_t SkipRecords(int64_t num_records) override { ++ int64_t skipped = 0; ++ for (const std::unique_ptr& reader : children_) { ++ skipped = reader->SkipRecords(num_records); ++ } ++ return skipped; ++ } ++ ++ int64_t ReadRecords(int64_t num_records) override { ++ int64_t read = 0; ++ for (const std::unique_ptr& reader : children_) { ++ read = reader->ReadRecords(num_records); ++ } ++ return read; ++ } ++ + Status BuildArray(int64_t length_upper_bound, + std::shared_ptr* out) override; + Status GetDefLevels(const int16_t** data, int64_t* length) override; +@@ -1228,6 +1304,23 @@ + std::unique_ptr result; + RETURN_NOT_OK(GetReader(manifest_.schema_fields[i], ctx, &result)); + *out = std::move(result); ++ return Status::OK(); ++} ++ ++::arrow::Status FileReaderImpl::GetColumn( ++ int i, const std::vector& column_indices, ++ FileColumnIteratorFactory iterator_factory, ++ std::unique_ptr* out) { ++ RETURN_NOT_OK(BoundsCheckColumn(i)); ++ auto ctx = std::make_shared(); ++ ctx->reader = reader_.get(); ++ ctx->pool = pool_; ++ ctx->iterator_factory = iterator_factory; ++ ctx->filter_leaves = true; ++ ctx->included_leaves = VectorToSharedSet(column_indices); ++ std::unique_ptr result; ++ RETURN_NOT_OK(GetReader(manifest_.schema_fields[i], ctx, &result)); ++ *out = std::move(result); + return Status::OK(); + } + +--- a/cpp/src/parquet/arrow/reader.h ++++ b/cpp/src/parquet/arrow/reader.h +@@ -48,9 +48,13 @@ + + class ColumnChunkReader; + class ColumnReader; ++class FileColumnIterator; + struct SchemaManifest; + class RowGroupReader; + ++using FileColumnIteratorFactory = ++ std::function; ++ + /// \brief Arrow read adapter class for deserializing Parquet files as Arrow row batches. + /// + /// This interfaces caters for different use cases and thus provides different +@@ -136,6 +140,27 @@ + // The indicated column index is relative to the schema + virtual ::arrow::Status GetColumn(int i, std::unique_ptr* out) = 0; + ++ /// \brief Return a ColumnReader with a custom FileColumnIteratorFactory ++ /// and leaf column filtering. ++ /// ++ /// This allows callers to customize page reading behavior (e.g., setting ++ /// data_page_filter for page-level skipping) and to select only specific ++ /// leaf columns within a nested field. The factory is called once per leaf ++ /// column included in column_indices. ++ /// ++ /// \param i top-level field index (same as GetColumn(int i, ...)) ++ /// \param column_indices leaf column indices to include (enables sub-column ++ /// projection within nested types) ++ /// \param iterator_factory factory to create FileColumnIterator per leaf ++ /// \param[out] out the ColumnReader (may be nullptr if all leaves are pruned) ++ virtual ::arrow::Status GetColumn( ++ int i, const std::vector& column_indices, ++ FileColumnIteratorFactory iterator_factory, ++ std::unique_ptr* out) { ++ return ::arrow::Status::NotImplemented( ++ "GetColumn with factory not implemented"); ++ } ++ + /// \brief Return arrow schema for all the columns. + virtual ::arrow::Status GetSchema(std::shared_ptr<::arrow::Schema>* out) = 0; + +@@ -316,6 +341,29 @@ + // the data available in the file. + virtual ::arrow::Status NextBatch(int64_t batch_size, + std::shared_ptr<::arrow::ChunkedArray>* out) = 0; ++ ++ /// \brief Prepare for filtered reading. Resets internal state and ++ /// pre-allocates buffers. Must be called before SkipRecords/ReadRecords. ++ virtual ::arrow::Status BeginRead(int64_t total_row_count) { ++ return ::arrow::Status::NotImplemented("BeginRead not implemented"); ++ } ++ ++ /// \brief Skip num_records top-level records. ++ /// Returns the actual number of records skipped. ++ virtual int64_t SkipRecords(int64_t num_records) { return 0; } ++ ++ /// \brief Read num_records top-level records into internal buffers. ++ /// Returns the actual number of records read. ++ virtual int64_t ReadRecords(int64_t num_records) { return 0; } ++ ++ /// \brief Build the Arrow array from previously loaded data. ++ /// For leaf readers, calls TransferColumnData if not already done. ++ /// For nested readers, assembles the nested array from child arrays. ++ virtual ::arrow::Status BuildArray( ++ int64_t length_upper_bound, ++ std::shared_ptr<::arrow::ChunkedArray>* out) { ++ return ::arrow::Status::NotImplemented("BuildArray not implemented"); ++ } + }; + + /// \brief Experimental helper class for bindings (like Python) that struggle +--- a/cpp/src/parquet/arrow/reader_internal.h ++++ b/cpp/src/parquet/arrow/reader_internal.h +@@ -26,6 +26,7 @@ + #include + #include + ++#include "parquet/arrow/reader.h" + #include "parquet/arrow/schema.h" + #include "parquet/column_reader.h" + #include "parquet/file_reader.h" +@@ -70,6 +71,13 @@ + + virtual ~FileColumnIterator() {} + ++ /// \brief Set a data_page_filter that will be applied to every PageReader ++ /// created by NextChunk(). This enables I/O-level page skipping. ++ void set_data_page_filter( ++ std::function filter) { ++ data_page_filter_ = std::move(filter); ++ } ++ + std::unique_ptr<::parquet::PageReader> NextChunk() { + if (row_groups_.empty()) { + return nullptr; +@@ -77,7 +85,11 @@ + + auto row_group_reader = reader_->RowGroup(row_groups_.front()); + row_groups_.pop_front(); +- return row_group_reader->GetColumnPageReader(column_index_); ++ auto page_reader = row_group_reader->GetColumnPageReader(column_index_); ++ if (page_reader && data_page_filter_) { ++ page_reader->set_data_page_filter(data_page_filter_); ++ } ++ return page_reader; + } + + const SchemaDescriptor* schema() const { return schema_; } +@@ -93,11 +105,9 @@ + ParquetFileReader* reader_; + const SchemaDescriptor* schema_; + std::deque row_groups_; ++ std::function data_page_filter_; + }; + +-using FileColumnIteratorFactory = +- std::function; +- + Status TransferColumnData(::parquet::internal::RecordReader* reader, + const std::shared_ptr<::arrow::Field>& value_field, + const ColumnDescriptor* descr, ::arrow::MemoryPool* pool, From 97ce120c62b21b99c2e10ec0b83780cc76793157 Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Thu, 9 Jul 2026 10:50:25 +0800 Subject: [PATCH 6/9] fix: support read leaf field --- cmake_modules/arrow.diff | 120 ++++++++---------- .../page_filtered_row_group_reader.cpp | 90 +++++-------- .../parquet/page_filtered_row_group_reader.h | 13 +- 3 files changed, 91 insertions(+), 132 deletions(-) diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff index ea2787535..6302c91fa 100644 --- a/cmake_modules/arrow.diff +++ b/cmake_modules/arrow.diff @@ -452,7 +452,7 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h Status GetSchema(std::shared_ptr<::arrow::Schema>* out) override { return FromParquetSchema(reader_->metadata()->schema(), reader_properties_, reader_->metadata()->key_value_metadata(), out); -@@ -493,10 +498,33 @@ +@@ -493,10 +498,42 @@ ::arrow::Status BuildArray(int64_t length_upper_bound, std::shared_ptr<::arrow::ChunkedArray>* out) final { @@ -466,96 +466,73 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h return Status::OK(); } -+ ::arrow::Status BeginRead(int64_t total_row_count) final { ++ ::arrow::Status LoadBatchWithRowFilter(const LeafRowFilter& get_leaf_filter) final { + BEGIN_PARQUET_CATCH_EXCEPTIONS ++ auto [pattern, total] = get_leaf_filter(input_->column_index()); + out_ = nullptr; + record_reader_->Reset(); -+ record_reader_->Reserve(total_row_count); ++ record_reader_->Reserve(total); ++ int64_t current = 0; ++ for (const auto& [skip, read] : pattern) { ++ if (skip > 0) { ++ record_reader_->SkipRecords(skip); ++ current += skip; ++ } ++ if (read > 0) { ++ record_reader_->ReadRecords(read); ++ current += read; ++ } ++ } ++ if (current < total) { ++ record_reader_->SkipRecords(total - current); ++ } ++ RETURN_NOT_OK( ++ TransferColumnData(record_reader_.get(), field_, descr_, ctx_->pool, &out_)); + return Status::OK(); + END_PARQUET_CATCH_EXCEPTIONS + } -+ -+ int64_t SkipRecords(int64_t num_records) final { -+ return record_reader_->SkipRecords(num_records); -+ } -+ -+ int64_t ReadRecords(int64_t num_records) final { -+ return record_reader_->ReadRecords(num_records); -+ } + const std::shared_ptr field() override { return field_; } private: -@@ -532,6 +560,18 @@ +@@ -532,6 +569,10 @@ return storage_reader_->LoadBatch(number_of_records); } -+ ::arrow::Status BeginRead(int64_t total_row_count) final { -+ return storage_reader_->BeginRead(total_row_count); -+ } -+ -+ int64_t SkipRecords(int64_t num_records) final { -+ return storage_reader_->SkipRecords(num_records); -+ } -+ -+ int64_t ReadRecords(int64_t num_records) final { -+ return storage_reader_->ReadRecords(num_records); ++ ::arrow::Status LoadBatchWithRowFilter(const LeafRowFilter& get_leaf_filter) final { ++ return storage_reader_->LoadBatchWithRowFilter(get_leaf_filter); + } + Status BuildArray(int64_t length_upper_bound, std::shared_ptr* out) override { std::shared_ptr storage; -@@ -576,6 +616,18 @@ +@@ -576,6 +617,10 @@ return item_reader_->LoadBatch(number_of_records); } -+ ::arrow::Status BeginRead(int64_t total_row_count) final { -+ return item_reader_->BeginRead(total_row_count); -+ } -+ -+ int64_t SkipRecords(int64_t num_records) final { -+ return item_reader_->SkipRecords(num_records); -+ } -+ -+ int64_t ReadRecords(int64_t num_records) final { -+ return item_reader_->ReadRecords(num_records); ++ ::arrow::Status LoadBatchWithRowFilter(const LeafRowFilter& get_leaf_filter) final { ++ return item_reader_->LoadBatchWithRowFilter(get_leaf_filter); + } + virtual ::arrow::Result> AssembleArray( std::shared_ptr data) { if (field_->type()->id() == ::arrow::Type::MAP) { -@@ -709,6 +761,30 @@ +@@ -709,6 +754,14 @@ } return Status::OK(); } + -+ ::arrow::Status BeginRead(int64_t total_row_count) override { ++ ::arrow::Status LoadBatchWithRowFilter(const LeafRowFilter& get_leaf_filter) override { + for (const std::unique_ptr& reader : children_) { -+ RETURN_NOT_OK(reader->BeginRead(total_row_count)); ++ RETURN_NOT_OK(reader->LoadBatchWithRowFilter(get_leaf_filter)); + } + return Status::OK(); + } -+ -+ int64_t SkipRecords(int64_t num_records) override { -+ int64_t skipped = 0; -+ for (const std::unique_ptr& reader : children_) { -+ skipped = reader->SkipRecords(num_records); -+ } -+ return skipped; -+ } -+ -+ int64_t ReadRecords(int64_t num_records) override { -+ int64_t read = 0; -+ for (const std::unique_ptr& reader : children_) { -+ read = reader->ReadRecords(num_records); -+ } -+ return read; -+ } + Status BuildArray(int64_t length_upper_bound, std::shared_ptr* out) override; Status GetDefLevels(const int16_t** data, int64_t* length) override; -@@ -1228,6 +1304,23 @@ +@@ -1228,6 +1281,23 @@ std::unique_ptr result; RETURN_NOT_OK(GetReader(manifest_.schema_fields[i], ctx, &result)); *out = std::move(result); @@ -581,7 +558,15 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h --- a/cpp/src/parquet/arrow/reader.h +++ b/cpp/src/parquet/arrow/reader.h -@@ -48,9 +48,13 @@ +@@ -21,6 +21,7 @@ + // N.B. we don't include async_generator.h as it's relatively heavy + #include + #include ++#include + #include + + #include "parquet/file_reader.h" +@@ -48,9 +49,13 @@ class ColumnChunkReader; class ColumnReader; @@ -595,7 +580,7 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h /// \brief Arrow read adapter class for deserializing Parquet files as Arrow row batches. /// /// This interfaces caters for different use cases and thus provides different -@@ -136,6 +140,27 @@ +@@ -136,6 +141,27 @@ // The indicated column index is relative to the schema virtual ::arrow::Status GetColumn(int i, std::unique_ptr* out) = 0; @@ -623,25 +608,24 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h /// \brief Return arrow schema for all the columns. virtual ::arrow::Status GetSchema(std::shared_ptr<::arrow::Schema>* out) = 0; -@@ -316,6 +341,29 @@ +@@ -316,6 +342,28 @@ // the data available in the file. virtual ::arrow::Status NextBatch(int64_t batch_size, std::shared_ptr<::arrow::ChunkedArray>* out) = 0; + -+ /// \brief Prepare for filtered reading. Resets internal state and -+ /// pre-allocates buffers. Must be called before SkipRecords/ReadRecords. -+ virtual ::arrow::Status BeginRead(int64_t total_row_count) { -+ return ::arrow::Status::NotImplemented("BeginRead not implemented"); ++ /// \brief Load batch with per-leaf row filtering. ++ /// ++ /// The callback is called once per leaf column with that leaf's column index, ++ /// returning (skip_read_pattern, total_row_count) for that specific leaf. ++ /// Each pair in skip_read_pattern is (num_records_to_skip, num_records_to_read). ++ /// After processing all pairs, remaining records up to total_row_count are skipped. ++ /// Must be followed by BuildArray() to get the result. ++ using LeafRowFilter = std::function< ++ std::pair>, int64_t>(int)>; ++ virtual ::arrow::Status LoadBatchWithRowFilter(const LeafRowFilter& get_leaf_filter) { ++ return ::arrow::Status::NotImplemented("LoadBatchWithRowFilter not implemented"); + } + -+ /// \brief Skip num_records top-level records. -+ /// Returns the actual number of records skipped. -+ virtual int64_t SkipRecords(int64_t num_records) { return 0; } -+ -+ /// \brief Read num_records top-level records into internal buffers. -+ /// Returns the actual number of records read. -+ virtual int64_t ReadRecords(int64_t num_records) { return 0; } -+ + /// \brief Build the Arrow array from previously loaded data. + /// For leaf readers, calls TransferColumnData if not already done. + /// For nested readers, assembles the nested array from child arrays. 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 ff26d54bb..3de9826a7 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp @@ -124,36 +124,16 @@ std::pair PageFilteredRowGroupReader::ComputeCompressedRowRa return {compressed, compressed_offset}; } -Status PageFilteredRowGroupReader::ExecuteSkipReadPattern( - ::parquet::arrow::ColumnReader* column_reader, const RowRanges& ranges, int64_t total_row_count, - int32_t row_group_index, int32_t field_index) { - int64_t current_row = 0; +std::vector> PageFilteredRowGroupReader::RowRangesToSkipReadPattern( + const RowRanges& ranges) { + std::vector> pattern; + int64_t current = 0; for (const auto& range : ranges.GetRanges()) { - if (range.from > current_row) { - int64_t to_skip = range.from - current_row; - int64_t skipped = column_reader->SkipRecords(to_skip); - if (skipped != to_skip) { - return Status::Invalid(fmt::format( - "PageFilteredRowGroupReader: expected to skip {} records but skipped {} " - "(row_group={}, field={})", - to_skip, skipped, row_group_index, field_index)); - } - current_row = range.from; - } - int64_t to_read = range.Count(); - int64_t read = column_reader->ReadRecords(to_read); - if (read != to_read) { - return Status::Invalid( - fmt::format("PageFilteredRowGroupReader: expected to read {} records but read {} " - "(row_group={}, field={}, range=[{},{}])", - to_read, read, row_group_index, field_index, range.from, range.to)); - } - current_row += to_read; - } - if (current_row < total_row_count) { - column_reader->SkipRecords(total_row_count - current_row); + int64_t skip = range.from > current ? range.from - current : 0; + pattern.emplace_back(skip, range.Count()); + current = range.to + 1; } - return Status::OK(); + return pattern; } Status PageFilteredRowGroupReader::WaitForPreBuffer( @@ -186,32 +166,14 @@ Result> PageFilteredRowGroupReader::ReadFil int32_t row_group_index, int32_t field_index, const std::vector& column_indices, const RowRanges& row_ranges, int64_t row_group_row_count, std::shared_ptr<::arrow::MemoryPool> pool) { - const auto& manifest = arrow_file_reader->manifest(); - const auto& schema_field = manifest.schema_fields[field_index]; - bool is_flat = schema_field.is_leaf(); - - // For flat columns: compute compressed RowRanges if OffsetIndex is available. - // data_page_filter + compressed_ranges enable I/O-level page skipping. - // For nested fields: use original row_ranges, decode-level skipping only. - RowRanges effective_ranges = row_ranges; - int64_t effective_total = row_group_row_count; - if (is_flat && rg_page_index_reader) { - auto offset_index = rg_page_index_reader->GetOffsetIndex(schema_field.column_index); - if (offset_index) { - auto [compressed, total] = - ComputeCompressedRowRanges(row_ranges, offset_index, row_group_row_count); - effective_ranges = std::move(compressed); - effective_total = total; - } - } - - // Factory: set data_page_filter only for flat columns with OffsetIndex + // Factory: set data_page_filter on every leaf (per-leaf OffsetIndex). + // data_page_filter enables I/O-level page skipping for all leaves. auto factory = - [row_group_index, is_flat, &rg_page_index_reader, &row_ranges, row_group_row_count]( + [row_group_index, &rg_page_index_reader, &row_ranges, row_group_row_count]( int col_idx, ::parquet::ParquetFileReader* reader) -> ::parquet::arrow::FileColumnIterator* { auto* iter = new ::parquet::arrow::FileColumnIterator(col_idx, reader, {row_group_index}); - if (is_flat && rg_page_index_reader) { + if (rg_page_index_reader) { auto offset_index = rg_page_index_reader->GetOffsetIndex(col_idx); if (offset_index) { iter->set_data_page_filter( @@ -230,16 +192,30 @@ Result> PageFilteredRowGroupReader::ReadFil return std::shared_ptr(); } - // Phase 1: Prepare for reading - PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->BeginRead(effective_total)); + // Per-leaf callback: each leaf gets its own skip/read pattern + total. + auto get_leaf_filter = + [&rg_page_index_reader, &row_ranges, row_group_row_count]( + int col_idx) -> std::pair>, int64_t> { + RowRanges effective_ranges = row_ranges; + int64_t effective_total = row_group_row_count; + if (rg_page_index_reader) { + auto offset_index = rg_page_index_reader->GetOffsetIndex(col_idx); + if (offset_index) { + auto [compressed, total] = + ComputeCompressedRowRanges(row_ranges, offset_index, row_group_row_count); + effective_ranges = std::move(compressed); + effective_total = total; + } + } + return {RowRangesToSkipReadPattern(effective_ranges), effective_total}; + }; - // Phase 2: Execute skip/read pattern - PAIMON_RETURN_NOT_OK(ExecuteSkipReadPattern(column_reader.get(), effective_ranges, - effective_total, row_group_index, field_index)); + // Load + filter + transfer (per-leaf, via tree delegation) + PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->LoadBatchWithRowFilter(get_leaf_filter)); - // Phase 3: Build the Arrow array (TransferColumnData for leaves + assemble for nested) + // Build the Arrow array (TransferColumnData for leaves + assemble for nested) std::shared_ptr chunked_array; - PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->BuildArray(effective_total, &chunked_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->BuildArray(row_group_row_count, &chunked_array)); return chunked_array; } diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader.h b/src/paimon/format/parquet/page_filtered_row_group_reader.h index f4f538bd7..c31fce313 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.h +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.h @@ -84,11 +84,6 @@ class PageFilteredRowGroupReader { const std::vector<::arrow::io::ReadRange>& page_ranges, std::shared_ptr<::arrow::MemoryPool> pool); - /// Execute the skip/read pattern on a ColumnReader based on RowRanges. - static Status ExecuteSkipReadPattern(::parquet::arrow::ColumnReader* column_reader, - const RowRanges& ranges, int64_t total_row_count, - int32_t row_group_index, int32_t field_index); - /// Create a data_page_filter callback for a column based on RowRanges + OffsetIndex. static std::function MakePageFilter( const RowRanges& row_ranges, const std::shared_ptr<::parquet::OffsetIndex>& offset_index, @@ -99,9 +94,13 @@ class PageFilteredRowGroupReader { const RowRanges& original_ranges, const std::shared_ptr<::parquet::OffsetIndex>& offset_index, int64_t row_group_row_count); + /// Convert RowRanges to skip/read pattern. Each pair is (skip, read). + static std::vector> RowRangesToSkipReadPattern( + const RowRanges& ranges); + /// Read a field (flat or nested) using ColumnReader tree. - /// For flat columns: sets data_page_filter + uses compressed RowRanges (I/O + decode skipping). - /// For nested fields: decode-level skipping only via SkipRecords/ReadRecords. + /// Sets data_page_filter on all leaves via factory, then uses + /// LoadBatchWithRowFilter with per-leaf compressed_ranges. static Result> ReadFilteredField( ::parquet::arrow::FileReader* arrow_file_reader, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, From df19c4f362934e36273f143575d47440bfd28c13 Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Thu, 9 Jul 2026 11:36:28 +0800 Subject: [PATCH 7/9] test: add test case for leaf field projection --- .../page_filtered_row_group_reader_test.cpp | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp index e20d1b24d..eae52db01 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp @@ -1177,4 +1177,53 @@ TEST_F(PageFilteredRowGroupReaderTest, MultipleNestedColumns) { ASSERT_TRUE(result->Equals(expected)); } +/// Test: sub-column projection of a struct type with page-level filtering. +/// +/// Schema: { id: int32, info: struct } +/// Read schema: { info: struct } — project only x, not y. +/// Predicate: id >= 70 → 30 rows expected. +/// Verifies that reading a sub-column of a nested struct works correctly +/// with page-level filtering and the ColumnReader tree (GetColumn + filter_leaves). +TEST_F(PageFilteredRowGroupReaderTest, NestedStructSubColumnProjection) { + std::string file_name = dir_->Str() + "/nested_struct_subcol.parquet"; + + auto field_x = arrow::field("x", arrow::int32()); + auto field_y = arrow::field("y", arrow::int32()); + auto field_id = arrow::field("id", arrow::int32()); + auto field_info = arrow::field("info", arrow::struct_({field_x, field_y})); + + auto id_array = MakeIdColumn(100); + auto info_array = MakeNestedStructData(100); + auto data = + arrow::StructArray::Make({id_array, info_array}, {field_id, field_info}).ValueOrDie(); + WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50); + + // Read only info.x (sub-column projection: only x, not y) + auto read_schema = arrow::schema({arrow::field("info", arrow::struct_({field_x}))}); + + auto predicate = PredicateBuilder::GreaterOrEqual( + /*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70)); + + std::shared_ptr result; + ReadWithPredicateImpl(file_name, read_schema, predicate, &result); + + ASSERT_TRUE(result); + ASSERT_EQ(30, result->length()); + + // Result is struct> + auto result_struct = std::dynamic_pointer_cast(result->chunk(0)); + ASSERT_TRUE(result_struct); + ASSERT_EQ(1, result_struct->num_fields()); + + auto info_result = std::dynamic_pointer_cast(result_struct->field(0)); + ASSERT_TRUE(info_result); + ASSERT_EQ(1, info_result->num_fields()); + + auto x_arr = std::dynamic_pointer_cast(info_result->field(0)); + ASSERT_TRUE(x_arr); + for (int32_t i = 0; i < 30; ++i) { + ASSERT_EQ((70 + i) * 100, x_arr->Value(i)) << "Mismatch at index " << i; + } +} + } // namespace paimon::parquet::test From dc19071e279af6aef3d728321787072c474b735c Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Thu, 9 Jul 2026 11:39:37 +0800 Subject: [PATCH 8/9] fix: build schema from actual column types --- .../page_filtered_row_group_reader.cpp | 37 ++++++------------- .../parquet/page_filtered_row_group_reader.h | 3 -- 2 files changed, 12 insertions(+), 28 deletions(-) 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 3de9826a7..864a1fb3e 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp @@ -226,17 +226,9 @@ Result> PageFilteredRowGroupReader::Re bool pre_buffered, const std::vector<::arrow::io::ReadRange>& page_ranges, int64_t max_chunksize, std::shared_ptr<::arrow::MemoryPool> pool) { auto parquet_reader = arrow_file_reader->parquet_reader(); - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr arrow_schema, - BuildPageFilteredSchema(arrow_file_reader, column_indices)); const auto& row_ranges = target_row_group.row_ranges; int32_t row_group_index = target_row_group.row_group_index; - if (row_ranges.IsEmpty()) { - PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr empty_table, - arrow::Table::MakeEmpty(arrow_schema, pool.get())); - return std::make_unique(std::move(empty_table), max_chunksize); - } - int64_t expected_rows = row_ranges.RowCount(); PAIMON_RETURN_NOT_OK(WaitForPreBuffer(parquet_reader, row_group_index, column_indices, @@ -279,7 +271,18 @@ Result> PageFilteredRowGroupReader::Re columns.push_back(std::move(chunked_array)); } - auto table = arrow::Table::Make(arrow_schema, std::move(columns), expected_rows); + // Build schema from actual column types (not SchemaManifest's full field). + // This handles sub-column projection: when filter_leaves prunes some children, + // the ChunkedArray type is the projected type (e.g. struct instead of struct), + // which may differ from the full field in SchemaManifest. + std::vector> result_fields; + for (size_t i = 0; i < columns.size(); ++i) { + const auto& field_name = manifest.schema_fields[field_indices[i]].field->name(); + result_fields.push_back(arrow::field(field_name, columns[i]->type())); + } + auto result_schema = arrow::schema(result_fields); + + auto table = arrow::Table::Make(result_schema, std::move(columns), expected_rows); return std::make_unique(std::move(table), max_chunksize); } @@ -351,20 +354,4 @@ std::vector<::arrow::io::ReadRange> PageFilteredRowGroupReader::ComputePageRange return ranges; } -Result> PageFilteredRowGroupReader::BuildPageFilteredSchema( - ::parquet::arrow::FileReader* file_reader, const std::vector& column_indices) { - // Use SchemaManifest to build schema with proper nested field structure. - // GetFieldIndices maps leaf column indices to top-level field indices, - // correctly handling nested types (List, Struct, Map). - const auto& manifest = file_reader->manifest(); - PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( - std::vector field_indices, - manifest.GetFieldIndices(std::vector(column_indices.begin(), column_indices.end()))); - std::vector> fields; - for (int field_idx : field_indices) { - fields.push_back(manifest.schema_fields[field_idx].field); - } - return arrow::schema(fields); -} - } // namespace paimon::parquet diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader.h b/src/paimon/format/parquet/page_filtered_row_group_reader.h index c31fce313..5e793aaeb 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.h +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.h @@ -107,9 +107,6 @@ class PageFilteredRowGroupReader { int32_t row_group_index, int32_t field_index, const std::vector& column_indices, const RowRanges& row_ranges, int64_t row_group_row_count, std::shared_ptr<::arrow::MemoryPool> pool); - - static Result> BuildPageFilteredSchema( - ::parquet::arrow::FileReader* file_reader, const std::vector& column_indices); }; } // namespace paimon::parquet From 2072e3ed47005f109f512eb5507be5769553194f Mon Sep 17 00:00:00 2001 From: zhouhongfeng Date: Fri, 10 Jul 2026 14:51:41 +0800 Subject: [PATCH 9/9] refractor --- src/paimon/format/parquet/page_filtered_row_group_reader.cpp | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) 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 864a1fb3e..0d8955698 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp @@ -271,10 +271,7 @@ Result> PageFilteredRowGroupReader::Re columns.push_back(std::move(chunked_array)); } - // Build schema from actual column types (not SchemaManifest's full field). - // This handles sub-column projection: when filter_leaves prunes some children, - // the ChunkedArray type is the projected type (e.g. struct instead of struct), - // which may differ from the full field in SchemaManifest. + // Build schema from actual column types std::vector> result_fields; for (size_t i = 0; i < columns.size(); ++i) { const auto& field_name = manifest.schema_fields[field_indices[i]].field->name();