Copilot commented on code in PR #23696:
URL: https://github.com/apache/datafusion/pull/23696#discussion_r3609712080
##########
datafusion/datasource-parquet/src/metrics.rs:
##########
@@ -61,6 +61,12 @@ pub struct ParquetFileMetrics {
/// the initial pruning but were proved unreachable mid-scan after the
/// dynamic filter tightened.
pub row_groups_pruned_dynamic_filter: Count,
+ /// Number of row groups for which the per-row
+ /// [`RowFilter`](parquet::arrow::arrow_reader::RowFilter) was skipped
+ /// because the static stats proved every row of the RG satisfies the
+ /// predicate. The decoder is rebuilt at the boundary with an empty
+ /// row filter so the upcoming RG decodes without per-row evaluation.
Review Comment:
The doc comment for `row_filter_skipped_fully_matched` says it counts row
groups, but the implementation increments only when the filter is toggled off
at an RG boundary (i.e. counts suppressions/toggles, not each fully-matched
RG). Please align the doc comment with the actual semantics (or adjust the
counter to truly count RGs).
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -227,6 +246,289 @@ impl FilterCandidateBuilder {
}
}
+/// Traverses a `PhysicalExpr` tree to determine if any column references would
+/// prevent the expression from being pushed down to the parquet decoder.
+///
+/// An expression cannot be pushed down if it references:
+/// - Unsupported nested columns (whole struct references or list fields that
are
+/// not covered by the supported predicate set)
+/// - Columns that don't exist in the file schema
+///
+/// Struct field access via `get_field` is supported when the resolved leaf
type
+/// is primitive (e.g. `get_field(struct_col, 'field') > 5`).
+struct PushdownChecker<'schema> {
+ /// Does the expression require any non-primitive columns (like structs)?
+ non_primitive_columns: bool,
+ /// Does the expression reference any columns not present in the file
schema?
+ projected_columns: bool,
+ /// Does the expression references a ScalarUDF that requires some rewrite
+ /// and therefore can't be pushed down into the row-filter.
+ has_unpushable_udfs: bool,
+ /// Indices into the file schema of columns required to evaluate the
expression.
+ /// Does not include struct columns accessed via `get_field`.
+ required_columns: Vec<usize>,
+ /// Struct field accesses via `get_field`.
+ struct_field_accesses: Vec<StructFieldAccess>,
+ /// Whether nested list columns are supported by the predicate semantics.
+ allow_list_columns: bool,
+ /// The Arrow schema of the parquet file.
+ file_schema: &'schema Schema,
+}
+
+impl<'schema> PushdownChecker<'schema> {
+ fn new(file_schema: &'schema Schema, allow_list_columns: bool) -> Self {
+ Self {
+ non_primitive_columns: false,
+ projected_columns: false,
+ has_unpushable_udfs: false,
+ required_columns: Vec::new(),
+ struct_field_accesses: Vec::new(),
+ allow_list_columns,
+ file_schema,
+ }
+ }
+
+ /// Checks whether a struct's root column exists in the file schema and,
if so,
+ /// records its index so the entire struct is decoded for filter
evaluation.
+ ///
+ /// This is called when we see a `get_field` expression that resolves to a
+ /// primitive leaf type. We only need the *root* column index because the
+ /// Parquet reader decodes all leaves of a struct together.
+ ///
+ /// # Example
+ ///
+ /// Given file schema `{a: Int32, s: Struct(foo: Utf8, bar: Int64)}` and
the
+ /// expression `get_field(s, 'foo') = 'hello'`:
+ ///
+ /// - `column_name` = `"s"` (the root struct column)
+ /// - `file_schema.index_of("s")` returns `1`
+ /// - We push `1` into `required_columns`
+ /// - Return `None` (no issue — traversal continues in the caller)
+ ///
+ /// If `"s"` is not in the file schema (e.g. a projected-away column), we
set
+ /// `projected_columns = true` and return `Jump` to skip the subtree.
+ fn check_struct_field_column(
+ &mut self,
+ column_name: &str,
+ field_path: Vec<String>,
+ ) -> Option<TreeNodeRecursion> {
+ let Ok(idx) = self.file_schema.index_of(column_name) else {
+ self.projected_columns = true;
+ return Some(TreeNodeRecursion::Jump);
+ };
+
+ self.struct_field_accesses.push(StructFieldAccess {
+ root_index: idx,
+ field_path,
+ });
+
+ None
+ }
+
+ fn check_single_column(&mut self, column_name: &str) ->
Option<TreeNodeRecursion> {
+ let idx = match self.file_schema.index_of(column_name) {
+ Ok(idx) => idx,
+ Err(_) => {
+ // Column does not exist in the file schema, so we can't push
this down.
+ self.projected_columns = true;
+ return Some(TreeNodeRecursion::Jump);
+ }
+ };
+
+ // Duplicates are handled by dedup() in into_sorted_columns()
+ self.required_columns.push(idx);
+ let data_type = self.file_schema.field(idx).data_type();
+
+ if DataType::is_nested(data_type) {
+ self.handle_nested_type(data_type)
+ } else {
+ None
+ }
+ }
+
+ /// Determines whether a nested data type can be pushed down to Parquet
decoding.
+ ///
+ /// Returns `Some(TreeNodeRecursion::Jump)` if the nested type prevents
pushdown,
+ /// `None` if the type is supported and pushdown can continue.
+ fn handle_nested_type(&mut self, data_type: &DataType) ->
Option<TreeNodeRecursion> {
+ if self.is_nested_type_supported(data_type) {
+ None
+ } else {
+ // Block pushdown for unsupported nested types:
+ // - Structs (regardless of predicate support)
+ // - Lists without supported predicates
+ self.non_primitive_columns = true;
+ Some(TreeNodeRecursion::Jump)
+ }
+ }
+
+ /// Checks if a nested data type is supported for list column pushdown.
+ ///
+ /// List columns are only supported if:
+ /// 1. The data type is a list variant (List, LargeList, or FixedSizeList)
+ /// 2. The expression contains supported list predicates (e.g.,
array_has_all)
+ fn is_nested_type_supported(&self, data_type: &DataType) -> bool {
+ let is_list = matches!(
+ data_type,
+ DataType::List(_) | DataType::LargeList(_) |
DataType::FixedSizeList(_, _)
+ );
+ self.allow_list_columns && is_list
+ }
+
+ #[inline]
+ fn prevents_pushdown(&self) -> bool {
+ self.non_primitive_columns || self.projected_columns ||
self.has_unpushable_udfs
+ }
+
+ /// Consumes the checker and returns sorted, deduplicated column indices
+ /// wrapped in a `PushdownColumns` struct.
+ ///
+ /// This method sorts the column indices and removes duplicates. The sort
+ /// is required because downstream code relies on column indices being in
+ /// ascending order for correct schema projection.
+ fn into_sorted_columns(mut self) -> PushdownColumns {
+ self.required_columns.sort_unstable();
+ self.required_columns.dedup();
+ PushdownColumns {
+ required_columns: self.required_columns,
+ struct_field_accesses: self.struct_field_accesses,
+ }
+ }
+}
+
+impl TreeNodeVisitor<'_> for PushdownChecker<'_> {
+ type Node = Arc<dyn PhysicalExpr>;
+
+ fn f_down(&mut self, node: &Self::Node) -> Result<TreeNodeRecursion> {
+ // Handle struct field access like `s['foo']['bar'] > 10`.
+ //
+ // DataFusion represents nested field access as
`get_field(Column("s"), "foo")`
+ // (or chained: `get_field(get_field(Column("s"), "foo"), "bar")`).
+ //
+ // We intercept the outermost `get_field` on the way *down* the tree so
+ // the visitor never reaches the raw `Column("s")` node. Without this,
+ // `check_single_column` would see that `s` is a Struct and reject it.
+ //
+ // The strategy:
+ // 1. Match `get_field` whose first arg is a `Column` (the struct
root).
+ // 2. Check that the *resolved* return type is primitive — meaning
we've
+ // drilled all the way to a leaf (e.g. `s['foo']` → Utf8).
+ // 3. Record the root column index via `check_struct_field_column`
and
+ // return `Jump` to skip visiting the children (the Column and the
+ // literal field-name args), since we've already handled them.
+ //
+ // If the return type is still nested (e.g. `s['nested_struct']` →
Struct),
+ // we fall through and let normal traversal continue, which will
+ // eventually reject the expression when it hits the struct Column.
+ if let Some(func) =
+
ScalarFunctionExpr::try_downcast_func::<GetFieldFunc>(node.as_ref())
+ {
+ let args = func.args();
+
+ if let Some(column) = args.first().and_then(|a|
a.downcast_ref::<Column>()) {
+ // for Map columns, get_field performs a runtime key lookup
rather than a
+ // schema-level field access so the entire Map column must be
read,
+ // we skip the struct field optimization and defer to normal
Column traversal
+ let is_map_column = self
+ .file_schema
+ .index_of(column.name())
+ .ok()
+ .map(|idx| {
+ matches!(
+ self.file_schema.field(idx).data_type(),
+ DataType::Map(_, _)
+ )
+ })
+ .unwrap_or(false);
+
+ let return_type = func.return_type();
+
+ if !is_map_column
+ && (!DataType::is_nested(return_type)
+ || self.is_nested_type_supported(return_type))
+ {
+ // try to resolve all field name arguments to strinrg
literals
Review Comment:
Typo in comment: “strinrg” → “string”.
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -797,9 +795,6 @@ impl ParquetMorselizer {
.transpose()?;
}
Review Comment:
`input_file_name()` must be rewritten to a per-file literal during Parquet
file open; otherwise it will be evaluated directly and error (see
`InputFileNameFunc::invoke_with_args`: "cannot be evaluated directly"). The
`ParquetMorselizer` no longer rewrites `input_file_name()` in the projection,
which will break Parquet queries that project/filter on it (e.g.
`datafusion/sqllogictest/test_files/input_file_name.slt`).
##########
datafusion/datasource-parquet/src/row_filter.rs:
##########
@@ -186,6 +189,22 @@ pub(crate) struct FilterCandidate {
read_plan: ParquetReadPlan,
}
+/// The result of resolving which Parquet leaf columns and Arrow schema fields
+/// are needed to evaluate an expression against a Parquet file
+///
+/// This is the shared output of the column resolution pipeline used by both
+/// the row filter to build `ArrowPredicate`s and the opener to build
`ProjectionMask`s
Review Comment:
`ParquetReadPlan` (and the associated column-resolution logic) is now
duplicated in this file, but the crate already defines the same concept in
`datafusion/datasource-parquet/src/projection_read_plan.rs` (e.g.
`ParquetReadPlan` at ~lines 44-58 there, plus `PushdownChecker`,
`StructFieldAccess`, etc.). Keeping two independent copies risks subtle drift
between projection planning and row-filter planning; consider reusing the
shared module instead of duplicating these types/functions here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
