perf: clip the row-filter read plan to the retained cast target

`check_cast_struct_field_access` pushed the struct root into
`required_columns`, so a retained Struct cast decoded every leaf of that
root — even though `retain_field_path` had already pruned the cast target
down to the selected path, leaving a cast that consumes a single leaf.

Record a `CastColumnAccess` instead and route `build_parquet_read_plan`
through `build_read_plan_with_cast_clipping`, which the projection path
already uses for this expression shape. Clipping follows the *cast
target*, not the `get_field` key, so an explicit query cast — whose
target names every field the user asked to convert — keeps its siblings
in the read and still evaluates their conversions. A target that covers
every leaf, or that cannot be clipped safely, falls back to a full read
of that root inside the helper.

`build_read_plan_with_cast_clipping` now returns its leaf indices, sorted
and deduplicated, so `size_of_columns` cannot double-count a repeat.

Also adds SQL-level coverage for the all-null Struct decimal behavior,
and records the one shape the container-unwrapping carve-out leaves open.

Author: adriangb

datafusion/datasource-parquet/src/projection_read_plan.rs

@@ -178,15 +178,16 @@ pub(crate) struct PushdownChecker<'schema> {
     /// Struct field accesses via `get_field`.
     struct_field_accesses: Vec<StructFieldAccess>,
     /// Whole-column casts to a narrower nested type
-    /// (`CAST(col AS narrower_struct)`). Only collected when
-    /// [`Self::with_cast_collection`] enables it (projection analysis);
-    /// filter pushdown leaves this off.
+    /// (`CAST(col AS narrower_struct)`), collected either when
+    /// [`Self::with_cast_collection`] enables it (projection analysis) or when
+    /// [`Self::allow_struct_casts`] accepts a retained cast under a `get_field`
+    /// (filter pushdown).
     cast_accesses: Vec<CastColumnAccess>,
     /// Whether to collect [`Self::cast_accesses`].
     collect_cast_accesses: bool,
     /// Allow `get_field(CAST(struct_column AS Struct(...)), 'field', ...)`
-    /// after schema adaptation, preserving the cast and reading the full source
-    /// Struct. Both source and target must be Struct types.
+    /// after schema adaptation, preserving the cast and reading the leaves its
+    /// target names. Both source and target must be Struct types.
     /// Planning keeps this disabled so explicit casts retain a residual filter.
     allow_struct_casts: bool,
     /// Whether nested list columns are supported by the predicate semantics.
@@ -258,8 +259,17 @@ impl<'schema> PushdownChecker<'schema> {
         None
     }
 
-    /// Preserve a Struct cast retained by schema adaptation and read its full
-    /// root. Pruning siblings or moving the cast could change errors or nulls.
+    /// Preserve a Struct cast retained by schema adaptation and record the
+    /// leaves its target consumes.
+    ///
+    /// The cast is kept intact — moving it could change errors or nulls — but
+    /// the target itself names every field the conversion touches, so the read
+    /// can be clipped to those leaves. `cast_struct_column` resolves source
+    /// children by name and ignores the rest, so a leaf the target does not
+    /// name cannot affect the result. Note this clips by the *cast target*, not
+    /// by the `get_field` key: an explicit query cast names every field the
+    /// user asked to convert, so its siblings stay in the read and their
+    /// conversions still run.
     fn check_cast_struct_field_access(
         &mut self,
         func: &ScalarFunctionExpr,
@@ -304,7 +314,10 @@ impl<'schema> PushdownChecker<'schema> {
                 .data_type();
         }
 
-        self.required_columns.push(index);
+        self.cast_accesses.push(CastColumnAccess {
+            root_index: index,
+            target_type: cast.cast_type().clone(),
+        });
         Some(TreeNodeRecursion::Jump)
     }
 
@@ -607,13 +620,14 @@ pub(crate) fn build_projection_read_plan(
     all_cast_accesses.retain(|c| all_root_indices.binary_search(&c.root_index).is_err());
 
     if !all_cast_accesses.is_empty() {
-        return build_read_plan_with_cast_clipping(
+        let (read_plan, _leaf_indices) = build_read_plan_with_cast_clipping(
             file_schema,
             schema_descr,
             &all_root_indices,
             &all_struct_accesses,
             &all_cast_accesses,
         );
+        return read_plan;
     }
 
     // when no struct field accesses were found, fall back to root-level projection
@@ -657,13 +671,16 @@ enum RootRead {
 ///   `nested_schema_pruning::clip_for_cast`), an access that resolves to no
 ///   leaf at all, or a merged leaf set whose emitted Arrow type can't be
 ///   derived safely, falls back to a full read of that root.
-fn build_read_plan_with_cast_clipping(
+///
+/// Also returns the resolved Parquet leaf indices, sorted and deduplicated, so
+/// callers can size the columns the decoder will read.
+pub(crate) fn build_read_plan_with_cast_clipping(
     file_schema: &Schema,
     schema_descr: &SchemaDescriptor,
     whole_root_indices: &[usize],
     struct_accesses: &[StructFieldAccess],
     cast_accesses: &[CastColumnAccess],
-) -> ParquetReadPlan {
+) -> (ParquetReadPlan, Vec<usize>) {
     // Every referenced root's Parquet leaves, grouped in one pass over the
     // schema descriptor rather than one `leaf_indices_for_roots` scan per
     // root (this function may look up several roots).
@@ -767,17 +784,24 @@ fn build_read_plan_with_cast_clipping(
         fields.push(Arc::new(field.clone()));
     }
     // `ProjectionMask::leaves` only flips flags in a `vec![false; num_columns]`,
-    // so `leaf_indices` needs no sorting or deduplication here.
-    ParquetReadPlan {
-        projection_mask: ProjectionMask::leaves(
-            schema_descr,
-            leaf_indices.iter().copied(),
-        ),
-        projected_schema: Arc::new(Schema::new_with_metadata(
-            fields,
-            file_schema.metadata().clone(),
-        )),
-    }
+    // so the mask itself needs neither ordering nor deduplication. Callers that
+    // size the read do care, so normalize before handing the indices back:
+    // `size_of_columns` sums per index and would double-count a repeat.
+    leaf_indices.sort_unstable();
+    leaf_indices.dedup();
+    (
+        ParquetReadPlan {
+            projection_mask: ProjectionMask::leaves(
+                schema_descr,
+                leaf_indices.iter().copied(),
+            ),
+            projected_schema: Arc::new(Schema::new_with_metadata(
+                fields,
+                file_schema.metadata().clone(),
+            )),
+        },
+        leaf_indices,
+    )
 }
 
 /// Groups every Parquet leaf index by its root (Arrow) column index, in one
@@ -1757,7 +1781,7 @@ mod test {
                 vec![Arc::new(Field::new("p", DataType::Int32, true))].into(),
             ),
         };
-        let read_plan = build_read_plan_with_cast_clipping(
+        let (read_plan, _leaf_indices) = build_read_plan_with_cast_clipping(
             &file_schema,
             schema_descr,
             &[],
@@ -1927,7 +1951,7 @@ mod test {
             Schema::new(vec![divergent("a", "p", "q"), divergent("b", "m", "n")]);
 
         // `a` is reached by a narrowing cast, `b` only by `get_field`.
-        let read_plan = build_read_plan_with_cast_clipping(
+        let (read_plan, _leaf_indices) = build_read_plan_with_cast_clipping(
             &file_schema,
             schema_descr,
             &[],

datafusion/datasource-parquet/src/row_filter.rs

@@ -87,6 +87,7 @@ use super::ParquetFileMetrics;
 use super::supported_predicates::supports_list_predicates;
 use crate::projection_read_plan::{
     ParquetReadPlan, PushdownChecker, PushdownColumns, assemble_read_plan,
+    build_read_plan_with_cast_clipping,
 };
 
 /// A "compiled" predicate passed to `ParquetRecordBatchStream` to perform
@@ -281,12 +282,26 @@ pub(crate) fn build_parquet_read_plan(
         return Ok(None);
     };
 
-    let (read_plan, leaf_indices) = assemble_read_plan(
-        &required_columns.required_columns,
-        &required_columns.struct_field_accesses,
-        file_schema,
-        schema_descr,
-    );
+    // A retained Struct cast names the fields its conversion touches, so the
+    // read is clipped to those leaves rather than decoding the whole root.
+    // A cast whose target covers every leaf, or that cannot be clipped safely,
+    // falls back to a full read of that root inside the helper.
+    let (read_plan, leaf_indices) = if required_columns.cast_accesses.is_empty() {
+        assemble_read_plan(
+            &required_columns.required_columns,
+            &required_columns.struct_field_accesses,
+            file_schema,
+            schema_descr,
+        )
+    } else {
+        build_read_plan_with_cast_clipping(
+            file_schema,
+            schema_descr,
+            &required_columns.required_columns,
+            &required_columns.struct_field_accesses,
+            &required_columns.cast_accesses,
+        )
+    };
 
     let required_bytes = size_of_columns(&leaf_indices, metadata)?;
 
@@ -1362,6 +1377,54 @@ mod test {
             .unwrap();
         let error = row_filter.evaluate(batch).unwrap_err().to_string();
         datafusion_common::assert_contains!(error, "While casting struct field 'label'");
+
+        // A retained cast whose target names only the selected field — the
+        // shape `retain_field_path` produces for an evolved decimal — clips the
+        // read to that field's leaf instead of decoding the whole root.
+        let narrow_cast_type =
+            DataType::Struct(vec![Field::new("value", DataType::Int32, false)].into());
+        let narrow_field = get_field().call(vec![
+            datafusion_expr::cast(col("s"), narrow_cast_type.clone()),
+            lit("value"),
+        ]);
+        let narrow_predicate = logical2physical(&narrow_field.gt(lit(5)), &file_schema);
+        let candidate =
+            FilterCandidateBuilder::new(narrow_predicate, Arc::clone(&file_schema))
+                .build(&metadata)
+                .expect("building narrow cast candidate")
+                .expect("a clipped struct cast must remain evaluable");
+        assert_eq!(
+            candidate.read_plan.projection_mask,
+            ProjectionMask::leaves(metadata.file_metadata().schema_descr(), [1]),
+            "the read must be clipped to the leaf the cast target names"
+        );
+        assert_eq!(candidate.read_plan.projected_schema.fields().len(), 1);
+        assert_eq!(
+            candidate.read_plan.projected_schema.field(0).data_type(),
+            &narrow_cast_type,
+            "sibling leaves must be pruned from the filter schema"
+        );
+
+        // The clipped schema must still evaluate: every row has value > 5.
+        let mut row_filter = DatafusionArrowPredicate::try_new(
+            candidate,
+            Count::new(),
+            Count::new(),
+            Time::new(),
+        )
+        .unwrap();
+        let batch = ParquetRecordBatchReaderBuilder::try_new(file.reopen().unwrap())
+            .unwrap()
+            .with_projection(row_filter.projection().clone())
+            .build()
+            .unwrap()
+            .next()
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            row_filter.evaluate(batch).unwrap(),
+            BooleanArray::from(vec![true, true, true])
+        );
     }
 
     /// Deeply nested get_field: get_field(struct_col, 'outer', 'inner') where the

datafusion/physical-expr-adapter/src/schema_rewriter.rs

@@ -530,7 +530,14 @@ impl DefaultPhysicalExprAdapterRewriter {
         // narrowable by a parent get_field.
         // Container casts involving Struct values must keep their existing
         // dispatch: Arrow can unwrap a container into a Struct where cast_column
-        // cannot.
+        // cannot. That leaves one shape uncovered: unwrapping hands the Struct
+        // to Arrow's own cast, which has no all-null shortcut, so a decimal
+        // below it can still fail on an all-null input (for example
+        // `Dictionary(Int8, Struct<b: Utf8>)` to `Struct<b: Decimal128(10, -1)>`).
+        // The Parquet reader does not produce dictionary-encoded Struct columns,
+        // so this is not reachable through a Parquet scan; closing it would mean
+        // telling "Arrow must unwrap this" apart from "Arrow will convert a
+        // decimal while unwrapping" rather than dropping the carve-out.
         let source_type = physical_struct_field.data_type();
         let target_type = logical_struct_field.data_type();
         let is_struct = |data_type: &DataType| matches!(data_type, DataType::Struct(_));

datafusion/sqllogictest/test_files/schema_evolution_nested.slt

@@ -122,3 +122,58 @@ FROM large_list_messages;
 ----
 1 10 NULL
 2 30 eth
+
+##########
+# An all-null Struct must stay null when a child evolves to a decimal whose
+# conversion is rejected while it is being set up, before any value is read.
+# A whole-Struct conversion never reaches the child; extracting the field first
+# would. This holds whether the decimal is the selected field itself or sits
+# inside a container below it.
+##########
+
+statement ok
+COPY (
+  SELECT 1 AS row_id,
+         CASE WHEN false THEN named_struct('a', arrow_cast('x', 'Utf8')) END AS s
+) TO 'test_files/scratch/schema_evolution_nested/null_struct_decimal/data.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE null_struct_decimal (
+  row_id INT,
+  s STRUCT<a DECIMAL(10, -1)>
+)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/schema_evolution_nested/null_struct_decimal/';
+
+query IR
+SELECT row_id, get_field(s, 'a') FROM null_struct_decimal;
+----
+1 NULL
+
+statement ok
+COPY (
+  SELECT 1 AS row_id,
+         CASE WHEN false
+              THEN named_struct('a', arrow_cast(['x'], 'List(Utf8)'))
+         END AS s
+) TO 'test_files/scratch/schema_evolution_nested/null_struct_list_decimal/data.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE null_struct_list_decimal (
+  row_id INT,
+  s STRUCT<a ARRAY<DECIMAL(10, -1)>>
+)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/schema_evolution_nested/null_struct_list_decimal/';
+
+query I?
+SELECT row_id, s FROM null_struct_list_decimal;
+----
+1 NULL
+
+query I?
+SELECT row_id, get_field(s, 'a') FROM null_struct_list_decimal;
+----
+1 NULL