@@ -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)
     }
 
@@ -518,8 +531,8 @@ pub(crate) struct PushdownColumns {
     /// Struct field accesses via `get_field`. Each entry records the root struct
     /// column index and the field path being accessed.
     pub(crate) struct_field_accesses: Vec<StructFieldAccess>,
-    /// Whole-column casts to a narrower nested type. Empty unless cast
-    /// collection was enabled on the checker.
+    /// Whole-column casts to a narrower nested type, collected for projections
+    /// or retained Struct casts accepted by the runtime filter checker.
     pub(crate) cast_accesses: Vec<CastColumnAccess>,
 }
 
@@ -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
@@ -645,8 +659,7 @@ enum RootRead {
 ///
 /// Per root, in ascending root-index order:
 /// - roots referenced as whole columns keep every leaf and their full
-///   physical field (whole-column reads take precedence; cast accesses on
-///   such roots were already dropped by the caller);
+///   physical field (whole-column reads take precedence over cast accesses);
 /// - roots consumed through one or more casts keep the union of the leaves
 ///   their targets name (see `crate::nested_schema_pruning`);
 /// - a root consumed through both casts and `get_field` accesses keeps the
@@ -657,13 +670,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 +783,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 +1780,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 +1950,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,
             &[],
 
@@ -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)?;
 
@@ -1237,11 +1252,12 @@ mod test {
     fn get_field_filter_candidate_has_correct_leaf_indices() {
         use arrow::array::{Int32Array, StringArray, StructArray};
 
-        // Schema: id (Int32), s (Struct{value: Int32, label: Utf8})
-        // Parquet leaves: id=0, s.value=1, s.label=2
+        // Schema: id (Int32), s (Struct{value: Int32, label: Utf8, unused: Utf8})
+        // Parquet leaves: id=0, s.value=1, s.label=2, s.unused=3
         let struct_fields: Fields = vec![
             Arc::new(Field::new("value", DataType::Int32, false)),
             Arc::new(Field::new("label", DataType::Utf8, false)),
+            Arc::new(Field::new("unused", DataType::Utf8, false)),
         ]
         .into();
         let schema = Arc::new(Schema::new(vec![
@@ -1258,6 +1274,9 @@ mod test {
                     vec![
                         Arc::new(Int32Array::from(vec![10, 20, 30])) as _,
                         Arc::new(StringArray::from(vec!["a", "b", "c"])) as _,
+                        Arc::new(StringArray::from(vec![
+                            "unused-a", "unused-b", "unused-c",
+                        ])) as _,
                     ],
                     None,
                 )),
@@ -1293,16 +1312,20 @@ mod test {
                 .expect("get_field filter on struct should be pushable");
 
         // The filter accesses only s.value, so only Parquet leaf 1 is needed.
-        // Leaf 2 (s.label) is not read, reducing unnecessary I/O.
+        // Neither sibling is read, reducing unnecessary I/O.
         let expected_mask =
             ProjectionMask::leaves(metadata.file_metadata().schema_descr(), [1]);
         assert_eq!(
             candidate.read_plan.projection_mask, expected_mask,
             "projection_mask should select only the accessed struct field leaf"
         );
 
-        // Schema adaptation can leave a Struct cast intact. Its runtime filter
-        // must read every sibling, while planning still rejects explicit casts.
+        // Schema adaptation retains Struct ancestors for some decimal conversions
+        // so an all-null parent can skip child conversion. Runtime filters must
+        // preserve every conversion named by the retained cast target.
+        // This target includes `label`, so its conversion must still run even
+        // though get_field selects only `value`. Planning keeps a residual filter
+        // for explicit Struct casts.
         let cast_type = DataType::Struct(
             vec![
                 Field::new("value", DataType::Int32, false),
@@ -1325,12 +1348,29 @@ mod test {
                 .build(&metadata)
                 .expect("building cast candidate")
                 .expect("an adapted struct cast must remain evaluable");
+        // Clip the unused sibling, but preserve the failing `label` conversion.
         let expected_mask =
-            ProjectionMask::roots(metadata.file_metadata().schema_descr(), [1]);
+            ProjectionMask::leaves(metadata.file_metadata().schema_descr(), [1, 2]);
         assert_eq!(candidate.read_plan.projection_mask, expected_mask);
+        let DataType::Struct(physical_fields) = file_schema.field(1).data_type() else {
+            unreachable!("s is a struct")
+        };
+        let clipped_field =
+            file_schema
+                .field(1)
+                .clone()
+                .with_data_type(DataType::Struct(
+                    physical_fields.iter().take(2).cloned().collect(),
+                ));
         assert_eq!(
             candidate.read_plan.projected_schema.as_ref(),
-            &file_schema.project(&[1]).unwrap()
+            &Schema::new(vec![clipped_field])
+        );
+        assert_eq!(
+            candidate.required_bytes,
+            (metadata.row_group(0).column(1).compressed_size()
+                + metadata.row_group(0).column(2).compressed_size()) as usize,
+            "filter cost must count only the leaves the cast target reads"
         );
 
         // A simultaneous direct access must not prune siblings that the cast
@@ -1362,6 +1402,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
 
@@ -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(_));