diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff index ae7752097..6302c91fa 100644 --- a/cmake_modules/arrow.diff +++ b/cmake_modules/arrow.diff @@ -431,9 +431,259 @@ 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); + +--- 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,42 @@ + + ::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 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); ++ 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 ++ } ++ + const std::shared_ptr field() override { return field_; } + + private: +@@ -532,6 +569,10 @@ + return storage_reader_->LoadBatch(number_of_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 +617,10 @@ + return item_reader_->LoadBatch(number_of_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 +754,14 @@ + } + return Status::OK(); + } ++ ++ ::arrow::Status LoadBatchWithRowFilter(const LeafRowFilter& get_leaf_filter) override { ++ for (const std::unique_ptr& reader : children_) { ++ RETURN_NOT_OK(reader->LoadBatchWithRowFilter(get_leaf_filter)); ++ } ++ return Status::OK(); ++ } ++ + Status BuildArray(int64_t length_upper_bound, + std::shared_ptr* out) override; + Status GetDefLevels(const int16_t** data, int64_t* length) override; +@@ -1228,6 +1281,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 +@@ -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; ++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 +141,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 +342,28 @@ + // the data available in the file. + virtual ::arrow::Status NextBatch(int64_t batch_size, + std::shared_ptr<::arrow::ChunkedArray>* out) = 0; ++ ++ /// \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 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, diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index e7d6bf606..555d14062 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" @@ -231,12 +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_->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; @@ -339,29 +339,6 @@ Status FileReaderWrapper::PrepareForReadingLazy( return Status::OK(); } -Status FileReaderWrapper::BuildPageFilteredSchema(const std::vector& column_indices) { - 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(); - 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); - } - 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; @@ -410,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; @@ -426,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 9c87438b8..0d8955698 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" @@ -122,86 +124,16 @@ std::pair PageFilteredRowGroupReader::ComputeCompressedRowRa return {compressed, compressed_offset}; } -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) { - 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 = record_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)); - } - current_row = range.from; - } - int64_t to_read = range.Count(); - int64_t read = record_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)); - } - current_row += to_read; + int64_t skip = range.from > current ? range.from - current : 0; + pattern.emplace_back(skip, range.Count()); + current = range.to + 1; } - if (current_row < total_row_count) { - record_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; + return pattern; } Status PageFilteredRowGroupReader::WaitForPreBuffer( @@ -228,27 +160,80 @@ Status PageFilteredRowGroupReader::WaitForPreBuffer( return Status::OK(); } -Result> PageFilteredRowGroupReader::ReadFilteredRowGroup( - ::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, +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& row_ranges = target_row_group.row_ranges; - int32_t row_group_index = target_row_group.row_group_index; + // 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, &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 (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; + }; - 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); + // 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(); } + // 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}; + }; + + // Load + filter + transfer (per-leaf, via tree delegation) + PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->LoadBatchWithRowFilter(get_leaf_filter)); + + // Build the Arrow array (TransferColumnData for leaves + assemble for nested) + std::shared_ptr chunked_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(column_reader->BuildArray(row_group_row_count, &chunked_array)); + + return chunked_array; +} + +Result> PageFilteredRowGroupReader::ReadFilteredRowGroup( + ::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(); + const auto& row_ranges = target_row_group.row_ranges; + int32_t row_group_index = target_row_group.row_group_index; + int64_t expected_rows = row_ranges.RowCount(); 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,29 +245,41 @@ 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)); - - if (chunked_array->length() != expected_rows) { - return Status::Invalid(fmt::format( - "PageFilteredRowGroupReader: column {} produced {} rows but expected {} " - "(row_group={})", - column_indices[i], chunked_array->length(), expected_rows, row_group_index)); + 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)); } 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 + 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); } 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..5e793aaeb 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,10 +45,9 @@ class PageFilteredRowGroupReader { ~PageFilteredRowGroupReader() = delete; /// Read a row group with page-level filtering. - /// @param parquet_reader The underlying ParquetFileReader + /// @param arrow_file_reader The Arrow FileReader for ColumnReader tree creation /// @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 @@ -56,12 +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::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. @@ -86,30 +84,29 @@ 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); - /// 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); + + /// 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. + /// 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, + 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..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 @@ -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)); @@ -932,10 +937,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))); } @@ -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)); @@ -1052,10 +1039,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))); } @@ -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)); @@ -1080,10 +1072,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))); } @@ -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(); @@ -1137,11 +1110,120 @@ 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))); } +/// 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)); +} + +/// 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 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,