diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index e7d6bf606..d3b17fff3 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,9 +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(), - pre_buffered, page_ranges, max_chunksize, pool_)); + file_reader_.get(), 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; filtered_global_offset_ = 0; @@ -339,29 +340,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; @@ -427,7 +405,9 @@ 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)); + PAIMON_ASSIGN_OR_RAISE(page_filtered_read_schema_, + PageFilteredRowGroupReader::BuildProjectedSchema( + file_reader_.get(), 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..f2642fc7b 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); 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..6f4603620 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.cpp @@ -19,15 +19,18 @@ #include #include "arrow/array.h" +#include "arrow/array/concatenate.h" #include "arrow/builder.h" #include "arrow/chunked_array.h" #include "arrow/io/caching.h" #include "arrow/io/interfaces.h" #include "arrow/table.h" +#include "arrow/util/bit_util.h" #include "arrow/util/future.h" #include "fmt/format.h" #include "paimon/common/utils/arrow/status_utils.h" #include "parquet/arrow/reader_internal.h" +#include "parquet/level_conversion.h" #include "parquet/metadata.h" #include "parquet/schema.h" @@ -58,8 +61,116 @@ class TableRecordBatchReader : public arrow::RecordBatchReader { arrow::TableBatchReader inner_; }; +/// Check if a SchemaField or any of its descendants is a repeated type +/// (List, Map, FixedSizeList, LargeList). This mirrors +/// ColumnReaderImpl::IsOrHasRepeatedChild() to decide whether to use +/// DefRepLevelsToBitmap (true) or DefLevelsToBitmap (false) for Struct assembly. +bool HasRepeatedDescendant(const ::parquet::arrow::SchemaField& field) { + if (field.is_leaf()) return false; + for (const auto& child : field.children) { + auto type_id = child.field->type()->id(); + if (type_id == ::arrow::Type::LIST || type_id == ::arrow::Type::MAP || + type_id == ::arrow::Type::FIXED_SIZE_LIST || type_id == ::arrow::Type::LARGE_LIST) { + return true; + } + if (HasRepeatedDescendant(child)) return true; + } + return false; +} + +::arrow::Result> ChunksToSingleArrayData( + const ::arrow::ChunkedArray& chunked) { + if (chunked.num_chunks() == 0) { + ARROW_ASSIGN_OR_RAISE(auto array, ::arrow::MakeArrayOfNull(chunked.type(), 0)); + return array->data(); + } + if (chunked.num_chunks() == 1) { + return chunked.chunk(0)->data(); + } + ARROW_ASSIGN_OR_RAISE(auto concatenated, ::arrow::Concatenate(chunked.chunks())); + return concatenated->data(); +} + } // namespace +std::shared_ptr PageFilteredRowGroupReader::BuildProjectedField( + const ::parquet::arrow::SchemaField& schema_field, const std::set& column_indices) { + if (schema_field.is_leaf()) { + if (column_indices.count(schema_field.column_index) > 0) { + return schema_field.field; + } + return nullptr; + } + + auto type = schema_field.field->type(); + auto type_id = type->id(); + + if (type_id == ::arrow::Type::STRUCT) { + std::vector> child_fields; + for (const auto& child : schema_field.children) { + auto projected = BuildProjectedField(child, column_indices); + if (projected) { + child_fields.push_back(projected); + } + } + if (child_fields.empty()) return nullptr; + return arrow::field(schema_field.field->name(), arrow::struct_(child_fields), + schema_field.field->nullable()); + } + + if (type_id == ::arrow::Type::LIST || type_id == ::arrow::Type::LARGE_LIST || + type_id == ::arrow::Type::FIXED_SIZE_LIST || type_id == ::arrow::Type::MAP) { + if (schema_field.children.empty()) return nullptr; + auto projected_child = BuildProjectedField(schema_field.children[0], column_indices); + if (!projected_child) return nullptr; + auto child_type = projected_child->type(); + if (type_id == ::arrow::Type::LIST) { + return arrow::field(schema_field.field->name(), arrow::list(child_type), + schema_field.field->nullable()); + } + if (type_id == ::arrow::Type::LARGE_LIST) { + return arrow::field(schema_field.field->name(), arrow::large_list(child_type), + schema_field.field->nullable()); + } + if (type_id == ::arrow::Type::FIXED_SIZE_LIST) { + auto& fsl_type = static_cast(*type); + return arrow::field(schema_field.field->name(), + arrow::fixed_size_list(child_type, fsl_type.list_size()), + schema_field.field->nullable()); + } + if (type_id == ::arrow::Type::MAP) { + if (child_type->id() == ::arrow::Type::STRUCT && child_type->num_fields() == 2) { + return arrow::field( + schema_field.field->name(), + arrow::map(child_type->field(0)->type(), child_type->field(1)->type()), + schema_field.field->nullable()); + } + return arrow::field(schema_field.field->name(), arrow::list(child_type), + schema_field.field->nullable()); + } + } + + return nullptr; +} + +Result> PageFilteredRowGroupReader::BuildProjectedSchema( + ::parquet::arrow::FileReader* arrow_file_reader, const std::vector& column_indices) { + const auto& manifest = arrow_file_reader->manifest(); + std::vector col_indices_vec(column_indices.begin(), column_indices.end()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::vector field_indices, + manifest.GetFieldIndices(col_indices_vec)); + + std::set col_set(column_indices.begin(), column_indices.end()); + std::vector> fields; + for (int field_idx : field_indices) { + auto projected = BuildProjectedField(manifest.schema_fields[field_idx], col_set); + if (projected) { + fields.push_back(projected); + } + } + return arrow::schema(std::move(fields)); +} + std::pair PageFilteredRowGroupReader::GetPageRowRange( const std::vector<::parquet::PageLocation>& page_locations, int32_t page_idx, int64_t row_group_row_count) { @@ -155,17 +266,18 @@ Status PageFilteredRowGroupReader::ExecuteSkipReadPattern( return Status::OK(); } -Result> PageFilteredRowGroupReader::ReadFilteredColumn( +Result, + std::shared_ptr<::parquet::internal::RecordReader>>> +PageFilteredRowGroupReader::ReadLeafColumn( 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) { + bool enable_page_filter, 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; @@ -176,18 +288,15 @@ Result> PageFilteredRowGroupReader::ReadFil auto page_reader = row_group_reader->GetColumnPageReader(column_index); - if (offset_index) { - // Set data_page_filter for I/O-level page skipping + if (enable_page_filter && offset_index) { 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 = @@ -201,7 +310,239 @@ Result> PageFilteredRowGroupReader::ReadFil PAIMON_RETURN_NOT_OK_FROM_ARROW(::parquet::arrow::TransferColumnData( record_reader.get(), field, col_descriptor, pool.get(), &chunked_array)); - return chunked_array; + return std::make_pair(std::move(chunked_array), std::move(record_reader)); +} + +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) { + PAIMON_ASSIGN_OR_RAISE(auto result, + ReadLeafColumn(row_group_reader, parquet_reader, rg_page_index_reader, + row_group_index, column_index, row_ranges, field, + row_group_row_count, /*enable_page_filter=*/true, pool)); + return result.first; +} + +Result, + std::shared_ptr<::parquet::internal::RecordReader>>> +PageFilteredRowGroupReader::ReadAndAssembleField( + const ::parquet::arrow::SchemaField& schema_field, ::parquet::ParquetFileReader* parquet_reader, + const std::shared_ptr<::parquet::RowGroupReader>& row_group_reader, + const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, + int32_t row_group_index, const std::vector& column_indices, + const RowRanges& row_ranges, const std::shared_ptr& field, + int64_t row_group_row_count, int64_t expected_rows, std::shared_ptr<::arrow::MemoryPool> pool, + bool is_top_level) { + namespace bit_util = ::arrow::bit_util; + + // For leaf/flat fields, use ReadLeafColumn. + // Top-level leaf columns get data_page_filter for I/O-level page skipping. + // Nested leaf columns (is_top_level=false) skip data_page_filter to preserve + // def/rep level synchronization across sibling leaf columns. + if (schema_field.is_leaf()) { + PAIMON_ASSIGN_OR_RAISE( + auto result, + ReadLeafColumn(row_group_reader, parquet_reader, rg_page_index_reader, row_group_index, + schema_field.column_index, row_ranges, field, row_group_row_count, + /*enable_page_filter=*/is_top_level, pool)); + return result; + } + + auto type_id = field->type()->id(); + + if (type_id == ::arrow::Type::STRUCT) { + // === Struct Assembly (mimicking StructReader::BuildArray) === + std::vector> child_data; + std::shared_ptr<::parquet::internal::RecordReader> def_level_reader; + auto& projected_type = static_cast(*field->type()); + + for (int j = 0; j < projected_type.num_fields(); ++j) { + const auto& projected_child_field = projected_type.field(j); + // Find the matching SchemaField child by name. + const ::parquet::arrow::SchemaField* child_schema = nullptr; + for (const auto& sf_child : schema_field.children) { + if (sf_child.field->name() == projected_child_field->name()) { + child_schema = &sf_child; + break; + } + } + if (!child_schema) continue; + + PAIMON_ASSIGN_OR_RAISE( + auto child_result, + ReadAndAssembleField(*child_schema, parquet_reader, row_group_reader, + rg_page_index_reader, row_group_index, column_indices, + row_ranges, projected_child_field, row_group_row_count, + expected_rows, pool, /*is_top_level=*/false)); + + if (!def_level_reader) { + def_level_reader = child_result.second; + } + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto array_data, + ChunksToSingleArrayData(*child_result.first)); + child_data.push_back(std::move(array_data)); + } + + // Build validity bitmap from def levels. + bool has_repeated_child = HasRepeatedDescendant(schema_field); + std::shared_ptr<::arrow::ResizableBuffer> null_bitmap; + ::parquet::internal::ValidityBitmapInputOutput validity_io; + validity_io.values_read_upper_bound = expected_rows; + validity_io.values_read = expected_rows; + + if (has_repeated_child) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + null_bitmap, ::arrow::AllocateResizableBuffer(bit_util::BytesForBits(expected_rows), + pool.get())); + validity_io.valid_bits = null_bitmap->mutable_data(); + + const int16_t* def_levels = def_level_reader->def_levels(); + const int16_t* rep_levels = def_level_reader->rep_levels(); + int64_t num_levels = def_level_reader->levels_position(); + ::parquet::internal::DefRepLevelsToBitmap(def_levels, rep_levels, num_levels, + schema_field.level_info, &validity_io); + } else if (field->nullable()) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + null_bitmap, ::arrow::AllocateResizableBuffer(bit_util::BytesForBits(expected_rows), + pool.get())); + validity_io.valid_bits = null_bitmap->mutable_data(); + + const int16_t* def_levels = def_level_reader->def_levels(); + int64_t num_levels = def_level_reader->levels_position(); + ::parquet::internal::DefLevelsToBitmap(def_levels, num_levels, schema_field.level_info, + &validity_io); + } + + if (null_bitmap) { + PAIMON_RETURN_NOT_OK_FROM_ARROW( + null_bitmap->Resize(bit_util::BytesForBits(validity_io.values_read))); + null_bitmap->ZeroPadding(); + } + + if (!field->nullable() && !has_repeated_child && !child_data.empty()) { + validity_io.values_read = child_data.front()->length; + } + + std::vector> buffers{ + validity_io.null_count > 0 ? null_bitmap : nullptr}; + auto data = std::make_shared<::arrow::ArrayData>(field->type(), validity_io.values_read, + std::move(buffers), std::move(child_data)); + auto result = ::arrow::MakeArray(data); + return std::make_pair(std::make_shared(result), def_level_reader); + } + + if (type_id == ::arrow::Type::LIST || type_id == ::arrow::Type::MAP || + type_id == ::arrow::Type::LARGE_LIST || type_id == ::arrow::Type::FIXED_SIZE_LIST) { + // === List/Map Assembly (mimicking ListReader::BuildArray) === + // Map is stored as list> in Parquet, so use List assembly. + const auto& child = schema_field.children[0]; + auto projected_child_field = field->type()->field(0); + PAIMON_ASSIGN_OR_RAISE( + auto child_result, + ReadAndAssembleField(child, parquet_reader, row_group_reader, rg_page_index_reader, + row_group_index, column_indices, row_ranges, projected_child_field, + row_group_row_count, expected_rows, pool, + /*is_top_level=*/false)); + + auto& item_record_reader = child_result.second; + const int16_t* def_levels = item_record_reader->def_levels(); + const int16_t* rep_levels = item_record_reader->rep_levels(); + int64_t num_levels = item_record_reader->levels_position(); + + bool is_large_list = (type_id == ::arrow::Type::LARGE_LIST); + bool is_fixed_size = (type_id == ::arrow::Type::FIXED_SIZE_LIST); + + std::shared_ptr<::arrow::ResizableBuffer> validity_buffer; + ::parquet::internal::ValidityBitmapInputOutput validity_io; + validity_io.values_read_upper_bound = expected_rows; + + if (field->nullable()) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + validity_buffer, ::arrow::AllocateResizableBuffer( + bit_util::BytesForBits(expected_rows), pool.get())); + validity_io.valid_bits = validity_buffer->mutable_data(); + } + + std::shared_ptr<::arrow::ResizableBuffer> offsets_buffer; + + if (is_large_list) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + offsets_buffer, + ::arrow::AllocateResizableBuffer( + sizeof(int64_t) * std::max(int64_t{1}, expected_rows + 1), pool.get())); + auto* offset_data = reinterpret_cast(offsets_buffer->mutable_data()); + offset_data[0] = 0; + ::parquet::internal::DefRepLevelsToList(def_levels, rep_levels, num_levels, + schema_field.level_info, &validity_io, + offset_data); + } else { + // LIST, MAP, and FIXED_SIZE_LIST all use int32 offsets initially. + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + offsets_buffer, + ::arrow::AllocateResizableBuffer( + sizeof(int32_t) * std::max(int64_t{1}, expected_rows + 1), pool.get())); + auto* offset_data = reinterpret_cast(offsets_buffer->mutable_data()); + offset_data[0] = 0; + ::parquet::internal::DefRepLevelsToList(def_levels, rep_levels, num_levels, + schema_field.level_info, &validity_io, + offset_data); + } + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(auto item_data, + ChunksToSingleArrayData(*child_result.first)); + + // Resize buffers to actual size. + if (offsets_buffer) { + PAIMON_RETURN_NOT_OK_FROM_ARROW( + offsets_buffer->Resize((validity_io.values_read + 1) * + (is_large_list ? sizeof(int64_t) : sizeof(int32_t)))); + } + if (validity_buffer) { + PAIMON_RETURN_NOT_OK_FROM_ARROW( + validity_buffer->Resize(bit_util::BytesForBits(validity_io.values_read))); + validity_buffer->ZeroPadding(); + } + + std::vector> buffers; + if (is_fixed_size) { + // Validate each list has the expected fixed size, then drop offsets. + const auto& fs_type = static_cast(*field->type()); + int32_t list_size = fs_type.list_size(); + const int32_t* offsets = reinterpret_cast(offsets_buffer->data()); + for (int64_t x = 1; x <= validity_io.values_read; ++x) { + int32_t size = offsets[x] - offsets[x - 1]; + if (size != list_size) { + return Status::Invalid( + fmt::format("Expected all lists to be of size={} but index {} had size={}", + list_size, x, size)); + } + } + // FixedSizeList only has a validity buffer (no offsets). + buffers.push_back(validity_io.null_count > 0 ? validity_buffer : nullptr); + } else { + buffers.push_back(validity_io.null_count > 0 ? validity_buffer : nullptr); + buffers.push_back(offsets_buffer); + } + + auto data = std::make_shared<::arrow::ArrayData>( + field->type(), validity_io.values_read, std::move(buffers), + std::vector>{item_data}, validity_io.null_count); + + if (type_id == ::arrow::Type::MAP) { + PAIMON_RETURN_NOT_OK_FROM_ARROW(::arrow::MapArray::ValidateChildData(data->child_data)); + } + + auto result = ::arrow::MakeArray(data); + return std::make_pair(std::make_shared(result), item_record_reader); + } + + return Status::Invalid(fmt::format("PageFilteredRowGroupReader: unsupported field type: {}", + field->type()->ToString())); } Status PageFilteredRowGroupReader::WaitForPreBuffer( @@ -229,7 +570,7 @@ Status PageFilteredRowGroupReader::WaitForPreBuffer( } Result> PageFilteredRowGroupReader::ReadFilteredRowGroup( - ::parquet::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, + ::parquet::arrow::FileReader* arrow_file_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, @@ -245,6 +586,8 @@ Result> PageFilteredRowGroupReader::Re int64_t expected_rows = row_ranges.RowCount(); + ::parquet::ParquetFileReader* parquet_reader = arrow_file_reader->parquet_reader(); + PAIMON_RETURN_NOT_OK(WaitForPreBuffer(parquet_reader, row_group_index, column_indices, cache_options, pre_buffered, page_ranges, pool)); @@ -260,23 +603,31 @@ Result> PageFilteredRowGroupReader::Re rg_page_index_reader = page_index_reader->RowGroup(row_group_index); } - // Read each column with page filtering + // Use SchemaManifest to group leaf columns by top-level field. + const auto& manifest = arrow_file_reader->manifest(); + std::vector col_indices_vec(column_indices.begin(), column_indices.end()); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::vector field_indices, + manifest.GetFieldIndices(col_indices_vec)); + + // Read each top-level field with page filtering std::vector> columns; - columns.reserve(column_indices.size()); + columns.reserve(field_indices.size()); - for (size_t i = 0; i < column_indices.size(); ++i) { + for (size_t i = 0; i < field_indices.size(); ++i) { + const auto& schema_field = manifest.schema_fields[field_indices[i]]; 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)); + auto field_result, + ReadAndAssembleField(schema_field, parquet_reader, row_group_reader, + rg_page_index_reader, row_group_index, column_indices, row_ranges, + arrow_schema->field(static_cast(i)), row_group_row_count, + expected_rows, pool)); + auto& chunked_array = field_result.first; if (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_indices[i], 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..ba9aeddd7 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader.h +++ b/src/paimon/format/parquet/page_filtered_row_group_reader.h @@ -20,6 +20,7 @@ #include #include #include +#include #include #include "arrow/io/caching.h" @@ -28,6 +29,8 @@ #include "arrow/type.h" #include "paimon/format/parquet/row_ranges.h" #include "paimon/result.h" +#include "parquet/arrow/reader.h" +#include "parquet/arrow/schema.h" #include "parquet/column_reader.h" #include "parquet/file_reader.h" #include "parquet/page_index.h" @@ -44,7 +47,7 @@ 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 (provides SchemaManifest and GetColumn) /// @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 @@ -56,7 +59,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::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, + ::parquet::arrow::FileReader* arrow_file_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, @@ -71,6 +74,20 @@ class PageFilteredRowGroupReader { ::parquet::ParquetFileReader* parquet_reader, const TargetRowGroup& target_row_group, const std::vector& column_indices); + /// Recursively build a projected Arrow field from a SchemaField, including only + /// leaf columns that are in column_indices. Returns nullptr if no leaves are + /// included. This enables sub-column projection (e.g., reading only struct + /// from a struct field). + static std::shared_ptr BuildProjectedField( + const ::parquet::arrow::SchemaField& schema_field, const std::set& column_indices); + + /// Build the projected Arrow schema for page-filtered reading. Groups leaf + /// column indices into top-level fields via SchemaManifest, then projects + /// each field to include only requested sub-columns. + static Result> BuildProjectedSchema( + ::parquet::arrow::FileReader* arrow_file_reader, + const std::vector& column_indices); + private: /// Get the [first_row, last_row] range of a page given page locations. static std::pair GetPageRowRange( @@ -92,6 +109,35 @@ class PageFilteredRowGroupReader { const RowRanges& ranges, int64_t total_row_count, int32_t row_group_index, int32_t column_index); + /// Read a leaf column and return both the array and the RecordReader. + /// The RecordReader is kept alive so callers can access def/rep levels + static Result, + std::shared_ptr<::parquet::internal::RecordReader>>> + ReadLeafColumn(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, + bool enable_page_filter, std::shared_ptr<::arrow::MemoryPool> pool); + + /// Read a single top-level field (which may be flat or nested) and assemble + /// the result array. For flat/leaf fields, delegates to ReadFilteredColumn with + /// data_page_filter + compressed ranges. For nested fields (Struct, List, Map), + /// reads all leaf columns via RecordReaders (without data_page_filter), then + /// manually assembles the nested array using def/rep levels — mimicking + /// StructReader::BuildArray and ListReader::BuildArray. + static Result, + std::shared_ptr<::parquet::internal::RecordReader>>> + ReadAndAssembleField( + const ::parquet::arrow::SchemaField& schema_field, + ::parquet::ParquetFileReader* parquet_reader, + const std::shared_ptr<::parquet::RowGroupReader>& row_group_reader, + const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, + int32_t row_group_index, const std::vector& column_indices, + const RowRanges& row_ranges, const std::shared_ptr& field, + int64_t row_group_row_count, int64_t expected_rows, + std::shared_ptr<::arrow::MemoryPool> pool, bool is_top_level = true); + /// 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, 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,