From 771f67a00fd39b280cee20d4b61d92a97cd14d85 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Tue, 25 Aug 2026 14:17:28 -0700 Subject: [PATCH 1/5] fix: preserve explicit Struct casts and evolved field semantics --- datafusion/core/tests/parquet/expr_adapter.rs | 118 ++++++++ .../src/projection_read_plan.rs | 66 ++++- .../datasource-parquet/src/row_filter.rs | 82 +++++- .../src/schema_rewriter.rs | 272 +++++++++++++++++- 4 files changed, 523 insertions(+), 15 deletions(-) diff --git a/datafusion/core/tests/parquet/expr_adapter.rs b/datafusion/core/tests/parquet/expr_adapter.rs index 3bee5e0070c58..1fba30717b945 100644 --- a/datafusion/core/tests/parquet/expr_adapter.rs +++ b/datafusion/core/tests/parquet/expr_adapter.rs @@ -790,6 +790,124 @@ 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; + 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') 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 datafusion_physical_plan::{collect, displayable}; + + let physical_fields: Fields = vec![Field::new("x", 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(StringArray::from(vec![None::<&str>, None]))], + 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", DataType::Decimal128(10, -1), true)].into(), + ), + true, + ), + ])); + let store = Arc::new(InMemory::new()) as Arc; + 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::(), + expected_rows, + "pushdown_filters={pushdown_filters}, predicate={predicate}" + ); + } + } + Ok(()) +} + #[tokio::test] async fn test_struct_schema_evolution_projection_and_filter() -> Result<()> { use std::collections::HashMap; diff --git a/datafusion/datasource-parquet/src/projection_read_plan.rs b/datafusion/datasource-parquet/src/projection_read_plan.rs index 7793852acf827..29bcc79c3d8ca 100644 --- a/datafusion/datasource-parquet/src/projection_read_plan.rs +++ b/datafusion/datasource-parquet/src/projection_read_plan.rs @@ -184,6 +184,9 @@ pub(crate) struct PushdownChecker<'schema> { cast_accesses: Vec, /// Whether to collect [`Self::cast_accesses`]. collect_cast_accesses: bool, + /// Allow field access through a retained Struct cast after schema adaptation. + /// 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. @@ -191,7 +194,11 @@ pub(crate) struct PushdownChecker<'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 +207,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 +256,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 { + 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::()?; + let column = cast.expr().downcast_ref::()?; + 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::()? + .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 { 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 +402,9 @@ impl TreeNodeVisitor<'_> for PushdownChecker<'_> { if let Some(func) = ScalarFunctionExpr::try_downcast_func::(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::()) { @@ -524,7 +585,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(); diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index 267f0f5917929..712402a012358 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -243,12 +243,17 @@ impl FilterCandidateBuilder { /// /// Returns `None` if the expression cannot be pushed down (e.g., references /// unsupported nested types or columns not in the file). +/// Struct casts are accepted only after schema adaptation, not while planning +/// against the table schema: adaptation may insert another cast underneath an +/// explicit cast, leaving an expression the runtime checker cannot handle. fn pushdown_columns( expr: &Arc, file_schema: &Schema, + allow_struct_casts: bool, ) -> Result> { let allow_list_columns = supports_list_predicates(expr); - let mut checker = PushdownChecker::new(file_schema, allow_list_columns); + let mut checker = + PushdownChecker::new(file_schema, allow_list_columns, allow_struct_casts); expr.visit(&mut checker)?; Ok((!checker.prevents_pushdown()).then(|| checker.into_sorted_columns())) } @@ -272,7 +277,7 @@ pub(crate) fn build_parquet_read_plan( ) -> Result> { let schema_descr = metadata.file_metadata().schema_descr(); - let Some(required_columns) = pushdown_columns(expr, file_schema)? else { + let Some(required_columns) = pushdown_columns(expr, file_schema, true)? else { return Ok(None); }; @@ -358,7 +363,7 @@ pub fn can_expr_be_pushed_down_with_schemas( expr: &Arc, file_schema: &Schema, ) -> bool { - match pushdown_columns(expr, file_schema) { + match pushdown_columns(expr, file_schema, false) { Ok(Some(_)) => true, Ok(None) | Err(_) => false, } @@ -1281,10 +1286,11 @@ mod test { let expr = get_field_expr.gt(Expr::Literal(ScalarValue::Int32(Some(5)), None)); let expr = logical2physical(&expr, &file_schema); - let candidate = FilterCandidateBuilder::new(expr, file_schema) - .build(&metadata) - .expect("building candidate") - .expect("get_field filter on struct should be pushable"); + let candidate = + FilterCandidateBuilder::new(Arc::clone(&expr), Arc::clone(&file_schema)) + .build(&metadata) + .expect("building candidate") + .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. @@ -1294,6 +1300,68 @@ mod test { 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. + let cast_type = DataType::Struct( + vec![ + Field::new("value", DataType::Int32, false), + Field::new("label", DataType::Int32, true), + ] + .into(), + ); + let cast_field = get_field().call(vec![ + datafusion_expr::cast(col("s"), cast_type), + lit("value"), + ]); + let projection = logical2physical(&cast_field, &file_schema); + let cast_predicate = logical2physical(&cast_field.gt(lit(5)), &file_schema); + assert!(!can_expr_be_pushed_down_with_schemas( + &cast_predicate, + &file_schema + )); + let candidate = + FilterCandidateBuilder::new(cast_predicate, Arc::clone(&file_schema)) + .build(&metadata) + .expect("building cast candidate") + .expect("an adapted struct cast must remain evaluable"); + let expected_mask = + ProjectionMask::roots(metadata.file_metadata().schema_descr(), [1]); + assert_eq!(candidate.read_plan.projection_mask, expected_mask); + assert_eq!( + candidate.read_plan.projected_schema.as_ref(), + &file_schema.project(&[1]).unwrap() + ); + + // A simultaneous direct access must not prune siblings that the cast + // needs in the output projection either. + let projection_plan = crate::projection_read_plan::build_projection_read_plan( + [expr, projection], + &file_schema, + metadata.file_metadata().schema_descr(), + ); + assert_eq!(projection_plan.projection_mask, expected_mask); + assert_eq!( + projection_plan.projected_schema, + candidate.read_plan.projected_schema + ); + + let mut row_filter = DatafusionArrowPredicate::try_new( + candidate, + Count::new(), + Count::new(), + Time::new(), + ) + .unwrap(); + let batch = builder + .with_projection(row_filter.projection().clone()) + .build() + .unwrap() + .next() + .unwrap() + .unwrap(); + let error = row_filter.evaluate(batch).unwrap_err().to_string(); + datafusion_common::assert_contains!(error, "While casting struct field 'label'"); } /// Deeply nested get_field: get_field(struct_col, 'outer', 'inner') where the diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index d8cf664707035..0bfca6b0e7db7 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -259,9 +259,10 @@ impl DefaultPhysicalExprAdapter { impl PhysicalExprAdapter for DefaultPhysicalExprAdapter { fn rewrite(&self, expr: Arc) -> Result> { - let rewriter = DefaultPhysicalExprAdapterRewriter { + let mut rewriter = DefaultPhysicalExprAdapterRewriter { logical_file_schema: Arc::clone(&self.logical_file_schema), physical_file_schema: Arc::clone(&self.physical_file_schema), + generated_struct_casts: Vec::new(), }; expr.transform(|expr| rewriter.rewrite_expr(Arc::clone(&expr))) .data() @@ -271,6 +272,9 @@ impl PhysicalExprAdapter for DefaultPhysicalExprAdapter { struct DefaultPhysicalExprAdapterRewriter { logical_file_schema: SchemaRef, physical_file_schema: SchemaRef, + // Retain generated casts so their pointer identity remains reliable even + // after a wider cast has been removed from the expression tree. + generated_struct_casts: Vec>, } /// Outcome of walking a `get_field` key path through nested struct fields. @@ -311,7 +315,7 @@ fn resolve_field_path<'a>( impl DefaultPhysicalExprAdapterRewriter { fn rewrite_expr( - &self, + &mut self, expr: Arc, ) -> Result>> { if let Some(transformed) = self.try_rewrite_struct_field_access(&expr)? { @@ -319,16 +323,29 @@ impl DefaultPhysicalExprAdapterRewriter { } if let Some(transformed) = self.try_narrow_struct_cast(&expr)? { + // A narrowed Struct cast may be accessed by another get_field. + self.record_generated_struct_cast(&transformed); return Ok(Transformed::yes(transformed)); } if let Some(column) = expr.downcast_ref::() { - return self.rewrite_column(Arc::clone(&expr), column); + let transformed = self.rewrite_column(Arc::clone(&expr), column)?; + self.record_generated_struct_cast(&transformed.data); + return Ok(transformed); } Ok(Transformed::no(expr)) } + fn record_generated_struct_cast(&mut self, expr: &Arc) { + if expr + .downcast_ref::() + .is_some_and(|cast| matches!(cast.cast_type(), DataType::Struct(_))) + { + self.generated_struct_casts.push(Arc::clone(expr)); + } + } + /// Rewrite `get_field(cast(s AS Struct<..>), 'f')` into /// `cast(get_field(s, 'f') AS )`. /// @@ -365,7 +382,9 @@ impl DefaultPhysicalExprAdapterRewriter { /// simplified to `get_field(s, 'a', 'b')`, so the whole field path is /// resolved here rather than just the first key. /// - /// Only struct casts are narrowed. `get_field` on a Map column performs a + /// Only struct casts introduced by this adapter are narrowed. Explicit + /// casts must still evaluate sibling conversions, which may fail. + /// `get_field` on a Map column performs a /// runtime key lookup rather than a schema-level field access, so the map /// value must keep its cast. fn try_narrow_struct_cast( @@ -384,6 +403,13 @@ impl DefaultPhysicalExprAdapterRewriter { let Some(cast) = source_expr.downcast_ref::() else { return Ok(None); }; + if !self + .generated_struct_casts + .iter() + .any(|generated| Arc::ptr_eq(generated, source_expr)) + { + return Ok(None); + } // Every key has to be a string literal, otherwise the leaf field // cannot be resolved statically. @@ -436,6 +462,18 @@ impl DefaultPhysicalExprAdapterRewriter { FieldPathResolution::NotAStruct => return Ok(None), }; + // Decimal conversions can fail during setup even for all-null inputs, + // while a Struct cast skips its children when the parent is all null. + // Keep that shortcut, at the cost of reading the whole Struct for an + // evolved decimal field. Same-type metadata casts remain safe to narrow. + let source_type = physical_struct_field.data_type(); + let target_type = logical_struct_field.data_type(); + if source_type != target_type + && (source_type.is_decimal() || target_type.is_decimal()) + { + return Ok(None); + } + // Rebuild `get_field` over the uncast struct so its return field is // recomputed from the physical field type. let mut args = Vec::with_capacity(get_field_expr.args().len()); @@ -448,12 +486,17 @@ impl DefaultPhysicalExprAdapterRewriter { Arc::new(get_field_expr.config_options().clone()), )?) as Arc; - if physical_struct_field == logical_struct_field { + // get_field inherits nullability from every parent along the path. + // Its complete return field can differ even when the leaf fields match. + let logical_return_field = expr.return_field(&self.logical_file_schema)?; + if physical_struct_field == logical_struct_field + && extracted.return_field(&self.physical_file_schema)? == logical_return_field + { return Ok(Some(extracted)); } Ok(Some(Arc::new(CastExpr::new_with_target_field( extracted, - Arc::clone(logical_struct_field), + logical_return_field, Some(cast.cast_options().clone()), )))) } @@ -1573,6 +1616,7 @@ mod tests { let rewriter = DefaultPhysicalExprAdapterRewriter { logical_file_schema: Arc::new(logical_schema), physical_file_schema: Arc::new(physical_schema), + generated_struct_casts: Vec::new(), }; // Test that when a field exists in physical schema, it returns None @@ -1655,6 +1699,222 @@ mod tests { ); } + /// Selecting one field of an explicit cast must still evaluate sibling + /// conversions, even when schema adaptation inserts another cast below it. + #[test] + fn test_narrow_struct_cast_preserves_explicit_cast_errors() -> Result<()> { + use arrow::array::{ArrayRef, Int16Array}; + + for adapt_input in [false, true] { + let physical_x_type = if adapt_input { + DataType::Int16 + } else { + DataType::Int32 + }; + let (logical_schema, physical_schema) = struct_schemas( + vec![ + Field::new("x", physical_x_type, true), + Field::new("y", DataType::Utf8, true), + ], + vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Utf8, true), + ], + ); + let DataType::Struct(physical_fields) = physical_schema.field(0).data_type() + else { + unreachable!() + }; + let x: ArrayRef = if adapt_input { + Arc::new(Int16Array::from(vec![1])) + } else { + Arc::new(Int32Array::from(vec![1])) + }; + let batch = RecordBatch::try_new( + Arc::clone(&physical_schema), + vec![Arc::new(StructArray::new( + physical_fields.clone(), + vec![x, Arc::new(StringArray::from(vec!["bad"]))], + None, + ))], + )?; + let user_cast = Arc::new(CastExpr::new( + Arc::new(Column::new("s", 0)), + DataType::Struct( + vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Int32, true), + ] + .into(), + ), + None, + )); + let expr = Arc::new(ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + vec![user_cast, Arc::new(Literal::new(ScalarValue::from("x")))], + &logical_schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + )?) as Arc; + let original_error = expr.evaluate(&batch).unwrap_err().to_string(); + assert_contains!(original_error, "While casting struct field 'y'"); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(logical_schema, physical_schema)?; + let rewritten = adapter.rewrite(expr)?; + let error = rewritten.evaluate(&batch).unwrap_err().to_string(); + assert_contains!(error, "While casting struct field 'y'"); + } + Ok(()) + } + + /// The result inherits nullability from every parent, not just the leaf. + /// Equal leaf fields do not justify dropping a cast if that loses the + /// logical return field's nullability. + #[test] + fn test_narrow_struct_cast_preserves_logical_return_field() -> Result<()> { + let metadata = HashMap::from([("logical_meta".to_string(), "1".to_string())]); + for nested in [false, true] { + let schema = |leaf: Field, parent_nullable| { + let (field, root_nullable) = if nested { + ( + Field::new( + "inner", + DataType::Struct(vec![leaf].into()), + parent_nullable, + ), + false, + ) + } else { + (leaf, parent_nullable) + }; + Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(vec![field].into()), + root_nullable, + )])) + }; + for same_leaf_type in [false, true] { + let physical_leaf = Field::new("x", DataType::Int32, false) + .with_metadata(metadata.clone()); + let logical_leaf = if same_leaf_type { + physical_leaf.clone() + } else { + Field::new("x", DataType::Int64, false).with_metadata(HashMap::from( + [("logical_meta".to_string(), "2".to_string())], + )) + }; + let physical_schema = schema(physical_leaf, false); + let logical_schema = schema(logical_leaf, true); + let mut args: Vec> = + vec![Arc::new(Column::new("s", 0))]; + if nested { + args.push(Arc::new(Literal::new(ScalarValue::from("inner")))); + } + args.push(Arc::new(Literal::new(ScalarValue::from("x")))); + let expr = Arc::new(ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + args, + &logical_schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + )?) as Arc; + let expected_field = expr.return_field(&logical_schema)?; + assert!(expected_field.is_nullable()); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(logical_schema, Arc::clone(&physical_schema))?; + let rewritten = adapter.rewrite(expr)?; + assert_eq!( + rewritten.return_field(&physical_schema)?, + expected_field, + "nested={nested}, same_leaf_type={same_leaf_type}" + ); + assert!(rewritten.nullable(&physical_schema)?); + } + } + Ok(()) + } + + /// Some decimal casts can fail while preparing the conversion, even for + /// an entirely null input. An all-null Struct skips its child conversions. + #[test] + fn test_narrow_struct_cast_preserves_all_null_decimal_casts() -> Result<()> { + use arrow::array::new_null_array; + use arrow::buffer::NullBuffer; + + for (physical_type, logical_type) in [ + (DataType::Decimal128(38, -38), DataType::Decimal128(38, 38)), + (DataType::Utf8, DataType::Decimal128(10, -1)), + ] { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new("x", physical_type.clone(), true)], + vec![Field::new("x", logical_type, true)], + ); + let DataType::Struct(physical_fields) = physical_schema.field(0).data_type() + else { + unreachable!() + }; + let batch = RecordBatch::try_new( + Arc::clone(&physical_schema), + vec![Arc::new(StructArray::new( + physical_fields.clone(), + vec![new_null_array(&physical_type, 2)], + Some(NullBuffer::new_null(2)), + ))], + )?; + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema)?; + let expr = get_field_expr(&logical_schema, "s", "x"); + + // Establish the result of the original whole-struct conversion. + let whole_struct_cast = adapter.rewrite(Arc::new(Column::new("s", 0)))?; + let original = Arc::clone(&expr).with_new_children(vec![ + whole_struct_cast, + Arc::new(Literal::new(ScalarValue::from("x"))), + ])?; + let expected = original.evaluate(&batch)?.into_array(batch.num_rows())?; + assert_eq!(expected.null_count(), batch.num_rows()); + + let rewritten = adapter.rewrite(expr)?; + let result = rewritten.evaluate(&batch)?.into_array(batch.num_rows())?; + assert_eq!(result.to_data(), expected.to_data()); + } + Ok(()) + } + + #[test] + fn test_narrow_struct_cast_keeps_matching_decimal_fields_optimized() -> Result<()> { + let decimal_type = DataType::Decimal128(10, -1); + for change_metadata in [false, true] { + let physical_field = Field::new("x", decimal_type.clone(), true); + let logical_field = if change_metadata { + physical_field.clone().with_metadata(HashMap::from([( + "logical_meta".to_string(), + "1".to_string(), + )])) + } else { + physical_field.clone() + }; + let (logical_schema, physical_schema) = struct_schemas( + vec![physical_field, Field::new("y", DataType::Int32, true)], + vec![logical_field, Field::new("y", DataType::Int64, true)], + ); + let expr = get_field_expr(&logical_schema, "s", "x"); + let expected_field = expr.return_field(&logical_schema)?; + let adapter = DefaultPhysicalExprAdapterFactory + .create(logical_schema, Arc::clone(&physical_schema))?; + let rewritten = adapter.rewrite(expr)?; + assert_eq!(rewritten.return_field(&physical_schema)?, expected_field); + let extracted = if change_metadata { + assert_cast_expr(&rewritten).expr() + } else { + &rewritten + }; + let get_field = extracted.downcast_ref::().unwrap(); + assert!(get_field.args()[0].downcast_ref::().is_some()); + } + Ok(()) + } + /// A struct field that only differs in a nested leaf type still ends up /// with a single cast on the extracted field. #[test] From 1717bb16ca419b22c6f22bb9cc69eb759b2e5d5a Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 26 Aug 2026 06:42:06 +0000 Subject: [PATCH 2/5] fix: avoid sibling casts and quadratic struct-cast tracking Keep evolved decimal conversions inside their Struct ancestors while restricting the cast target to the requested field path. This preserves all-null shortcuts without evaluating unselected sibling conversions. Index generated casts by pointer while retaining their Arc allocations, avoiding repeated linear identity scans for large expressions. Cover flat and nested decimal access and selective Parquet filters with pushdown enabled and disabled. --- datafusion/core/tests/parquet/expr_adapter.rs | 80 +++++++++++++ .../src/schema_rewriter.rs | 111 ++++++++++++++++-- 2 files changed, 182 insertions(+), 9 deletions(-) diff --git a/datafusion/core/tests/parquet/expr_adapter.rs b/datafusion/core/tests/parquet/expr_adapter.rs index 1fba30717b945..2581c9618d279 100644 --- a/datafusion/core/tests/parquet/expr_adapter.rs +++ b/datafusion/core/tests/parquet/expr_adapter.rs @@ -908,6 +908,86 @@ async fn test_all_null_struct_decimal_cast_filter_pushdown() -> Result<()> { 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; + 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; diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 0bfca6b0e7db7..0b681c0acb3cf 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -262,7 +262,7 @@ impl PhysicalExprAdapter for DefaultPhysicalExprAdapter { let mut rewriter = DefaultPhysicalExprAdapterRewriter { logical_file_schema: Arc::clone(&self.logical_file_schema), physical_file_schema: Arc::clone(&self.physical_file_schema), - generated_struct_casts: Vec::new(), + generated_struct_casts: HashMap::new(), }; expr.transform(|expr| rewriter.rewrite_expr(Arc::clone(&expr))) .data() @@ -274,7 +274,7 @@ struct DefaultPhysicalExprAdapterRewriter { physical_file_schema: SchemaRef, // Retain generated casts so their pointer identity remains reliable even // after a wider cast has been removed from the expression tree. - generated_struct_casts: Vec>, + generated_struct_casts: HashMap<*const (), Arc>, } /// Outcome of walking a `get_field` key path through nested struct fields. @@ -313,6 +313,24 @@ fn resolve_field_path<'a>( } } +/// Retain a field path without changing its ancestors' metadata or nullability. +fn retain_field_path(field: &FieldRef, path: &[&str]) -> Option { + let Some((name, rest)) = path.split_first() else { + return Some(Arc::clone(field)); + }; + let DataType::Struct(fields) = field.data_type() else { + return None; + }; + let child = fields.iter().find(|child| child.name() == *name)?; + let child = retain_field_path(child, rest)?; + Some(Arc::new( + field + .as_ref() + .clone() + .with_data_type(DataType::Struct(vec![child].into())), + )) +} + impl DefaultPhysicalExprAdapterRewriter { fn rewrite_expr( &mut self, @@ -342,7 +360,8 @@ impl DefaultPhysicalExprAdapterRewriter { .downcast_ref::() .is_some_and(|cast| matches!(cast.cast_type(), DataType::Struct(_))) { - self.generated_struct_casts.push(Arc::clone(expr)); + self.generated_struct_casts + .insert(Arc::as_ptr(expr).cast::<()>(), Arc::clone(expr)); } } @@ -405,8 +424,7 @@ impl DefaultPhysicalExprAdapterRewriter { }; if !self .generated_struct_casts - .iter() - .any(|generated| Arc::ptr_eq(generated, source_expr)) + .contains_key(&Arc::as_ptr(source_expr).cast::<()>()) { return Ok(None); } @@ -464,14 +482,25 @@ impl DefaultPhysicalExprAdapterRewriter { // Decimal conversions can fail during setup even for all-null inputs, // while a Struct cast skips its children when the parent is all null. - // Keep that shortcut, at the cost of reading the whole Struct for an - // evolved decimal field. Same-type metadata casts remain safe to narrow. + // Keep the Struct ancestors for that shortcut, but exclude unselected + // siblings whose conversions may fail. Same-type metadata casts remain + // safe to narrow to a scalar cast. let source_type = physical_struct_field.data_type(); let target_type = logical_struct_field.data_type(); if source_type != target_type && (source_type.is_decimal() || target_type.is_decimal()) { - return Ok(None); + let Some(target_field) = retain_field_path(cast.target_field(), &field_path) + else { + return Ok(None); + }; + let mut args = get_field_expr.args().to_vec(); + args[0] = Arc::new(CastExpr::new_with_target_field( + Arc::clone(inner), + target_field, + Some(cast.cast_options().clone()), + )); + return Arc::clone(expr).with_new_children(args).map(Some); } // Rebuild `get_field` over the uncast struct so its return field is @@ -1616,7 +1645,7 @@ mod tests { let rewriter = DefaultPhysicalExprAdapterRewriter { logical_file_schema: Arc::new(logical_schema), physical_file_schema: Arc::new(physical_schema), - generated_struct_casts: Vec::new(), + generated_struct_casts: HashMap::new(), }; // Test that when a field exists in physical schema, it returns None @@ -1915,6 +1944,70 @@ mod tests { Ok(()) } + #[test] + fn test_narrow_decimal_struct_cast_ignores_siblings() -> Result<()> { + use arrow::array::ArrayRef; + use datafusion_physical_expr::planner::logical2physical; + + for nested in [false, true] { + let mut physical_fields = vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Utf8, true), + ]; + let mut logical_fields = vec![ + Field::new("x", DataType::Decimal128(10, 2), true).with_metadata( + HashMap::from([("logical_meta".to_string(), "1".to_string())]), + ), + Field::new("y", DataType::Int32, true), + ]; + let mut column = Arc::new(StructArray::new( + physical_fields.clone().into(), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(StringArray::from(vec!["bad"])), + ], + None, + )) as ArrayRef; + let mut args = vec![datafusion_expr::col("s")]; + if nested { + physical_fields = vec![ + Field::new("inner", DataType::Struct(physical_fields.into()), true), + Field::new("y", DataType::Utf8, true), + ]; + logical_fields = vec![ + Field::new("inner", DataType::Struct(logical_fields.into()), true), + Field::new("y", DataType::Int32, true), + ]; + column = Arc::new(StructArray::new( + physical_fields.clone().into(), + vec![column, Arc::new(StringArray::from(vec!["bad"]))], + None, + )); + args.push(datafusion_expr::lit("inner")); + } + args.push(datafusion_expr::lit("x")); + let (logical_schema, physical_schema) = + struct_schemas(physical_fields, logical_fields); + let expr = logical2physical( + &datafusion_functions::core::get_field().call(args), + &logical_schema, + ); + let expected_field = expr.return_field(&logical_schema)?; + let batch = RecordBatch::try_new(Arc::clone(&physical_schema), vec![column])?; + let adapter = DefaultPhysicalExprAdapterFactory + .create(logical_schema, Arc::clone(&physical_schema))?; + let rewritten = adapter.rewrite(expr)?; + assert_eq!(rewritten.return_field(&physical_schema)?, expected_field); + let values = rewritten.evaluate(&batch)?.into_array(1)?; + assert_eq!( + ScalarValue::try_from_array(&values, 0)?, + ScalarValue::Decimal128(Some(100), 10, 2), + "nested={nested}" + ); + } + Ok(()) + } + /// A struct field that only differs in a nested leaf type still ends up /// with a single cast on the extracted field. #[test] From 801bb0ecee511460488da866318a44ee4a5f5149 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 26 Aug 2026 17:30:54 +0000 Subject: [PATCH 3/5] docs: clarify struct-cast rewrite invariants Document the traversal order, per-rewrite lifetime, and retained Arc ownership required by generated-cast tracking, including why provenance stays local. Explain how retaining only the selected cast-target path avoids sibling conversions while preserving the all-null Struct shortcut. --- .../physical-expr-adapter/src/schema_rewriter.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 0b681c0acb3cf..833fd4e0b496b 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -272,8 +272,12 @@ impl PhysicalExprAdapter for DefaultPhysicalExprAdapter { struct DefaultPhysicalExprAdapterRewriter { logical_file_schema: SchemaRef, physical_file_schema: SchemaRef, - // Retain generated casts so their pointer identity remains reliable even - // after a wider cast has been removed from the expression tree. + // A fresh map is created for each `rewrite()` call. Tracking relies on + // bottom-up `transform` traversal: a generated child cast is recorded before + // its parent `get_field` sees the same Arc allocation. Owned Arc clones keep + // recorded allocations alive even after removal from the tree, preventing + // pointer-address reuse. Keeping provenance here avoids adding markers to + // expression types or threading it through rewrite results. generated_struct_casts: HashMap<*const (), Arc>, } @@ -313,7 +317,9 @@ fn resolve_field_path<'a>( } } -/// Retain a field path without changing its ancestors' metadata or nullability. +/// Retain only the selected field path in a cast target, preserving its Struct +/// ancestors' metadata and nullability. This excludes unselected sibling +/// conversions while keeping the all-null Struct shortcut for decimal casts. fn retain_field_path(field: &FieldRef, path: &[&str]) -> Option { let Some((name, rest)) = path.split_first() else { return Some(Arc::clone(field)); From e79e91236e662d3084c7656eae78751027d1633d Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 26 Aug 2026 23:14:54 +0000 Subject: [PATCH 4/5] fix: preserve null semantics for decimal containers --- datafusion/core/tests/parquet/expr_adapter.rs | 120 ++++--- .../src/projection_read_plan.rs | 4 +- .../src/schema_rewriter.rs | 323 +++++++++++++++--- 3 files changed, 348 insertions(+), 99 deletions(-) diff --git a/datafusion/core/tests/parquet/expr_adapter.rs b/datafusion/core/tests/parquet/expr_adapter.rs index 2581c9618d279..dfed2eb5bda74 100644 --- a/datafusion/core/tests/parquet/expr_adapter.rs +++ b/datafusion/core/tests/parquet/expr_adapter.rs @@ -840,69 +840,77 @@ async fn test_explicit_struct_cast_projection_preserves_sibling_errors() -> Resu #[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}; - let physical_fields: Fields = vec![Field::new("x", DataType::Utf8, true)].into(); - let batch = RecordBatch::try_from_iter(vec![ - ("row_id", Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef), + for (physical_type, logical_type) in [ + (DataType::Utf8, DataType::Decimal128(10, -1)), ( - "s", - Arc::new(StructArray::new( - physical_fields, - vec![Arc::new(StringArray::from(vec![None::<&str>, None]))], - Some(NullBuffer::new_null(2)), - )) as ArrayRef, + DataType::new_list(DataType::Utf8, true), + DataType::new_list(DataType::Decimal128(10, -1), true), ), - ])?; - 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, -1), true)].into(), + ] { + 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, ), - true, - ), - ])); - let store = Arc::new(InMemory::new()) as Arc; - 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 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; + 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::(), + expected_rows, + "physical_type={physical_type:?}, pushdown_filters={pushdown_filters}, predicate={predicate}" ); } - let batches = collect(plan, ctx.task_ctx()).await?; - assert_eq!( - batches.iter().map(RecordBatch::num_rows).sum::(), - expected_rows, - "pushdown_filters={pushdown_filters}, predicate={predicate}" - ); } } Ok(()) diff --git a/datafusion/datasource-parquet/src/projection_read_plan.rs b/datafusion/datasource-parquet/src/projection_read_plan.rs index 29bcc79c3d8ca..c10995f3d15d7 100644 --- a/datafusion/datasource-parquet/src/projection_read_plan.rs +++ b/datafusion/datasource-parquet/src/projection_read_plan.rs @@ -184,7 +184,9 @@ pub(crate) struct PushdownChecker<'schema> { cast_accesses: Vec, /// Whether to collect [`Self::cast_accesses`]. collect_cast_accesses: bool, - /// Allow field access through a retained Struct cast after schema adaptation. + /// 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. diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 833fd4e0b496b..c2b10bec384b4 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -29,7 +29,7 @@ use arrow::datatypes::{DataType, FieldRef, Fields, SchemaRef}; use datafusion_common::{ DataFusionError, Result, ScalarValue, exec_err, metadata::FieldMetadata, - nested_struct::validate_data_type_compatibility, + nested_struct::{requires_nested_struct_cast, validate_data_type_compatibility}, tree_node::{Transformed, TransformedResult, TreeNode}, }; use datafusion_functions::core::getfield::GetFieldFunc; @@ -317,6 +317,40 @@ fn resolve_field_path<'a>( } } +/// Whether a type or any nested value type matches the predicate. +fn contains_type(data_type: &DataType, predicate: &impl Fn(&DataType) -> bool) -> bool { + if predicate(data_type) { + return true; + } + match data_type { + DataType::List(field) + | DataType::LargeList(field) + | DataType::FixedSizeList(field, _) + | DataType::ListView(field) + | DataType::LargeListView(field) + | DataType::RunEndEncoded(_, field) => { + contains_type(field.data_type(), predicate) + } + DataType::Map(entries, _) => { + // The entries Struct is a layout wrapper, not a Struct-valued child. + let DataType::Struct(fields) = entries.data_type() else { + return false; + }; + fields + .iter() + .any(|field| contains_type(field.data_type(), predicate)) + } + DataType::Struct(fields) => fields + .iter() + .any(|field| contains_type(field.data_type(), predicate)), + DataType::Union(fields, _) => fields + .iter() + .any(|(_, field)| contains_type(field.data_type(), predicate)), + DataType::Dictionary(_, values) => contains_type(values, predicate), + _ => false, + } +} + /// Retain only the selected field path in a cast target, preserving its Struct /// ancestors' metadata and nullability. This excludes unselected sibling /// conversions while keeping the all-null Struct shortcut for decimal casts. @@ -486,15 +520,32 @@ impl DefaultPhysicalExprAdapterRewriter { FieldPathResolution::NotAStruct => return Ok(None), }; - // Decimal conversions can fail during setup even for all-null inputs, - // while a Struct cast skips its children when the parent is all null. + // Decimal conversions, including those inside containers, can fail + // during setup even for all-null inputs, while a Struct cast skips its + // children when the parent is all null. // Keep the Struct ancestors for that shortcut, but exclude unselected // siblings whose conversions may fail. Same-type metadata casts remain // safe to narrow to a scalar cast. + // A Struct-to-Struct leaf cast keeps its own shortcut and must remain + // 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. 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(_)); if source_type != target_type - && (source_type.is_decimal() || target_type.is_decimal()) + && !matches!( + (source_type, target_type), + (DataType::Struct(_), DataType::Struct(_)) + ) + && (contains_type(source_type, &DataType::is_decimal) + || contains_type(target_type, &DataType::is_decimal)) + && (source_type.is_decimal() + || target_type.is_decimal() + || requires_nested_struct_cast(source_type, target_type) + || (!contains_type(source_type, &is_struct) + && !contains_type(target_type, &is_struct))) { let Some(target_field) = retain_field_path(cast.target_field(), &field_path) else { @@ -1702,6 +1753,119 @@ mod tests { (logical, physical) } + fn decimal_cast_leaf_types(data_type: DataType) -> Vec { + let item = Arc::new(Field::new("item", data_type.clone(), true)); + vec![ + data_type.clone(), + DataType::List(Arc::clone(&item)), + DataType::LargeList(Arc::clone(&item)), + DataType::FixedSizeList(Arc::clone(&item), 2), + DataType::ListView(Arc::clone(&item)), + DataType::LargeListView(item), + DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct( + vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", data_type.clone(), true), + ] + .into(), + ), + false, + )), + false, + ), + DataType::Dictionary(Box::new(DataType::Int8), Box::new(data_type.clone())), + DataType::new_list( + DataType::Struct( + vec![Field::new("value", data_type.clone(), true)].into(), + ), + true, + ), + DataType::new_list(DataType::new_list(data_type, true), true), + ] + } + + #[test] + fn test_narrow_struct_cast_preserves_struct_unwrapping() -> Result<()> { + use arrow::array::{ + ArrayRef, Decimal128Array, DictionaryArray, Int8Array, ListArray, + }; + use arrow::buffer::OffsetBuffer; + use arrow::datatypes::Int8Type; + + let values = Arc::new(StructArray::new( + vec![Field::new("value", DataType::Int32, true)].into(), + vec![Arc::new(Int32Array::from(vec![1]))], + None, + )) as ArrayRef; + let dictionary = Arc::new(DictionaryArray::::try_new( + Int8Array::from(vec![0]), + values, + )?) as ArrayRef; + let expected_struct = Arc::new(StructArray::new( + vec![Field::new("value", DataType::Decimal128(10, 2), true)].into(), + vec![Arc::new( + Decimal128Array::from(vec![100]).with_precision_and_scale(10, 2)?, + )], + None, + )) as ArrayRef; + let wrap_list = |values: ArrayRef| -> ArrayRef { + Arc::new(ListArray::new( + Arc::new(Field::new("item", values.data_type().clone(), true)), + OffsetBuffer::from_lengths([1]), + values, + None, + )) + }; + let wrap_dictionary = |values: ArrayRef| -> ArrayRef { + Arc::new( + DictionaryArray::::try_new(Int8Array::from(vec![0]), values) + .unwrap(), + ) + }; + for (label, physical, expected) in [ + ( + "direct Dictionary", + Arc::clone(&dictionary), + Arc::clone(&expected_struct), + ), + ( + "List of Dictionary", + wrap_list(Arc::clone(&dictionary)), + wrap_list(Arc::clone(&expected_struct)), + ), + ( + "Dictionary of Dictionary", + wrap_dictionary(dictionary), + wrap_dictionary(expected_struct), + ), + ] { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new("x", physical.data_type().clone(), true)], + vec![Field::new("x", expected.data_type().clone(), true)], + ); + let DataType::Struct(fields) = physical_schema.field(0).data_type() else { + unreachable!() + }; + let batch = RecordBatch::try_new( + Arc::clone(&physical_schema), + vec![Arc::new(StructArray::new( + fields.clone(), + vec![physical], + None, + ))], + )?; + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema)?; + let rewritten = adapter.rewrite(get_field_expr(&logical_schema, "s", "x"))?; + let actual = rewritten.evaluate(&batch)?.into_array(1)?; + assert_eq!(actual.to_data(), expected.to_data(), "{label}"); + } + Ok(()) + } + /// `s['x']` where the file stores `x` as `Int32` and the table declares /// `Int64` must cast the extracted field, not the whole struct, so that /// the column stays visible under the `get_field`. @@ -1709,29 +1873,37 @@ mod tests { /// See . #[test] fn test_narrow_struct_cast_to_field_access() { - let (logical_schema, physical_schema) = struct_schemas( - vec![Field::new("x", DataType::Int32, true)], - vec![Field::new("x", DataType::Int64, true)], - ); - - let adapter = DefaultPhysicalExprAdapterFactory - .create(Arc::clone(&logical_schema), physical_schema) - .unwrap(); - let rewritten = adapter - .rewrite(get_field_expr(&logical_schema, "s", "x")) - .unwrap(); + for (physical_type, logical_type) in [ + (DataType::Int32, DataType::Int64), + ( + DataType::new_list(DataType::Int32, true), + DataType::new_list(DataType::Int64, true), + ), + ] { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new("x", physical_type.clone(), true)], + vec![Field::new("x", logical_type.clone(), true)], + ); - let cast = assert_cast_expr(&rewritten); - assert_eq!(cast.cast_type(), &DataType::Int64); - let get_field = cast - .expr() - .downcast_ref::() - .expect("Expected get_field under the cast"); - assert_eq!(get_field.return_type(), &DataType::Int32); - assert!( - get_field.args()[0].downcast_ref::().is_some(), - "the struct column must not be hidden behind a cast, got: {rewritten}" - ); + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let rewritten = adapter + .rewrite(get_field_expr(&logical_schema, "s", "x")) + .unwrap(); + + let cast = assert_cast_expr(&rewritten); + assert_eq!(cast.cast_type(), &logical_type); + let get_field = cast + .expr() + .downcast_ref::() + .expect("Expected get_field under the cast"); + assert_eq!(get_field.return_type(), &physical_type); + assert!( + get_field.args()[0].downcast_ref::().is_some(), + "the struct column must not be hidden behind a cast, got: {rewritten}" + ); + } } /// Selecting one field of an explicit cast must still evaluate sibling @@ -1875,11 +2047,40 @@ mod tests { fn test_narrow_struct_cast_preserves_all_null_decimal_casts() -> Result<()> { use arrow::array::new_null_array; use arrow::buffer::NullBuffer; + use arrow::datatypes::{UnionFields, UnionMode}; for (physical_type, logical_type) in [ (DataType::Decimal128(38, -38), DataType::Decimal128(38, 38)), (DataType::Utf8, DataType::Decimal128(10, -1)), - ] { + (DataType::Decimal128(38, -39), DataType::Int64), + ] + .into_iter() + .flat_map(|(physical, logical)| { + decimal_cast_leaf_types(physical) + .into_iter() + .zip(decimal_cast_leaf_types(logical)) + }) + // A directly decimal target must still retain the ancestor even when + // an unrelated Union arm contains a Struct. + .chain([( + DataType::Union( + UnionFields::try_new( + [0, 1], + [ + Field::new("string", DataType::Utf8, true), + Field::new( + "struct", + DataType::Struct( + vec![Field::new("z", DataType::Int32, true)].into(), + ), + true, + ), + ], + )?, + UnionMode::Dense, + ), + DataType::Decimal128(10, -1), + )]) { let (logical_schema, physical_schema) = struct_schemas( vec![Field::new("x", physical_type.clone(), true)], vec![Field::new("x", logical_type, true)], @@ -1918,8 +2119,11 @@ mod tests { #[test] fn test_narrow_struct_cast_keeps_matching_decimal_fields_optimized() -> Result<()> { - let decimal_type = DataType::Decimal128(10, -1); - for change_metadata in [false, true] { + for (decimal_type, change_metadata) in + decimal_cast_leaf_types(DataType::Decimal128(10, -1)) + .into_iter() + .flat_map(|data_type| [(data_type.clone(), false), (data_type, true)]) + { let physical_field = Field::new("x", decimal_type.clone(), true); let logical_field = if change_metadata { physical_field.clone().with_metadata(HashMap::from([( @@ -1955,23 +2159,46 @@ mod tests { use arrow::array::ArrayRef; use datafusion_physical_expr::planner::logical2physical; - for nested in [false, true] { + for (nested, list_leaf) in + [(false, false), (true, false), (false, true), (true, true)] + { + let (physical_type, logical_type, x, expected) = if list_leaf { + ( + DataType::new_list(DataType::Int32, true), + DataType::new_list(DataType::Decimal128(10, 2), true), + ScalarValue::new_list( + &[ScalarValue::Int32(Some(1))], + &DataType::Int32, + true, + ) as ArrayRef, + ScalarValue::List(ScalarValue::new_list( + &[ScalarValue::Decimal128(Some(100), 10, 2)], + &DataType::Decimal128(10, 2), + true, + )), + ) + } else { + ( + DataType::Int32, + DataType::Decimal128(10, 2), + Arc::new(Int32Array::from(vec![1])) as ArrayRef, + ScalarValue::Decimal128(Some(100), 10, 2), + ) + }; let mut physical_fields = vec![ - Field::new("x", DataType::Int32, true), + Field::new("x", physical_type, true), Field::new("y", DataType::Utf8, true), ]; let mut logical_fields = vec![ - Field::new("x", DataType::Decimal128(10, 2), true).with_metadata( - HashMap::from([("logical_meta".to_string(), "1".to_string())]), - ), + Field::new("x", logical_type, true).with_metadata(HashMap::from([( + "logical_meta".to_string(), + "1".to_string(), + )])), Field::new("y", DataType::Int32, true), ]; let mut column = Arc::new(StructArray::new( physical_fields.clone().into(), - vec![ - Arc::new(Int32Array::from(vec![1])), - Arc::new(StringArray::from(vec!["bad"])), - ], + vec![x, Arc::new(StringArray::from(vec!["bad"]))], None, )) as ArrayRef; let mut args = vec![datafusion_expr::col("s")]; @@ -2007,8 +2234,8 @@ mod tests { let values = rewritten.evaluate(&batch)?.into_array(1)?; assert_eq!( ScalarValue::try_from_array(&values, 0)?, - ScalarValue::Decimal128(Some(100), 10, 2), - "nested={nested}" + expected, + "nested={nested}, list_leaf={list_leaf}" ); } Ok(()) @@ -2021,12 +2248,24 @@ mod tests { let (logical_schema, physical_schema) = struct_schemas( vec![Field::new( "inner", - DataType::Struct(vec![Field::new("x", DataType::Utf8, true)].into()), + DataType::Struct( + vec![ + Field::new("x", DataType::Utf8, true), + Field::new("y", DataType::Utf8, true), + ] + .into(), + ), true, )], vec![Field::new( "inner", - DataType::Struct(vec![Field::new("x", DataType::Utf8View, true)].into()), + DataType::Struct( + vec![ + Field::new("x", DataType::Utf8View, true), + Field::new("y", DataType::Decimal128(10, -1), true), + ] + .into(), + ), true, )], ); From ec4b3f282cfb82693cd1fac89eced5f7f5a2dd14 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Thu, 27 Aug 2026 21:13:53 +0000 Subject: [PATCH 5/5] perf: prune retained Struct cast reads in Parquet filters Reuse projection read clipping for retained Struct casts while preserving every conversion named by the cast target and sizing only selected leaves. Add SQL coverage for explicit sibling errors and all-null scalar/List decimal adaptation, strengthen partial-clipping coverage, and clarify why Struct ancestors remain around the covered decimal conversions. Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> --- .../src/projection_read_plan.rs | 79 +++++++---- .../datasource-parquet/src/row_filter.rs | 114 ++++++++++++++-- .../src/schema_rewriter.rs | 9 +- .../test_files/schema_evolution_nested.slt | 127 +++++++++++++++++- 4 files changed, 286 insertions(+), 43 deletions(-) diff --git a/datafusion/datasource-parquet/src/projection_read_plan.rs b/datafusion/datasource-parquet/src/projection_read_plan.rs index c10995f3d15d7..9c7028cb773b1 100644 --- a/datafusion/datasource-parquet/src/projection_read_plan.rs +++ b/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, /// 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, /// 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, - /// 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, } @@ -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) { // 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, &[], diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index 712402a012358..833764ba0809f 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/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)?; @@ -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,7 +1312,7 @@ 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!( @@ -1301,8 +1320,12 @@ mod test { "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 diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index c2b10bec384b4..2548ffc6fb1b7 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/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)` to `Struct`). + // 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(_)); diff --git a/datafusion/sqllogictest/test_files/schema_evolution_nested.slt b/datafusion/sqllogictest/test_files/schema_evolution_nested.slt index 53bc16fe51508..ce6b50d5d104b 100644 --- a/datafusion/sqllogictest/test_files/schema_evolution_nested.slt +++ b/datafusion/sqllogictest/test_files/schema_evolution_nested.slt @@ -16,7 +16,7 @@ # under the License. ########## -# End-user-facing happy-path coverage for nested list/struct Parquet schema evolution. +# End-user-facing coverage for nested list/struct Parquet schema evolution. # # These tests generate mixed-schema parquet files through SQL COPY statements and # query them through CREATE EXTERNAL TABLE, rather than constructing batches @@ -122,3 +122,128 @@ FROM large_list_messages; ---- 1 10 NULL 2 30 eth + +########## +# Schema adaptation widens x from Int32 to BIGINT. An explicit SQL Struct cast +# also converts y, so selecting only x must not hide that invalid conversion. +########## + +statement ok +COPY ( + SELECT named_struct('x', arrow_cast(1, 'Int32'), 'y', arrow_cast('bad', 'Utf8')) AS s +) TO 'test_files/scratch/schema_evolution_nested/explicit_struct_cast/data.parquet' +STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE explicit_struct_cast ( + s STRUCT +) +STORED AS PARQUET +LOCATION 'test_files/scratch/schema_evolution_nested/explicit_struct_cast/'; + +statement error While casting struct field 'y' +SELECT get_field(CAST(s AS STRUCT), 'x') FROM explicit_struct_cast; + +########## +# 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. +########## + +# Disable statistics-based shortcuts so the filters below evaluate the values. +statement ok +SET datafusion.execution.collect_statistics = false; + +statement ok +SET datafusion.execution.parquet.pruning = false; + +statement ok +SET datafusion.execution.parquet.enable_page_index = false; + +statement ok +SET datafusion.execution.parquet.pushdown_filters = false; + +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 +) +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> +) +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 + +# The same null-parent semantics apply with and without Parquet filter pushdown. +query I +SELECT row_id FROM null_struct_decimal WHERE get_field(s, 'a') IS NULL; +---- +1 + +query I +SELECT row_id FROM null_struct_list_decimal WHERE get_field(s, 'a') IS NULL; +---- +1 + +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +query I +SELECT row_id FROM null_struct_decimal WHERE get_field(s, 'a') IS NULL; +---- +1 + +query I +SELECT row_id FROM null_struct_list_decimal WHERE get_field(s, 'a') IS NULL; +---- +1 + +statement ok +RESET datafusion.execution.collect_statistics; + +statement ok +RESET datafusion.execution.parquet.pruning; + +statement ok +RESET datafusion.execution.parquet.enable_page_index; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters;