fix: preserve explicit Struct casts and evolved field semantics

datafusion/core/tests/parquet/expr_adapter.rs

@@ -790,6 +790,212 @@ async fn test_physical_expr_adapter_with_non_null_defaults() {
     assert_batches_eq!(expected, &batches);
 }
 
+#[tokio::test]
+async fn test_explicit_struct_cast_projection_preserves_sibling_errors() -> Result<()> {
+    let physical_fields: Fields = vec![
+        Field::new("x", DataType::Int32, true),
+        Field::new("y", DataType::Utf8, true),
+    ]
+    .into();
+    let batch = RecordBatch::try_from_iter(vec![(
+        "s",
+        Arc::new(StructArray::new(
+            physical_fields,
+            vec![
+                Arc::new(Int32Array::from(vec![1])) as ArrayRef,
+                Arc::new(StringArray::from(vec!["bad"])) as ArrayRef,
+            ],
+            None,
+        )) as ArrayRef,
+    )])?;
+    let table_schema = Arc::new(Schema::new(vec![Field::new(
+        "s",
+        DataType::Struct(
+            vec![
+                Field::new("x", DataType::Int64, true),
+                Field::new("y", DataType::Utf8, true),
+            ]
+            .into(),
+        ),
+        true,
+    )]));
+    let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+    write_parquet(batch, Arc::clone(&store), "explicit_cast/data.parquet").await;
+    let ctx = test_context();
+    register_memory_listing_table(&ctx, store, "memory:///explicit_cast/", table_schema)
+        .await;
+
+    // The file requires schema adaptation for x. The explicit SQL cast also
+    // converts y, and selecting x must not hide that invalid conversion.
+    let error = ctx
+        .sql("SELECT get_field(CAST(s AS STRUCT<x BIGINT, y INT>), 'x') FROM t")
+        .await?
+        .collect()
+        .await
+        .unwrap_err()
+        .to_string();
+    datafusion_common::assert_contains!(error, "While casting struct field 'y'");
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_all_null_struct_decimal_cast_filter_pushdown() -> Result<()> {
+    use arrow::array::new_null_array;
+    use datafusion_physical_plan::{collect, displayable};
+
+    for (physical_type, logical_type) in [
+        (DataType::Utf8, DataType::Decimal128(10, -1)),
+        (
+            DataType::new_list(DataType::Utf8, true),
+            DataType::new_list(DataType::Decimal128(10, -1), true),
+        ),
+    ] {
+        let physical_fields: Fields =
+            vec![Field::new("x", physical_type.clone(), true)].into();
+        let batch = RecordBatch::try_from_iter(vec![
+            ("row_id", Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef),
+            (
+                "s",
+                Arc::new(StructArray::new(
+                    physical_fields,
+                    vec![new_null_array(&physical_type, 2)],
+                    Some(NullBuffer::new_null(2)),
+                )) as ArrayRef,
+            ),
+        ])?;
+        let table_schema = Arc::new(Schema::new(vec![
+            Field::new("row_id", DataType::Int32, false),
+            Field::new(
+                "s",
+                DataType::Struct(vec![Field::new("x", logical_type, true)].into()),
+                true,
+            ),
+        ]));
+        let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+        write_parquet(batch, Arc::clone(&store), "null_decimal/data.parquet").await;
+
+        for pushdown_filters in [false, true] {
+            let mut config = SessionConfig::new()
+                .with_collect_statistics(false)
+                .with_parquet_pruning(false)
+                .with_parquet_page_index_pruning(false);
+            config.options_mut().execution.parquet.pushdown_filters = pushdown_filters;
+            let ctx = SessionContext::new_with_config(config);
+            register_memory_listing_table(
+                &ctx,
+                Arc::clone(&store),
+                "memory:///null_decimal/",
+                Arc::clone(&table_schema),
+            )
+            .await;
+
+            for (predicate, expected_rows) in [("IS NULL", 2), ("IS NOT NULL", 0)] {
+                let plan = ctx
+                    .sql(&format!(
+                        "SELECT row_id FROM t WHERE get_field(s, 'x') {predicate}"
+                    ))
+                    .await?
+                    .create_physical_plan()
+                    .await?;
+                if pushdown_filters {
+                    let plan_text = displayable(plan.as_ref()).indent(false).to_string();
+                    assert!(
+                        !plan_text.contains("FilterExec"),
+                        "the scan must fully handle the filter: {plan_text}"
+                    );
+                }
+                let batches = collect(plan, ctx.task_ctx()).await?;
+                assert_eq!(
+                    batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+                    expected_rows,
+                    "physical_type={physical_type:?}, pushdown_filters={pushdown_filters}, predicate={predicate}"
+                );
+            }
+        }
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_evolved_decimal_ignores_unselected_sibling() -> Result<()> {
+    let physical_fields: Fields = vec![
+        Field::new("x", DataType::Int32, true),
+        Field::new("y", DataType::Utf8, true),
+    ]
+    .into();
+    let batch = RecordBatch::try_from_iter(vec![
+        ("row_id", Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef),
+        (
+            "s",
+            Arc::new(StructArray::new(
+                physical_fields,
+                vec![
+                    Arc::new(Int32Array::from(vec![1, 0])),
+                    Arc::new(StringArray::from(vec!["bad", "bad"])),
+                ],
+                None,
+            )) as ArrayRef,
+        ),
+    ])?;
+    let table_schema = Arc::new(Schema::new(vec![
+        Field::new("row_id", DataType::Int32, false),
+        Field::new(
+            "s",
+            DataType::Struct(
+                vec![
+                    Field::new("x", DataType::Decimal128(10, 2), true),
+                    Field::new("y", DataType::Int32, true),
+                ]
+                .into(),
+            ),
+            true,
+        ),
+    ]));
+    let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+    write_parquet(batch, Arc::clone(&store), "decimal_sibling/data.parquet").await;
+
+    for pushdown_filters in [false, true] {
+        let mut config = SessionConfig::new()
+            .with_collect_statistics(false)
+            .with_parquet_pruning(false)
+            .with_parquet_page_index_pruning(false);
+        config.options_mut().execution.parquet.pushdown_filters = pushdown_filters;
+        let ctx = SessionContext::new_with_config(config);
+        register_memory_listing_table(
+            &ctx,
+            Arc::clone(&store),
+            "memory:///decimal_sibling/",
+            Arc::clone(&table_schema),
+        )
+        .await;
+
+        // Adapting x must not evaluate the invalid conversion of y.
+        for (sql, expected) in [
+            (
+                "SELECT get_field(s, 'x') AS x FROM t",
+                vec![
+                    "+------+", "| x    |", "+------+", "| 1.00 |", "| 0.00 |",
+                    "+------+",
+                ],
+            ),
+            (
+                "SELECT row_id FROM t WHERE get_field(s, 'x') > 0",
+                vec![
+                    "+--------+",
+                    "| row_id |",
+                    "+--------+",
+                    "| 1      |",
+                    "+--------+",
+                ],
+            ),
+        ] {
+            let batches = ctx.sql(sql).await?.collect().await?;
+            assert_batches_eq!(expected, &batches);
+        }
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_struct_schema_evolution_projection_and_filter() -> Result<()> {
     use std::collections::HashMap;

datafusion/datasource-parquet/src/projection_read_plan.rs

@@ -184,14 +184,23 @@ pub(crate) struct PushdownChecker<'schema> {
     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.
+    /// 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.
     allow_list_columns: bool,
     /// The Arrow schema of the parquet file.
     file_schema: &'schema Schema,
 }
 
 impl<'schema> PushdownChecker<'schema> {
-    pub(crate) fn new(file_schema: &'schema Schema, allow_list_columns: bool) -> Self {
+    pub(crate) fn new(
+        file_schema: &'schema Schema,
+        allow_list_columns: bool,
+        allow_struct_casts: bool,
+    ) -> Self {
         Self {
             non_primitive_columns: false,
             projected_columns: false,
@@ -200,6 +209,7 @@ impl<'schema> PushdownChecker<'schema> {
             struct_field_accesses: Vec::new(),
             cast_accesses: Vec::new(),
             collect_cast_accesses: false,
+            allow_struct_casts,
             allow_list_columns,
             file_schema,
         }
@@ -248,6 +258,56 @@ 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.
+    fn check_cast_struct_field_access(
+        &mut self,
+        func: &ScalarFunctionExpr,
+    ) -> Option<TreeNodeRecursion> {
+        if !self.allow_struct_casts {
+            return None;
+        }
+        let (source, field_names) = func.args().split_first()?;
+        if field_names.is_empty() {
+            return None;
+        }
+        let cast = source.downcast_ref::<CastExpr>()?;
+        let column = cast.expr().downcast_ref::<Column>()?;
+        let index = self.file_schema.index_of(column.name()).ok()?;
+        if !matches!(
+            self.file_schema.field(index).data_type(),
+            DataType::Struct(_)
+        ) {
+            return None;
+        }
+        let return_type = func.return_type();
+        if DataType::is_nested(return_type) && !self.is_nested_type_supported(return_type)
+        {
+            return None;
+        }
+
+        // Every key must resolve through Struct fields in the cast target.
+        // In particular, a key following a Map field is a runtime lookup.
+        let mut data_type = cast.cast_type();
+        for field_name in field_names {
+            let name = field_name
+                .downcast_ref::<Literal>()?
+                .value()
+                .try_as_str()
+                .flatten()?;
+            let DataType::Struct(fields) = data_type else {
+                return None;
+            };
+            data_type = fields
+                .iter()
+                .find(|field| field.name() == name)?
+                .data_type();
+        }
+
+        self.required_columns.push(index);
+        Some(TreeNodeRecursion::Jump)
+    }
+
     fn check_single_column(&mut self, column_name: &str) -> Option<TreeNodeRecursion> {
         let Ok(idx) = self.file_schema.index_of(column_name) else {
             // Column does not exist in the file schema, so we can't push this down.
@@ -344,6 +404,9 @@ impl TreeNodeVisitor<'_> for PushdownChecker<'_> {
         if let Some(func) =
             ScalarFunctionExpr::try_downcast_func::<GetFieldFunc>(node.as_ref())
         {
+            if let Some(recursion) = self.check_cast_struct_field_access(func) {
+                return Ok(recursion);
+            }
             let args = func.args();
 
             if let Some(column) = args.first().and_then(|a| a.downcast_ref::<Column>()) {
@@ -524,7 +587,8 @@ pub(crate) fn build_projection_read_plan(
     let mut all_cast_accesses = Vec::new();
 
     for expr in exprs {
-        let mut checker = PushdownChecker::new(file_schema, true).with_cast_collection();
+        let mut checker =
+            PushdownChecker::new(file_schema, true, false).with_cast_collection();
         let _ = expr.visit(&mut checker);
         let columns = checker.into_sorted_columns();