From 59da02fbb80f7886051f1e1979447191c6b3a088 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 19:19:03 +0000 Subject: [PATCH 1/7] fix: apply struct field filters when the file schema needs adaptation When the declared table schema differs from the physical file schema for a struct column, the expression adapter wraps the whole struct column in a cast, so `s['x']` becomes `get_field(cast(s AS Struct<..>), 'x')`. That hides the column from consumers that pattern match on `get_field(column, 'f')`. The Parquet scan is one such consumer: it decides at planning time (against the table schema) that a struct-field predicate can be evaluated as a row filter and reports it as fully handled, so `FilterExec` is removed from the plan. At runtime the row filter builder no longer recognizes the adapted expression, silently drops the predicate, and the query returns unfiltered rows. Narrow the cast to the field that is actually read: `get_field(cast(s AS Struct<..>), 'x')` becomes `cast(get_field(s, 'x') AS )`. This keeps the column visible under the `get_field`, and also avoids materializing a whole cast struct just to read one field. Fields that are missing from the file collapse to a typed null literal, matching what the struct cast would have produced. `get_field` on a Map column is a runtime key lookup rather than a schema-level field access, so those keep the whole-column cast. Closes #24109. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01MebN5PsVnYvXUeVKju5K7P --- .../datasource-parquet/src/opener/mod.rs | 105 +++++++ .../src/schema_rewriter.rs | 285 ++++++++++++++++++ .../test_files/parquet_filter_pushdown.slt | 47 +++ 3 files changed, 437 insertions(+) diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index b3ce024d66f1f..f7aad78a33274 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -3920,4 +3920,109 @@ mod test { assert_eq!(rows, 5); } } + + /// Filters on struct fields (`s['x'] = 200`) must still be applied when the + /// table schema disagrees with the physical file schema, which forces the + /// expression adapter to insert a cast. + /// + /// See . + mod struct_field_pushdown { + use super::*; + use arrow::array::{Int32Array, StructArray}; + use arrow::datatypes::Fields; + use datafusion_functions::core::get_field; + + /// Writes `s: Struct` with `x` values 100, 200, 300 and + /// returns the physical file schema plus the written file. + async fn write_struct_file( + store: &Arc, + path: &str, + ) -> (SchemaRef, PartitionedFile) { + let physical_struct_fields: Fields = + vec![Field::new("x", DataType::Int32, true)].into(); + let struct_array = StructArray::new( + physical_struct_fields.clone(), + vec![Arc::new(Int32Array::from(vec![100, 200, 300])) as _], + None, + ); + let schema = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(physical_struct_fields), + true, + )])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(struct_array) as _], + ) + .unwrap(); + let data_size = write_parquet(Arc::clone(store), path, batch).await; + let file = + PartitionedFile::new(path.to_string(), u64::try_from(data_size).unwrap()); + (schema, file) + } + + /// `s: Struct` — the same column as on disk, but with a + /// wider leaf type so the adapter has to insert a cast. + fn logical_schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(vec![Field::new("x", DataType::Int64, true)].into()), + true, + )])) + } + + fn struct_field_eq( + schema: &SchemaRef, + value: ScalarValue, + ) -> Arc { + let expr = get_field().call(vec![col("s"), lit("x")]).eq(lit(value)); + logical2physical(&expr, schema) + } + + #[tokio::test] + async fn test_struct_field_filter_with_schema_adaptation() { + let store = Arc::new(InMemory::new()) as Arc; + let (_physical_schema, file) = + write_struct_file(&store, "struct_cast.parquet").await; + let logical_schema = logical_schema(); + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&logical_schema)) + .with_projection_indices(&[0]) + .with_predicate(struct_field_eq( + &logical_schema, + ScalarValue::Int64(Some(200)), + )) + .with_pushdown_filters(true) + .build(); + + let stream = open_file(&opener, file).await.unwrap(); + let (_batches, rows) = count_batches_and_rows(stream).await; + assert_eq!(rows, 1, "predicate on struct field must be applied"); + } + + /// Control: with matching schemas the filter has always worked. + #[tokio::test] + async fn test_struct_field_filter_without_schema_adaptation() { + let store = Arc::new(InMemory::new()) as Arc; + let (physical_schema, file) = + write_struct_file(&store, "struct_nocast.parquet").await; + + let opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&physical_schema)) + .with_projection_indices(&[0]) + .with_predicate(struct_field_eq( + &physical_schema, + ScalarValue::Int32(Some(200)), + )) + .with_pushdown_filters(true) + .build(); + + let stream = open_file(&opener, file).await.unwrap(); + let (_batches, rows) = count_batches_and_rows(stream).await; + assert_eq!(rows, 1); + } + } } diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 152859b1b78fb..0191d326ae2c4 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -282,6 +282,10 @@ impl DefaultPhysicalExprAdapterRewriter { return Ok(Transformed::yes(transformed)); } + if let Some(transformed) = self.try_narrow_struct_cast(&expr)? { + return Ok(Transformed::yes(transformed)); + } + if let Some(column) = expr.downcast_ref::() { return self.rewrite_column(Arc::clone(&expr), column); } @@ -289,6 +293,98 @@ impl DefaultPhysicalExprAdapterRewriter { Ok(Transformed::no(expr)) } + /// Rewrite `get_field(cast(s AS Struct<..>), 'f')` into + /// `cast(get_field(s, 'f') AS )`. + /// + /// Expressions are rewritten bottom-up, so by the time we reach a + /// `get_field` node its struct argument has already been wrapped in a cast + /// by [`Self::rewrite_column`] whenever the logical and physical struct + /// types differ. Casting the whole struct just to read one field is + /// wasteful, and — more importantly — it hides the underlying column from + /// consumers that pattern match on `get_field(column, 'f')`. The Parquet + /// scan is one such consumer: it decides at planning time (against the + /// table schema) that a struct-field predicate can be evaluated as a row + /// filter, then fails to build that row filter at runtime because the + /// adapted expression no longer has a bare column under the `get_field`, + /// silently dropping the predicate and returning unfiltered rows. + /// + /// See . + /// + /// Only struct casts are narrowed. `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( + &self, + expr: &Arc, + ) -> Result>> { + let Some(get_field_expr) = + ScalarFunctionExpr::try_downcast_func::(expr.as_ref()) + else { + return Ok(None); + }; + let [source_expr, field_name_expr] = get_field_expr.args() else { + return Ok(None); + }; + let Some(cast) = source_expr.downcast_ref::() else { + return Ok(None); + }; + let Some(field_name) = field_name_expr + .downcast_ref::() + .and_then(|lit| lit.value().try_as_str().flatten()) + else { + return Ok(None); + }; + let DataType::Struct(logical_struct_fields) = cast.target_field().data_type() + else { + return Ok(None); + }; + let Some(logical_struct_field) = logical_struct_fields + .iter() + .find(|f| f.name() == field_name) + else { + return Ok(None); + }; + + let inner = cast.expr(); + let DataType::Struct(physical_struct_fields) = + inner.data_type(&self.physical_file_schema)? + else { + return Ok(None); + }; + let Some(physical_struct_field) = physical_struct_fields + .iter() + .find(|f| f.name() == field_name) + else { + // The file does not have this field at all, so reading it yields + // null. Note that the cast would have produced the same value: + // struct casts fill missing target fields with nulls. + let null_value = + ScalarValue::Null.cast_to(logical_struct_field.data_type())?; + return Ok(Some(Arc::new(Literal::new_with_metadata( + null_value, + Some(FieldMetadata::from(logical_struct_field.as_ref())), + )))); + }; + + // Rebuild `get_field` over the uncast struct so its return field is + // recomputed from the physical field type. + let extracted = Arc::new(ScalarFunctionExpr::try_new( + Arc::new(get_field_expr.fun().clone()), + vec![Arc::clone(inner), Arc::clone(field_name_expr)], + &self.physical_file_schema, + Arc::new(get_field_expr.config_options().clone()), + )?) as Arc; + + if physical_struct_field == logical_struct_field { + return Ok(Some(extracted)); + } + Ok(Some(Arc::new(CastExpr::new_with_target_field( + extracted, + Arc::clone(logical_struct_field), + Some(cast.cast_options().clone()), + )))) + } + /// Attempt to rewrite struct field access expressions to return null if the field does not exist in the physical schema. /// Note that this does *not* handle nested struct fields, only top-level struct field access. /// See for more details. @@ -1426,6 +1522,195 @@ mod tests { // datafusion/core/tests/parquet/schema_adapter.rs provide better coverage for this functionality. } + /// Build `get_field(column, 'field')` against `schema`. + fn get_field_expr( + schema: &Schema, + column: &str, + field: &str, + ) -> Arc { + let index = schema.index_of(column).unwrap(); + Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + vec![ + Arc::new(Column::new(column, index)), + Arc::new(Literal::new(ScalarValue::from(field))), + ], + schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + ) + .unwrap(), + ) + } + + fn struct_schemas( + physical_fields: Vec, + logical_fields: Vec, + ) -> (SchemaRef, SchemaRef) { + let physical = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(physical_fields.into()), + true, + )])); + let logical = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(logical_fields.into()), + true, + )])); + (logical, physical) + } + + /// `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`. + /// + /// 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(); + + 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}" + ); + } + + /// A struct field that only differs in a nested leaf type still ends up + /// with a single cast on the extracted field. + #[test] + fn test_narrow_struct_cast_nested_field_access() { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new( + "inner", + DataType::Struct(vec![Field::new("x", DataType::Utf8, true)].into()), + true, + )], + vec![Field::new( + "inner", + DataType::Struct(vec![Field::new("x", DataType::Utf8View, true)].into()), + true, + )], + ); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let outer = get_field_expr(&logical_schema, "s", "inner"); + let expr = Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + vec![outer, Arc::new(Literal::new(ScalarValue::from("x")))], + &logical_schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + ) + .unwrap(), + ) as Arc; + + let rewritten = adapter.rewrite(expr).unwrap(); + + let cast = assert_cast_expr(&rewritten); + assert_eq!(cast.cast_type(), &DataType::Utf8View); + let outer_get_field = cast + .expr() + .downcast_ref::() + .expect("Expected get_field under the cast"); + let inner_get_field = outer_get_field.args()[0] + .downcast_ref::() + .expect("Expected a nested get_field"); + assert!( + inner_get_field.args()[0].downcast_ref::().is_some(), + "the struct column must not be hidden behind a cast, got: {rewritten}" + ); + } + + /// Accessing a field the file does not have yields a typed null literal. + #[test] + fn test_narrow_struct_cast_missing_field() { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new("x", DataType::Int32, true)], + vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Utf8, true), + ], + ); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let rewritten = adapter + .rewrite(get_field_expr(&logical_schema, "s", "y")) + .unwrap(); + + let literal = rewritten + .downcast_ref::() + .expect("Expected a null literal"); + assert_eq!(*literal.value(), ScalarValue::Utf8(None)); + } + + /// `get_field` on a Map column is a runtime key lookup, not a schema-level + /// field access, so the map value must keep its cast. + #[test] + fn test_map_field_access_keeps_cast() { + let map_type = |value_type: DataType| { + DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct( + vec![ + Field::new("keys", DataType::Utf8, false), + Field::new("values", value_type, true), + ] + .into(), + ), + false, + )), + false, + ) + }; + let physical_schema = Arc::new(Schema::new(vec![Field::new( + "s", + map_type(DataType::Int32), + true, + )])); + let logical_schema = Arc::new(Schema::new(vec![Field::new( + "s", + map_type(DataType::Int64), + true, + )])); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let rewritten = adapter + .rewrite(get_field_expr(&logical_schema, "s", "k")) + .unwrap(); + + let get_field = rewritten + .downcast_ref::() + .expect("Expected the get_field to be preserved"); + assert!( + get_field.args()[0].downcast_ref::().is_some(), + "map columns must keep the whole-column cast, got: {rewritten}" + ); + } + // ============================================================================ // BatchAdapterFactory and BatchAdapter tests // ============================================================================ diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index cb3be93191fb2..e2874464bcbd7 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -890,6 +890,53 @@ set datafusion.execution.parquet.pushdown_filters = false; statement ok DROP TABLE t_struct_filter; +########## +# Regression test for https://github.com/apache/datafusion/issues/24109 +# +# When the declared table schema differs from the physical file schema, the +# expression adapter inserts a cast. Casting the whole struct column hid the +# column from the row filter builder: the scan reported `s['x'] = 200` as +# fully handled (so `FilterExec` was removed from the plan) but then could not +# build the row filter, silently dropping the predicate and returning all rows. +########## + +statement ok +set datafusion.execution.parquet.pushdown_filters = true; + +statement ok +COPY ( + SELECT + column1 as id, + named_struct('x', arrow_cast(column2, 'Int32')) as s + FROM VALUES (1, 100), (2, 200), (3, 300) +) TO 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet' +STORED AS PARQUET; + +# `x` is stored as Int32 in the file but declared as BIGINT here, which forces +# the scan to adapt the struct column. +statement ok +CREATE EXTERNAL TABLE t_struct_schema_cast (id BIGINT, s STRUCT) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet'; + +query II +SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] = 200; +---- +2 200 + +# Conjunction of a struct-field filter and a primitive filter. +query II +SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] > 100 AND id > 2; +---- +3 300 + +# Clean up +statement ok +set datafusion.execution.parquet.pushdown_filters = false; + +statement ok +DROP TABLE t_struct_schema_cast; + ########## # Regression test for https://github.com/apache/datafusion/issues/20937 # From ebe6b18be1144a6a706529a4b3d6a5caa489cf88 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 19:54:27 +0000 Subject: [PATCH 2/7] fix: narrow struct casts through the whole get_field key path `get_field` has a flattened multi-key form: the logical simplifier rewrites `s['a']['b']` into `get_field(s, 'a', 'b')`. The narrowing rule only matched the two-argument form, so nested field access kept the whole-struct cast and stayed exposed to the wrong-results bug it was meant to fix. Resolve the full key path through nested struct fields on both the logical (cast target) and physical sides, and rebuild `get_field` with every key preserved. A path whose leaf is missing from the file still collapses to a typed null literal; a path that runs through a non-struct field is left alone. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01MebN5PsVnYvXUeVKju5K7P --- .../src/schema_rewriter.rs | 204 +++++++++++++++--- .../test_files/parquet_filter_pushdown.slt | 24 +++ 2 files changed, 202 insertions(+), 26 deletions(-) diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 0191d326ae2c4..4dfeaba8aa74a 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -25,7 +25,7 @@ use std::hash::Hash; use std::sync::Arc; use arrow::array::RecordBatch; -use arrow::datatypes::{DataType, FieldRef, SchemaRef}; +use arrow::datatypes::{DataType, FieldRef, Fields, SchemaRef}; use datafusion_common::{ DataFusionError, Result, ScalarValue, exec_err, metadata::FieldMetadata, @@ -273,6 +273,35 @@ struct DefaultPhysicalExprAdapterRewriter { physical_file_schema: SchemaRef, } +/// Outcome of walking a `get_field` key path through nested struct fields. +enum FieldPathResolution<'a> { + /// The leaf field the path points at. + Found(&'a FieldRef), + /// Some key along the path does not exist, so the access reads as null. + Missing, + /// An intermediate field is not a struct, so the path cannot be resolved + /// statically. + NotAStruct, +} + +/// Follow a `get_field` key path (`['a', 'b']` for `s['a']['b']`) through +/// nested struct fields. +fn resolve_field_path<'a>(fields: &'a Fields, path: &[&str]) -> FieldPathResolution<'a> { + let Some((field_name, rest)) = path.split_first() else { + return FieldPathResolution::NotAStruct; + }; + let Some(field) = fields.iter().find(|f| f.name() == field_name) else { + return FieldPathResolution::Missing; + }; + if rest.is_empty() { + return FieldPathResolution::Found(field); + } + match field.data_type() { + DataType::Struct(nested_fields) => resolve_field_path(nested_fields, rest), + _ => FieldPathResolution::NotAStruct, + } +} + impl DefaultPhysicalExprAdapterRewriter { fn rewrite_expr( &self, @@ -310,6 +339,10 @@ impl DefaultPhysicalExprAdapterRewriter { /// /// See . /// + /// `get_field` also has a flattened multi-key form: `s['a']['b']` is + /// 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 /// runtime key lookup rather than a schema-level field access, so the map /// value must keep its cast. @@ -322,25 +355,36 @@ impl DefaultPhysicalExprAdapterRewriter { else { return Ok(None); }; - let [source_expr, field_name_expr] = get_field_expr.args() else { + let Some((source_expr, field_name_exprs)) = get_field_expr.args().split_first() + else { return Ok(None); }; - let Some(cast) = source_expr.downcast_ref::() else { + if field_name_exprs.is_empty() { return Ok(None); - }; - let Some(field_name) = field_name_expr - .downcast_ref::() - .and_then(|lit| lit.value().try_as_str().flatten()) - else { + } + let Some(cast) = source_expr.downcast_ref::() else { return Ok(None); }; + + // Every key has to be a string literal, otherwise the leaf field + // cannot be resolved statically. + let mut field_path = Vec::with_capacity(field_name_exprs.len()); + for field_name_expr in field_name_exprs { + let Some(field_name) = field_name_expr + .downcast_ref::() + .and_then(|lit| lit.value().try_as_str().flatten()) + else { + return Ok(None); + }; + field_path.push(field_name); + } + let DataType::Struct(logical_struct_fields) = cast.target_field().data_type() else { return Ok(None); }; - let Some(logical_struct_field) = logical_struct_fields - .iter() - .find(|f| f.name() == field_name) + let FieldPathResolution::Found(logical_struct_field) = + resolve_field_path(logical_struct_fields, &field_path) else { return Ok(None); }; @@ -351,26 +395,32 @@ impl DefaultPhysicalExprAdapterRewriter { else { return Ok(None); }; - let Some(physical_struct_field) = physical_struct_fields - .iter() - .find(|f| f.name() == field_name) - else { - // The file does not have this field at all, so reading it yields - // null. Note that the cast would have produced the same value: - // struct casts fill missing target fields with nulls. - let null_value = - ScalarValue::Null.cast_to(logical_struct_field.data_type())?; - return Ok(Some(Arc::new(Literal::new_with_metadata( - null_value, - Some(FieldMetadata::from(logical_struct_field.as_ref())), - )))); - }; + let physical_struct_field = + match resolve_field_path(&physical_struct_fields, &field_path) { + FieldPathResolution::Found(field) => field, + FieldPathResolution::Missing => { + // The file does not have this field at all, so reading it + // yields null. Note that the cast would have produced the + // same value: struct casts fill missing target fields with + // nulls. + let null_value = + ScalarValue::Null.cast_to(logical_struct_field.data_type())?; + return Ok(Some(Arc::new(Literal::new_with_metadata( + null_value, + Some(FieldMetadata::from(logical_struct_field.as_ref())), + )))); + } + FieldPathResolution::NotAStruct => 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()); + args.push(Arc::clone(inner)); + args.extend(field_name_exprs.iter().map(Arc::clone)); let extracted = Arc::new(ScalarFunctionExpr::try_new( Arc::new(get_field_expr.fun().clone()), - vec![Arc::clone(inner), Arc::clone(field_name_expr)], + args, &self.physical_file_schema, Arc::new(get_field_expr.config_options().clone()), )?) as Arc; @@ -1640,6 +1690,108 @@ mod tests { ); } + /// `s['inner']['x']` is simplified to the flattened `get_field(s, 'inner', + /// 'x')`, so the whole key path has to be resolved. + #[test] + fn test_narrow_struct_cast_flattened_field_path() { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new( + "inner", + DataType::Struct(vec![Field::new("x", DataType::Utf8, true)].into()), + true, + )], + vec![Field::new( + "inner", + DataType::Struct(vec![Field::new("x", DataType::Utf8View, true)].into()), + true, + )], + ); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let expr = Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + vec![ + Arc::new(Column::new("s", 0)), + Arc::new(Literal::new(ScalarValue::from("inner"))), + Arc::new(Literal::new(ScalarValue::from("x"))), + ], + &logical_schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + ) + .unwrap(), + ) as Arc; + + let rewritten = adapter.rewrite(expr).unwrap(); + + let cast = assert_cast_expr(&rewritten); + assert_eq!(cast.cast_type(), &DataType::Utf8View); + let get_field = cast + .expr() + .downcast_ref::() + .expect("Expected get_field under the cast"); + assert_eq!(get_field.return_type(), &DataType::Utf8); + assert_eq!( + get_field.args().len(), + 3, + "the full key path must be preserved, got: {rewritten}" + ); + assert!( + get_field.args()[0].downcast_ref::().is_some(), + "the struct column must not be hidden behind a cast, got: {rewritten}" + ); + } + + /// A key path whose leaf is missing from the file still resolves to a + /// typed null literal. + #[test] + fn test_narrow_struct_cast_flattened_field_path_missing_leaf() { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new( + "inner", + DataType::Struct(vec![Field::new("x", DataType::Int32, true)].into()), + true, + )], + vec![Field::new( + "inner", + DataType::Struct( + vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Utf8, true), + ] + .into(), + ), + true, + )], + ); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let expr = Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + vec![ + Arc::new(Column::new("s", 0)), + Arc::new(Literal::new(ScalarValue::from("inner"))), + Arc::new(Literal::new(ScalarValue::from("y"))), + ], + &logical_schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + ) + .unwrap(), + ) as Arc; + + let rewritten = adapter.rewrite(expr).unwrap(); + + let literal = rewritten + .downcast_ref::() + .expect("Expected a null literal"); + assert_eq!(*literal.value(), ScalarValue::Utf8(None)); + } + /// Accessing a field the file does not have yields a typed null literal. #[test] fn test_narrow_struct_cast_missing_field() { diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index e2874464bcbd7..afdd2f8fe8808 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -930,6 +930,27 @@ SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] > 100 AND id > 2; ---- 3 300 +# Same, for a nested field path. `s['inner']['x']` is simplified to the +# flattened `get_field(s, 'inner', 'x')`, which takes a different code path. +statement ok +COPY ( + SELECT + column1 as id, + named_struct('inner', named_struct('x', arrow_cast(column2, 'Int32'))) as s + FROM VALUES (1, 100), (2, 200), (3, 300) +) TO 'test_files/scratch/parquet_filter_pushdown/struct_nested_schema_cast.parquet' +STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE t_struct_nested_schema_cast (id BIGINT, s STRUCT>) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_nested_schema_cast.parquet'; + +query II +SELECT id, s['inner']['x'] FROM t_struct_nested_schema_cast WHERE s['inner']['x'] = 200; +---- +2 200 + # Clean up statement ok set datafusion.execution.parquet.pushdown_filters = false; @@ -937,6 +958,9 @@ set datafusion.execution.parquet.pushdown_filters = false; statement ok DROP TABLE t_struct_schema_cast; +statement ok +DROP TABLE t_struct_nested_schema_cast; + ########## # Regression test for https://github.com/apache/datafusion/issues/20937 # From 53c97c8f64fc029ba93cf5c464252577a834cd35 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 20:17:45 +0000 Subject: [PATCH 3/7] test: move struct field pushdown coverage to sqllogictest Drop the opener-level Rust tests in favour of SLT, which covers the same ground end to end through the planner. Adds a matching-schema control and a missing-field case alongside the existing adapted-schema tests, so the deleted Rust coverage is preserved. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01MebN5PsVnYvXUeVKju5K7P --- .../datasource-parquet/src/opener/mod.rs | 105 ------------------ .../test_files/parquet_filter_pushdown.slt | 28 +++++ 2 files changed, 28 insertions(+), 105 deletions(-) diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index f7aad78a33274..b3ce024d66f1f 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -3920,109 +3920,4 @@ mod test { assert_eq!(rows, 5); } } - - /// Filters on struct fields (`s['x'] = 200`) must still be applied when the - /// table schema disagrees with the physical file schema, which forces the - /// expression adapter to insert a cast. - /// - /// See . - mod struct_field_pushdown { - use super::*; - use arrow::array::{Int32Array, StructArray}; - use arrow::datatypes::Fields; - use datafusion_functions::core::get_field; - - /// Writes `s: Struct` with `x` values 100, 200, 300 and - /// returns the physical file schema plus the written file. - async fn write_struct_file( - store: &Arc, - path: &str, - ) -> (SchemaRef, PartitionedFile) { - let physical_struct_fields: Fields = - vec![Field::new("x", DataType::Int32, true)].into(); - let struct_array = StructArray::new( - physical_struct_fields.clone(), - vec![Arc::new(Int32Array::from(vec![100, 200, 300])) as _], - None, - ); - let schema = Arc::new(Schema::new(vec![Field::new( - "s", - DataType::Struct(physical_struct_fields), - true, - )])); - let batch = RecordBatch::try_new( - Arc::clone(&schema), - vec![Arc::new(struct_array) as _], - ) - .unwrap(); - let data_size = write_parquet(Arc::clone(store), path, batch).await; - let file = - PartitionedFile::new(path.to_string(), u64::try_from(data_size).unwrap()); - (schema, file) - } - - /// `s: Struct` — the same column as on disk, but with a - /// wider leaf type so the adapter has to insert a cast. - fn logical_schema() -> SchemaRef { - Arc::new(Schema::new(vec![Field::new( - "s", - DataType::Struct(vec![Field::new("x", DataType::Int64, true)].into()), - true, - )])) - } - - fn struct_field_eq( - schema: &SchemaRef, - value: ScalarValue, - ) -> Arc { - let expr = get_field().call(vec![col("s"), lit("x")]).eq(lit(value)); - logical2physical(&expr, schema) - } - - #[tokio::test] - async fn test_struct_field_filter_with_schema_adaptation() { - let store = Arc::new(InMemory::new()) as Arc; - let (_physical_schema, file) = - write_struct_file(&store, "struct_cast.parquet").await; - let logical_schema = logical_schema(); - - let opener = ParquetMorselizerBuilder::new() - .with_store(Arc::clone(&store)) - .with_schema(Arc::clone(&logical_schema)) - .with_projection_indices(&[0]) - .with_predicate(struct_field_eq( - &logical_schema, - ScalarValue::Int64(Some(200)), - )) - .with_pushdown_filters(true) - .build(); - - let stream = open_file(&opener, file).await.unwrap(); - let (_batches, rows) = count_batches_and_rows(stream).await; - assert_eq!(rows, 1, "predicate on struct field must be applied"); - } - - /// Control: with matching schemas the filter has always worked. - #[tokio::test] - async fn test_struct_field_filter_without_schema_adaptation() { - let store = Arc::new(InMemory::new()) as Arc; - let (physical_schema, file) = - write_struct_file(&store, "struct_nocast.parquet").await; - - let opener = ParquetMorselizerBuilder::new() - .with_store(Arc::clone(&store)) - .with_schema(Arc::clone(&physical_schema)) - .with_projection_indices(&[0]) - .with_predicate(struct_field_eq( - &physical_schema, - ScalarValue::Int32(Some(200)), - )) - .with_pushdown_filters(true) - .build(); - - let stream = open_file(&opener, file).await.unwrap(); - let (_batches, rows) = count_batches_and_rows(stream).await; - assert_eq!(rows, 1); - } - } } diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index afdd2f8fe8808..8e109444440a0 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -930,6 +930,28 @@ SELECT id, s['x'] FROM t_struct_schema_cast WHERE s['x'] > 100 AND id > 2; ---- 3 300 +# Control: the same file read through a schema that matches it exactly, so no +# cast is inserted. This path has always worked. +statement ok +CREATE EXTERNAL TABLE t_struct_no_schema_cast (id BIGINT, s STRUCT) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet'; + +query II +SELECT id, s['x'] FROM t_struct_no_schema_cast WHERE s['x'] = 200; +---- +2 200 + +# A field that the file does not have reads as null, so it filters nothing in. +statement ok +CREATE EXTERNAL TABLE t_struct_missing_field (id BIGINT, s STRUCT) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet'; + +query II +SELECT id, s['x'] FROM t_struct_missing_field WHERE s['missing'] = 200; +---- + # Same, for a nested field path. `s['inner']['x']` is simplified to the # flattened `get_field(s, 'inner', 'x')`, which takes a different code path. statement ok @@ -958,6 +980,12 @@ set datafusion.execution.parquet.pushdown_filters = false; statement ok DROP TABLE t_struct_schema_cast; +statement ok +DROP TABLE t_struct_no_schema_cast; + +statement ok +DROP TABLE t_struct_missing_field; + statement ok DROP TABLE t_struct_nested_schema_cast; From 602858d1836fff2c749fd599eca15d960bcfafbc Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 5 Aug 2026 20:57:33 +0000 Subject: [PATCH 4/7] test: cover the uncast and no-adaptation paths of the struct cast narrowing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Coverage analysis of the narrowing showed two reachable branches with no test behind them: a struct column that needs no adaptation at all, and one where only a sibling field forced the column-level cast, so the accessed field needs no cast of its own. Both matter — the first is the common case the rewrite must not disturb, the second is where the cast disappears rather than moving. The remaining uncovered branches in the function are guards against shapes that cannot reach it: a `get_field` with fewer than two arguments, and any key path running through a non-struct field, which the column-level cast validation rejects first. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01MebN5PsVnYvXUeVKju5K7P --- .../src/schema_rewriter.rs | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 4dfeaba8aa74a..f8fa45c04c6d2 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -1690,6 +1690,77 @@ mod tests { ); } + /// A struct column that needs no adaptation at all is left completely + /// alone — the narrowing must not disturb the common case. + #[test] + fn test_narrow_struct_cast_leaves_matching_schema_alone() { + let (logical_schema, physical_schema) = struct_schemas( + vec![Field::new("x", DataType::Int32, true)], + vec![Field::new("x", DataType::Int32, true)], + ); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let expr = get_field_expr(&logical_schema, "s", "x"); + + let rewritten = adapter.rewrite(Arc::clone(&expr)).unwrap(); + + assert_eq!( + rewritten.to_string(), + expr.to_string(), + "an unadapted struct column must pass through untouched" + ); + } + + /// When the accessed field has the same type in both schemas, the struct + /// cast disappears entirely rather than being replaced by a field cast: + /// only a sibling field forced the column-level cast in the first place. + #[test] + fn test_narrow_struct_cast_drops_cast_when_field_types_match() { + let (logical_schema, physical_schema) = struct_schemas( + vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Int32, true), + ], + vec![ + Field::new("x", DataType::Int32, true), + Field::new("y", DataType::Int64, true), + ], + ); + + let adapter = DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema) + .unwrap(); + let expr = Arc::new( + ScalarFunctionExpr::try_new( + Arc::new(datafusion_expr::ScalarUDF::from(GetFieldFunc::new())), + vec![ + Arc::new(Column::new("s", 0)), + Arc::new(Literal::new(ScalarValue::from("x"))), + ], + &logical_schema, + Arc::new(datafusion_common::config::ConfigOptions::default()), + ) + .unwrap(), + ) as Arc; + + let rewritten = adapter.rewrite(expr).unwrap(); + + assert!( + rewritten.downcast_ref::().is_none(), + "`x` has the same type in both schemas, so no cast is needed, got: {rewritten}" + ); + let get_field = rewritten + .downcast_ref::() + .expect("Expected a bare get_field"); + 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}" + ); + } + /// `s['inner']['x']` is simplified to the flattened `get_field(s, 'inner', /// 'x')`, so the whole key path has to be resolved. #[test] From 2bbe3d32c479e1bcafeb0caa717d0903b32932cc Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 17:54:04 +0000 Subject: [PATCH 5/7] test: keep nested pruning when a struct root is read whole and by field Narrowing the struct cast makes `DefaultPhysicalExprAdapter` produce a shape `build_read_plan_with_cast_clipping` did not handle at the time this PR was written: `SELECT s, s['y']` over a narrowed struct reads the whole column through the cast and the field through a `get_field` on the bare column, and a root carrying both access kinds fell back to reading every physical leaf. #24315 has since generalized that function to keep the union of the leaves both access kinds need, so the code fix this commit originally carried is no longer necessary; only the test expectations remain. `select s['x'] from narrow` now clips all the way down to `x` (146 -> 75 bytes) because the field access no longer hides behind a whole-struct cast, and `select s, s['y'] from narrow` still reads only the narrow leaves. Co-Authored-By: Claude Opus 5 --- .../parquet_nested_schema_pruning.slt | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt index 0aa171824fb5f..151accc1add01 100644 --- a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt +++ b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt @@ -125,20 +125,22 @@ explain analyze select s from full_schema; ---- Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219] -# `get_field` on a schema-narrowed struct becomes `get_field(CAST(s), 'x')`; -# the read clips to the cast target (every field the *narrow* schema -# declares), not further down to just `x`. The fair "nothing was clipped" -# baseline is therefore reading every physical leaf of `s` -# (`select s from full_schema` above), not the same `get_field` query against -# `full_schema` -- that one needs no cast at all and takes `get_field`'s own, -# more precise, single-leaf pushdown path. +# `get_field` on a schema-narrowed struct is rewritten to +# `CAST(get_field(s, 'x'))` rather than `get_field(CAST(s), 'x')`, so it takes +# `get_field`'s own single-leaf pushdown path: the read clips all the way down +# to `x`, not just to the fields the *narrow* schema declares. That is why +# this reads fewer bytes than `select s from narrow` above, which still needs +# every narrow leaf. query TT explain analyze select s['x'] from narrow; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=146] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=75] # Mixed access -- the whole (narrowed) column and a subfield of it -- still -# reads only the narrow schema's leaves. +# reads only the narrow schema's leaves. The whole-column read goes through +# the cast while the subfield read is a `get_field` on the bare column, so the +# root carries both kinds of access at once and the read keeps the union of +# the leaves they need. query TT explain analyze select s, s['y'] from narrow; ---- From 6b348509efd800d3dbbb2a443b82672435a85d40 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 21:41:57 +0000 Subject: [PATCH 6/7] review: document the rewrite's rationale and broaden the SQL coverage Addresses review feedback on #24125: - Say *why* the cast is narrowed (reading one field should not cost a whole struct; keeping the column visible keeps every `get_field(column, 'f')` consumer working) and why it is fixed in the adapter rather than by teaching one consumer to see through casts. - Drop the description of the pre-fix row-filter behaviour, which is an implementation detail that will date. - Note that the empty-path arm of `resolve_field_path` is a defensive default, not a claim about empty paths. - Run the compound filter against the matching-schema table too, and check that declaring a field the file lacks leaves the fields it does have answering correctly. - Trim the sqllogictest comments to what the queries demonstrate rather than how the read plan gets there. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01MebN5PsVnYvXUeVKju5K7P --- .../src/schema_rewriter.rs | 36 ++++++++++++++----- .../test_files/parquet_filter_pushdown.slt | 32 ++++++++++++----- .../parquet_nested_schema_pruning.slt | 14 +++----- 3 files changed, 54 insertions(+), 28 deletions(-) diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index f8fa45c04c6d2..72b4a750cc595 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -288,6 +288,9 @@ enum FieldPathResolution<'a> { /// nested struct fields. fn resolve_field_path<'a>(fields: &'a Fields, path: &[&str]) -> FieldPathResolution<'a> { let Some((field_name, rest)) = path.split_first() else { + // Defensive default: callers reject an empty key list before getting + // here, so this is unreachable rather than a claim that an empty path + // names a non-struct. return FieldPathResolution::NotAStruct; }; let Some(field) = fields.iter().find(|f| f.name() == field_name) else { @@ -328,16 +331,31 @@ impl DefaultPhysicalExprAdapterRewriter { /// Expressions are rewritten bottom-up, so by the time we reach a /// `get_field` node its struct argument has already been wrapped in a cast /// by [`Self::rewrite_column`] whenever the logical and physical struct - /// types differ. Casting the whole struct just to read one field is - /// wasteful, and — more importantly — it hides the underlying column from - /// consumers that pattern match on `get_field(column, 'f')`. The Parquet - /// scan is one such consumer: it decides at planning time (against the - /// table schema) that a struct-field predicate can be evaluated as a row - /// filter, then fails to build that row filter at runtime because the - /// adapted expression no longer has a bare column under the `get_field`, - /// silently dropping the predicate and returning unfiltered rows. + /// types differ. /// - /// See . + /// Narrowing that cast is worthwhile for two reasons: + /// + /// 1. Reading one field should not cost a whole struct. The wide form + /// casts every field of the column — including ones the query never + /// reads — to produce a value that is immediately discarded except for + /// one field. + /// 2. It keeps the column visible. Consumers throughout the codebase + /// pattern match on `get_field(column, 'f')` to recognise a struct + /// field access; a cast between the `get_field` and its column defeats + /// that match, and each such consumer then falls back to whatever it + /// does for an unrecognised expression. + /// + /// The Parquet scan is one such consumer, and the reason this is a + /// correctness fix rather than only an optimisation: it decides at + /// planning time, against the table schema, that a struct-field predicate + /// can be evaluated as a row filter, and reports the predicate as fully + /// handled. See . + /// + /// Fixing it here rather than teaching that one consumer to see through + /// casts is deliberate: the adapter is where the obscuring cast is + /// introduced, so every consumer benefits, and no consumer has to loosen + /// its pattern to accept arbitrary casts between a `get_field` and its + /// column. /// /// `get_field` also has a flattened multi-key form: `s['a']['b']` is /// simplified to `get_field(s, 'a', 'b')`, so the whole field path is diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index 8e109444440a0..f7d23001fcf50 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -891,13 +891,11 @@ statement ok DROP TABLE t_struct_filter; ########## -# Regression test for https://github.com/apache/datafusion/issues/24109 +# Filters on struct fields, with pushdown enabled, where the declared table +# schema differs from the physical file schema and the scan therefore has to +# adapt the struct column. # -# When the declared table schema differs from the physical file schema, the -# expression adapter inserts a cast. Casting the whole struct column hid the -# column from the row filter builder: the scan reported `s['x'] = 200` as -# fully handled (so `FilterExec` was removed from the plan) but then could not -# build the row filter, silently dropping the predicate and returning all rows. +# See https://github.com/apache/datafusion/issues/24109. ########## statement ok @@ -942,18 +940,34 @@ SELECT id, s['x'] FROM t_struct_no_schema_cast WHERE s['x'] = 200; ---- 2 200 -# A field that the file does not have reads as null, so it filters nothing in. +query II +SELECT id, s['x'] FROM t_struct_no_schema_cast WHERE s['x'] > 100 AND id > 2; +---- +3 300 + +# Declaring a field the file does not have must not disturb the fields it does +# have. statement ok CREATE EXTERNAL TABLE t_struct_missing_field (id BIGINT, s STRUCT) STORED AS PARQUET LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_schema_cast.parquet'; +query II +SELECT id, s['x'] FROM t_struct_missing_field WHERE s['x'] = 200; +---- +2 200 + +query II +SELECT id, s['x'] FROM t_struct_missing_field WHERE s['x'] > 100 AND id > 2; +---- +3 300 + +# The absent field itself reads as null, so it matches nothing. query II SELECT id, s['x'] FROM t_struct_missing_field WHERE s['missing'] = 200; ---- -# Same, for a nested field path. `s['inner']['x']` is simplified to the -# flattened `get_field(s, 'inner', 'x')`, which takes a different code path. +# Same, for a nested field path. statement ok COPY ( SELECT diff --git a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt index 151accc1add01..a99839ea70a0d 100644 --- a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt +++ b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt @@ -125,22 +125,16 @@ explain analyze select s from full_schema; ---- Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219] -# `get_field` on a schema-narrowed struct is rewritten to -# `CAST(get_field(s, 'x'))` rather than `get_field(CAST(s), 'x')`, so it takes -# `get_field`'s own single-leaf pushdown path: the read clips all the way down -# to `x`, not just to the fields the *narrow* schema declares. That is why -# this reads fewer bytes than `select s from narrow` above, which still needs -# every narrow leaf. +# Selecting a single field of `s` reads fewer bytes than selecting `s` itself +# (above): the read clips down to `x` rather than to every field the narrow +# schema declares. query TT explain analyze select s['x'] from narrow; ---- Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=75] # Mixed access -- the whole (narrowed) column and a subfield of it -- still -# reads only the narrow schema's leaves. The whole-column read goes through -# the cast while the subfield read is a `get_field` on the bare column, so the -# root carries both kinds of access at once and the read keeps the union of -# the leaves they need. +# reads only the narrow schema's leaves. query TT explain analyze select s, s['y'] from narrow; ---- From 6081cc14230f78f3c8a1bc23059914c31fb1ff42 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 12 Aug 2026 16:39:40 -0400 Subject: [PATCH 7/7] review: make an empty get_field key path unrepresentable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `resolve_field_path` had to handle an empty key path even though callers guarantee at least one key, and returned `NotAStruct` for it — which is not what that variant means. Take the first key as its own parameter so the invariant is carried by the signature and there is no empty path to resolve. The redundant `field_name_exprs.is_empty()` guard at the call site goes away with it, subsumed by the `split_first` the new signature requires. Co-Authored-By: Claude Opus 5 --- .../src/schema_rewriter.rs | 35 +++++++++++-------- 1 file changed, 20 insertions(+), 15 deletions(-) diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 72b4a750cc595..5347b7ff603fe 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -286,21 +286,25 @@ enum FieldPathResolution<'a> { /// Follow a `get_field` key path (`['a', 'b']` for `s['a']['b']`) through /// nested struct fields. -fn resolve_field_path<'a>(fields: &'a Fields, path: &[&str]) -> FieldPathResolution<'a> { - let Some((field_name, rest)) = path.split_first() else { - // Defensive default: callers reject an empty key list before getting - // here, so this is unreachable rather than a claim that an empty path - // names a non-struct. - return FieldPathResolution::NotAStruct; - }; +/// +/// The first key is taken separately from the rest so that the "at least one +/// key" invariant is carried by the signature: there is no empty path to +/// resolve. +fn resolve_field_path<'a>( + fields: &'a Fields, + field_name: &str, + rest: &[&str], +) -> FieldPathResolution<'a> { let Some(field) = fields.iter().find(|f| f.name() == field_name) else { return FieldPathResolution::Missing; }; - if rest.is_empty() { + let Some((next_field_name, rest)) = rest.split_first() else { return FieldPathResolution::Found(field); - } + }; match field.data_type() { - DataType::Struct(nested_fields) => resolve_field_path(nested_fields, rest), + DataType::Struct(nested_fields) => { + resolve_field_path(nested_fields, next_field_name, rest) + } _ => FieldPathResolution::NotAStruct, } } @@ -377,9 +381,6 @@ impl DefaultPhysicalExprAdapterRewriter { else { return Ok(None); }; - if field_name_exprs.is_empty() { - return Ok(None); - } let Some(cast) = source_expr.downcast_ref::() else { return Ok(None); }; @@ -396,13 +397,17 @@ impl DefaultPhysicalExprAdapterRewriter { }; field_path.push(field_name); } + // A `get_field` with no keys is not a field access we can narrow. + let Some((first_key, rest_keys)) = field_path.split_first() else { + return Ok(None); + }; let DataType::Struct(logical_struct_fields) = cast.target_field().data_type() else { return Ok(None); }; let FieldPathResolution::Found(logical_struct_field) = - resolve_field_path(logical_struct_fields, &field_path) + resolve_field_path(logical_struct_fields, first_key, rest_keys) else { return Ok(None); }; @@ -414,7 +419,7 @@ impl DefaultPhysicalExprAdapterRewriter { return Ok(None); }; let physical_struct_field = - match resolve_field_path(&physical_struct_fields, &field_path) { + match resolve_field_path(&physical_struct_fields, first_key, rest_keys) { FieldPathResolution::Found(field) => field, FieldPathResolution::Missing => { // The file does not have this field at all, so reading it