Skip to content

feat(parquet): support page index filter for nested column type (paimon side) #409

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zhf999 wants to merge 5 commits into from
Closed

feat(parquet): support page index filter for nested column type (paimon side) #409

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4907017
feat: support read nested fields
zhf999 Jul 9, 2026
91c5074
test: add test cases
zhf999 Jul 8, 2026
03a21c9
test: add test case for leaf field projection
zhf999 Jul 9, 2026
cb01e93
fix: support leaf column
zhf999 Jul 9, 2026
575726d
refractor
zhf999 Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 7 additions & 27 deletions src/paimon/format/parquet/file_reader_wrapper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -234,9 +235,9 @@ Result<std::shared_ptr<arrow::RecordBatch>> 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;
Expand Down Expand Up @@ -339,29 +340,6 @@ Status FileReaderWrapper::PrepareForReadingLazy(
return Status::OK();
}

Status FileReaderWrapper::BuildPageFilteredSchema(const std::vector<int32_t>& column_indices) {
if (page_filtered_read_schema_) {
return Status::OK();
}
std::shared_ptr<arrow::Schema> schema;
PAIMON_RETURN_NOT_OK_FROM_ARROW(file_reader_->GetSchema(&schema));
auto parquet_schema = file_reader_->parquet_reader()->metadata()->schema();
std::vector<std::shared_ptr<arrow::Field>> 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<int32_t>& column_indices) {
std::vector<::arrow::io::ReadRange> ranges;
Expand Down Expand Up @@ -427,7 +405,9 @@ Status FileReaderWrapper::PrepareForReading(const std::vector<TargetRowGroup>& 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();
Expand Down
3 changes: 0 additions & 3 deletions src/paimon/format/parquet/file_reader_wrapper.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,9 +157,6 @@ class FileReaderWrapper {
/// Read next batch from the fully-matched batch_reader_. Returns nullptr when exhausted.
Result<std::shared_ptr<arrow::RecordBatch>> NextFullyMatched();

/// Build page_filtered_read_schema_ from the given column indices. No-op if already built.
Status BuildPageFilteredSchema(const std::vector<int32_t>& column_indices);

/// Collect all byte ranges that need pre-buffering (page-filtered + fully-matched).
std::vector<::arrow::io::ReadRange> CollectPreBufferRanges(
const std::vector<int32_t>& column_indices);
Expand Down
Loading