This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new b9968a06ea refactor: de-duplicate parquet read plan construction
(#23426)
b9968a06ea is described below
commit b9968a06eadcc52d970b617d4845034a251184ac
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Thu Jul 9 21:13:05 2026 -0500
refactor: de-duplicate parquet read plan construction (#23426)
## Which issue does this PR close?
- N/A — follow-up cleanup requested in review of #23396, in two
comments:
[`root_level_plan`](https://github.com/apache/datafusion/pull/23396#discussion_r3552883479)
and
[`assemble_read_plan`](https://github.com/apache/datafusion/pull/23396#discussion_r3552881821).
Two independent commits, each green on its own; best reviewed commit by
commit.
## Rationale for this change
@mbutrovich spotted two pieces of duplication in
`projection_read_plan.rs` while reviewing #23396, both deferred out of
that (move-only) PR:
1. `build_projection_read_plan` has three early-return paths — the
all-plain-columns fast path, the no-struct-columns fast path, and the
no-struct-accesses fallback. Each repeated the same
`ProjectionMask::roots` + `Schema::project` +
construct-`ParquetReadPlan` sequence.
2. `build_projection_read_plan` and
`row_filter::build_parquet_read_plan` both end in the same chain to
build a leaf-level plan:
```
leaf_indices_for_roots → resolve_struct_field_leaves → extend/sort/dedup
→ ProjectionMask::leaves → build_filter_schema → ParquetReadPlan
```
They differ only in that the row filter also sizes the resulting leaves
(`required_bytes`) and returns an `Option`.
## What changes are included in this PR?
**Commit 1 — `root_level_plan`:** a new private helper taking sorted,
deduplicated root indices and decoding every leaf below each root. The
three early-return paths delegate to it. Net −9 lines.
**Commit 2 — `assemble_read_plan`:** shared by both callers. It returns
the `ParquetReadPlan` plus the resolved leaf indices, which is what lets
the row filter keep computing `required_bytes` without duplicating leaf
resolution. `build_parquet_read_plan` shrinks from ~30 lines to ~10.
`leaf_indices_for_roots`, `resolve_struct_field_leaves` and
`build_filter_schema` become private to `projection_read_plan`, since
`assemble_read_plan` is now their only caller.
No behavior change in either commit.
## Are these changes tested?
Yes, by existing tests: `datafusion-datasource-parquet` unit tests (158
pass) and the `parquet_integration` suite in `datafusion` (213 pass),
both unchanged and both green at each commit. The row-filter path
touched by commit 2 is covered by `parquet::filter_pushdown` and the
struct-field pushdown tests in `row_filter.rs`.
## Are there any user-facing changes?
No.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01BsRGsN18YBCji5iFTQmWpr
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../datasource-parquet/src/projection_read_plan.rs | 117 ++++++++++++---------
datafusion/datasource-parquet/src/row_filter.rs | 31 +-----
2 files changed, 69 insertions(+), 79 deletions(-)
diff --git a/datafusion/datasource-parquet/src/projection_read_plan.rs
b/datafusion/datasource-parquet/src/projection_read_plan.rs
index 46caed6616..c9d8beab14 100644
--- a/datafusion/datasource-parquet/src/projection_read_plan.rs
+++ b/datafusion/datasource-parquet/src/projection_read_plan.rs
@@ -363,18 +363,7 @@ pub(crate) fn build_projection_read_plan(
root_indices.sort_unstable();
root_indices.dedup();
- let projection_mask =
- ProjectionMask::roots(schema_descr, root_indices.iter().copied());
- let projected_schema = Arc::new(
- file_schema
- .project(&root_indices)
- .expect("valid column indices"),
- );
-
- return ParquetReadPlan {
- projection_mask,
- projected_schema,
- };
+ return root_level_plan(&root_indices, file_schema, schema_descr);
}
// secondary fast path: if the schema has no struct columns, we can skip
@@ -393,19 +382,7 @@ pub(crate) fn build_projection_read_plan(
root_indices.sort_unstable();
root_indices.dedup();
- let projection_mask =
- ProjectionMask::roots(schema_descr, root_indices.iter().copied());
-
- let projected_schema = Arc::new(
- file_schema
- .project(&root_indices)
- .expect("valid column indices"),
- );
-
- return ParquetReadPlan {
- projection_mask,
- projected_schema,
- };
+ return root_level_plan(&root_indices, file_schema, schema_descr);
}
let mut all_root_indices = Vec::new();
@@ -426,38 +403,74 @@ pub(crate) fn build_projection_read_plan(
// when no struct field accesses were found, fall back to root-level
projection
// to match the performance of the simple path
if all_struct_accesses.is_empty() {
- let projection_mask =
- ProjectionMask::roots(schema_descr,
all_root_indices.iter().copied());
- let projected_schema = Arc::new(
- file_schema
- .project(&all_root_indices)
- .expect("valid column indices"),
- );
-
- return ParquetReadPlan {
- projection_mask,
- projected_schema,
- };
+ return root_level_plan(&all_root_indices, file_schema, schema_descr);
}
- let leaf_indices = {
- let mut out =
- leaf_indices_for_roots(all_root_indices.iter().copied(),
schema_descr);
- let struct_leaf_indices =
- resolve_struct_field_leaves(&all_struct_accesses, file_schema,
schema_descr);
+ let (read_plan, _leaf_indices) = assemble_read_plan(
+ &all_root_indices,
+ &all_struct_accesses,
+ file_schema,
+ schema_descr,
+ );
- out.extend_from_slice(&struct_leaf_indices);
- out.sort_unstable();
- out.dedup();
+ read_plan
+}
- out
- };
+/// Builds a leaf-level [`ParquetReadPlan`] covering `root_indices` in full
plus
+/// the individual leaves reached by `struct_field_accesses`.
+///
+/// `root_indices` must be sorted, deduplicated indices into `file_schema`.
+///
+/// Also returns the resolved Parquet leaf indices, sorted and deduplicated, so
+/// callers can size the columns the decoder will read.
+pub(crate) fn assemble_read_plan(
+ root_indices: &[usize],
+ struct_field_accesses: &[StructFieldAccess],
+ file_schema: &Schema,
+ schema_descr: &SchemaDescriptor,
+) -> (ParquetReadPlan, Vec<usize>) {
+ let mut leaf_indices =
+ leaf_indices_for_roots(root_indices.iter().copied(), schema_descr);
+ leaf_indices.extend_from_slice(&resolve_struct_field_leaves(
+ struct_field_accesses,
+ file_schema,
+ schema_descr,
+ ));
+ leaf_indices.sort_unstable();
+ leaf_indices.dedup();
let projection_mask =
ProjectionMask::leaves(schema_descr, leaf_indices.iter().copied());
-
let projected_schema =
- build_filter_schema(file_schema, &all_root_indices,
&all_struct_accesses);
+ build_filter_schema(file_schema, root_indices, struct_field_accesses);
+
+ (
+ ParquetReadPlan {
+ projection_mask,
+ projected_schema,
+ },
+ leaf_indices,
+ )
+}
+
+/// Builds a [`ParquetReadPlan`] that decodes whole root columns.
+///
+/// `root_indices` must be sorted, deduplicated indices into `file_schema`.
Every
+/// leaf below each root is decoded, and the projected schema keeps each root
+/// field's full type. Callers that need to decode only some leaves of a struct
+/// root must build the plan from leaf indices instead.
+fn root_level_plan(
+ root_indices: &[usize],
+ file_schema: &Schema,
+ schema_descr: &SchemaDescriptor,
+) -> ParquetReadPlan {
+ let projection_mask =
+ ProjectionMask::roots(schema_descr, root_indices.iter().copied());
+ let projected_schema = Arc::new(
+ file_schema
+ .project(root_indices)
+ .expect("valid column indices"),
+ );
ParquetReadPlan {
projection_mask,
@@ -465,7 +478,7 @@ pub(crate) fn build_projection_read_plan(
}
}
-pub(crate) fn leaf_indices_for_roots<I>(
+fn leaf_indices_for_roots<I>(
root_indices: I,
schema_descr: &SchemaDescriptor,
) -> Vec<usize>
@@ -491,7 +504,7 @@ where
/// For every `StructFieldAccess`, finds the leaf columns in the Parquet schema
/// whose path matches the struct root name + field path. This avoids reading
all
/// leaves of a struct when only specific fields are needed
-pub(crate) fn resolve_struct_field_leaves(
+fn resolve_struct_field_leaves(
accesses: &[StructFieldAccess],
file_schema: &Schema,
schema_descr: &SchemaDescriptor,
@@ -530,7 +543,7 @@ pub(crate) fn resolve_struct_field_leaves(
/// For struct columns accessed via `get_field`, a pruned struct type is
created
/// containing only the fields along the access path. Note: it must match the
schema
/// that the Parquet reader produces when projecting specific struct leaves
-pub(crate) fn build_filter_schema(
+fn build_filter_schema(
file_schema: &Schema,
regular_indices: &[usize],
struct_field_accesses: &[StructFieldAccess],
diff --git a/datafusion/datasource-parquet/src/row_filter.rs
b/datafusion/datasource-parquet/src/row_filter.rs
index 4505f7fd62..a375e6611e 100644
--- a/datafusion/datasource-parquet/src/row_filter.rs
+++ b/datafusion/datasource-parquet/src/row_filter.rs
@@ -86,8 +86,7 @@ use datafusion_physical_plan::metrics;
use super::ParquetFileMetrics;
use super::supported_predicates::supports_list_predicates;
use crate::projection_read_plan::{
- ParquetReadPlan, PushdownChecker, PushdownColumns, build_filter_schema,
- leaf_indices_for_roots, resolve_struct_field_leaves,
+ ParquetReadPlan, PushdownChecker, PushdownColumns, assemble_read_plan,
};
/// A "compiled" predicate passed to `ParquetRecordBatchStream` to perform
@@ -269,38 +268,16 @@ pub(crate) fn build_parquet_read_plan(
return Ok(None);
};
- let root_indices = &required_columns.required_columns;
-
- let mut leaf_indices =
- leaf_indices_for_roots(root_indices.iter().copied(), schema_descr);
-
- let struct_leaf_indices = resolve_struct_field_leaves(
+ let (read_plan, leaf_indices) = assemble_read_plan(
+ &required_columns.required_columns,
&required_columns.struct_field_accesses,
file_schema,
schema_descr,
);
- leaf_indices.extend_from_slice(&struct_leaf_indices);
- leaf_indices.sort_unstable();
- leaf_indices.dedup();
let required_bytes = size_of_columns(&leaf_indices, metadata)?;
- let projection_mask =
- ProjectionMask::leaves(schema_descr, leaf_indices.iter().copied());
-
- let projected_schema = build_filter_schema(
- file_schema,
- root_indices,
- &required_columns.struct_field_accesses,
- );
-
- Ok(Some((
- ParquetReadPlan {
- projection_mask,
- projected_schema,
- },
- required_bytes,
- )))
+ Ok(Some((read_plan, required_bytes)))
}
/// Checks if a predicate expression can be pushed down to the parquet decoder.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
