From fd23584a28eb53758476f7877e228b2752592deb Mon Sep 17 00:00:00 2001 From: grorge Date: Mon, 26 Oct 2026 10:31:31 +0800 Subject: [PATCH 1/3] fix: return a scalar field from a scalar struct in GetStructField A scalar struct input returned its field as a one-row array, which does not match the batch's row count beside a column. Return a scalar instead, wrapping a list field directly so its element field keeps its metadata. Assisted-by: Claude Code (claude-opus-5-5) --- .../src/struct_funcs/get_struct_field.rs | 116 +++++++++++++++++- .../expressions/struct/get_struct_field.sql | 12 ++ 2 files changed, 122 insertions(+), 6 deletions(-) diff --git a/native/spark-expr/src/struct_funcs/get_struct_field.rs b/native/spark-expr/src/struct_funcs/get_struct_field.rs index b815b4ed9a0..272313ec333 100644 --- a/native/spark-expr/src/struct_funcs/get_struct_field.rs +++ b/native/spark-expr/src/struct_funcs/get_struct_field.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use arrow::array::{Array, StructArray}; +use arrow::array::{Array, ArrayRef, AsArray, StructArray}; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use datafusion::common::{DataFusionError, Result as DataFusionResult, ScalarValue}; @@ -99,8 +99,10 @@ impl PhysicalExpr for GetStructField { self.ordinal, )?)) } - ColumnarValue::Scalar(ScalarValue::Struct(struct_array)) => Ok(ColumnarValue::Array( - child_with_parent_nulls(&struct_array, self.ordinal)?, + // A scalar struct holds one row, so its field is a scalar too. A one-row array would + // not match the batch's row count. + ColumnarValue::Scalar(ScalarValue::Struct(struct_array)) => Ok(ColumnarValue::Scalar( + single_row_scalar(child_with_parent_nulls(&struct_array, self.ordinal)?)?, )), value => Err(DataFusionError::Execution(format!( "Expected a struct array, got {value:?}" @@ -123,6 +125,20 @@ impl PhysicalExpr for GetStructField { } } +/// The scalar that a one-row array holds. A list wraps the array as is: `try_from_array` rebuilds +/// a list's element field from its name and nullability, which drops metadata such as a Variant +/// element's extension type, so the scalar's type would no longer match `data_type`. +fn single_row_scalar(array: ArrayRef) -> DataFusionResult { + Ok(match array.data_type() { + DataType::List(_) => ScalarValue::List(Arc::new(array.as_list::().clone())), + DataType::LargeList(_) => ScalarValue::LargeList(Arc::new(array.as_list::().clone())), + DataType::FixedSizeList(_, _) => { + ScalarValue::FixedSizeList(Arc::new(array.as_fixed_size_list().clone())) + } + _ => ScalarValue::try_from_array(&array, 0)?, + }) +} + impl Display for GetStructField { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!( @@ -136,10 +152,98 @@ impl Display for GetStructField { #[cfg(test)] mod tests { use super::*; - use arrow::array::{ArrayRef, Int64Array}; - use arrow::buffer::NullBuffer; + use arrow::array::{Int64Array, ListArray, NullArray}; + use arrow::buffer::{NullBuffer, OffsetBuffer}; use arrow::datatypes::Fields; - use datafusion::physical_expr::expressions::Column; + use datafusion::physical_expr::expressions::{Column, Literal}; + + // A field of a scalar struct is a scalar, so a consumer broadcasts it over the batch instead of + // reading a one-row array beside full columns. A null struct gives a null field, for a typed + // field and a Null-typed one. + #[test] + fn field_of_scalar_struct_is_scalar() { + let fields: Fields = Fields::from(vec![ + Field::new("a", DataType::Int64, true), + Field::new("n", DataType::Null, true), + ]); + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int64Array::from(vec![0_i64, 1, 2, 3]))], + ) + .unwrap(); + + let present = ScalarValue::Struct(Arc::new(StructArray::new( + fields.clone(), + vec![ + Arc::new(Int64Array::from(vec![7_i64])), + Arc::new(NullArray::new(1)), + ], + None, + ))); + let absent = ScalarValue::Struct(Arc::new(StructArray::new_null(fields.clone(), 1))); + // A null struct whose child slot still holds a value: the field is null all the same. + let hidden = ScalarValue::Struct(Arc::new(StructArray::new( + fields, + vec![ + Arc::new(Int64Array::from(vec![7_i64])), + Arc::new(NullArray::new(1)), + ], + Some(NullBuffer::new_null(1)), + ))); + + for (scalar, ordinal, expected) in [ + (present.clone(), 0, ScalarValue::Int64(Some(7))), + (present, 1, ScalarValue::Null), + (absent.clone(), 0, ScalarValue::Int64(None)), + (absent, 1, ScalarValue::Null), + (hidden, 0, ScalarValue::Int64(None)), + ] { + let expr = GetStructField::new(Arc::new(Literal::new(scalar)), ordinal); + match expr.evaluate(&batch).unwrap() { + ColumnarValue::Scalar(value) => assert_eq!(value, expected), + ColumnarValue::Array(array) => { + panic!("expected a scalar for ordinal {ordinal}, got array {array:?}") + } + } + } + } + + // A list field of a scalar struct keeps its element field, metadata included, so the scalar's + // type is the one `data_type` declares. + #[test] + fn list_field_of_scalar_struct_keeps_its_type() { + let item = + Arc::new(Field::new("item", DataType::Int64, true).with_metadata( + [("ARROW:extension:name".to_string(), "test.ext".to_string())].into(), + )); + let list = ListArray::new( + Arc::clone(&item), + OffsetBuffer::from_lengths([2]), + Arc::new(Int64Array::from(vec![1_i64, 2])), + None, + ); + let fields: Fields = Fields::from(vec![Field::new("l", DataType::List(item), true)]); + let scalar = ScalarValue::Struct(Arc::new(StructArray::new( + fields, + vec![Arc::new(list)], + None, + ))); + let schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![Arc::new(Int64Array::from(vec![0_i64, 1, 2]))], + ) + .unwrap(); + + let expr = GetStructField::new(Arc::new(Literal::new(scalar)), 0); + match expr.evaluate(&batch).unwrap() { + ColumnarValue::Scalar(value) => { + assert_eq!(value.data_type(), expr.data_type(&schema).unwrap()) + } + ColumnarValue::Array(array) => panic!("expected a scalar, got array {array:?}"), + } + } // A field of a NULL struct must be NULL (Spark semantics) even when the child buffer holds a // non-null value at that row -- Arrow stores child validity independently of the parent diff --git a/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql b/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql index 6bfb2875000..84acd379fe7 100644 --- a/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql +++ b/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql @@ -38,3 +38,15 @@ INSERT INTO test_struct_null_field VALUES (1), (-1), (NULL) query SELECT s.x, s.n FROM ( SELECT IF(a > 0, named_struct('x', a, 'n', null), null) AS s FROM test_struct_null_field) + +-- A field of a struct that native evaluates as a scalar (a cast of named_struct over a scalar +-- subquery) is one value for every row. The table is one file, so at the default batch size its +-- four rows share a batch. +statement +CREATE TABLE test_struct_scalar(a int) USING parquet + +statement +INSERT INTO test_struct_scalar SELECT /*+ REPARTITION(1) */ * FROM VALUES (1), (-1), (NULL), (4) AS v(a) + +query +SELECT a, CAST(named_struct('x', (SELECT max(a) FROM test_struct_scalar)) AS STRUCT).x, CAST(named_struct('x', (SELECT max(a) FROM test_struct_scalar WHERE a > 100)) AS STRUCT).x FROM test_struct_scalar From 57baca6647e88877dd74c5b02780ef435be234ea Mon Sep 17 00:00:00 2001 From: grorge Date: Mon, 2 Nov 2026 20:13:08 +0800 Subject: [PATCH 2/3] fix: keep the element field of a list view field from a scalar struct Assisted-by: Claude Code (claude-opus-5-5) --- .../src/struct_funcs/get_struct_field.rs | 59 +++++++++++++------ 1 file changed, 40 insertions(+), 19 deletions(-) diff --git a/native/spark-expr/src/struct_funcs/get_struct_field.rs b/native/spark-expr/src/struct_funcs/get_struct_field.rs index 272313ec333..311d891ce97 100644 --- a/native/spark-expr/src/struct_funcs/get_struct_field.rs +++ b/native/spark-expr/src/struct_funcs/get_struct_field.rs @@ -135,6 +135,12 @@ fn single_row_scalar(array: ArrayRef) -> DataFusionResult { DataType::FixedSizeList(_, _) => { ScalarValue::FixedSizeList(Arc::new(array.as_fixed_size_list().clone())) } + DataType::ListView(_) => { + ScalarValue::ListView(Arc::new(array.as_list_view::().clone())) + } + DataType::LargeListView(_) => { + ScalarValue::LargeListView(Arc::new(array.as_list_view::().clone())) + } _ => ScalarValue::try_from_array(&array, 0)?, }) } @@ -152,8 +158,8 @@ impl Display for GetStructField { #[cfg(test)] mod tests { use super::*; - use arrow::array::{Int64Array, ListArray, NullArray}; - use arrow::buffer::{NullBuffer, OffsetBuffer}; + use arrow::array::{Int64Array, LargeListViewArray, ListArray, ListViewArray, NullArray}; + use arrow::buffer::{NullBuffer, OffsetBuffer, ScalarBuffer}; use arrow::datatypes::Fields; use datafusion::physical_expr::expressions::{Column, Literal}; @@ -217,18 +223,29 @@ mod tests { Arc::new(Field::new("item", DataType::Int64, true).with_metadata( [("ARROW:extension:name".to_string(), "test.ext".to_string())].into(), )); - let list = ListArray::new( - Arc::clone(&item), - OffsetBuffer::from_lengths([2]), - Arc::new(Int64Array::from(vec![1_i64, 2])), - None, - ); - let fields: Fields = Fields::from(vec![Field::new("l", DataType::List(item), true)]); - let scalar = ScalarValue::Struct(Arc::new(StructArray::new( - fields, - vec![Arc::new(list)], - None, - ))); + let values: ArrayRef = Arc::new(Int64Array::from(vec![1_i64, 2])); + let lists: Vec = vec![ + Arc::new(ListArray::new( + Arc::clone(&item), + OffsetBuffer::from_lengths([2]), + Arc::clone(&values), + None, + )), + Arc::new(ListViewArray::new( + Arc::clone(&item), + ScalarBuffer::from(vec![0_i32]), + ScalarBuffer::from(vec![2_i32]), + Arc::clone(&values), + None, + )), + Arc::new(LargeListViewArray::new( + Arc::clone(&item), + ScalarBuffer::from(vec![0_i64]), + ScalarBuffer::from(vec![2_i64]), + Arc::clone(&values), + None, + )), + ]; let schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]); let batch = RecordBatch::try_new( Arc::new(schema.clone()), @@ -236,12 +253,16 @@ mod tests { ) .unwrap(); - let expr = GetStructField::new(Arc::new(Literal::new(scalar)), 0); - match expr.evaluate(&batch).unwrap() { - ColumnarValue::Scalar(value) => { - assert_eq!(value.data_type(), expr.data_type(&schema).unwrap()) + for list in lists { + let fields = Fields::from(vec![Field::new("l", list.data_type().clone(), true)]); + let scalar = ScalarValue::Struct(Arc::new(StructArray::new(fields, vec![list], None))); + let expr = GetStructField::new(Arc::new(Literal::new(scalar)), 0); + match expr.evaluate(&batch).unwrap() { + ColumnarValue::Scalar(value) => { + assert_eq!(value.data_type(), expr.data_type(&schema).unwrap()) + } + ColumnarValue::Array(array) => panic!("expected a scalar, got array {array:?}"), } - ColumnarValue::Array(array) => panic!("expected a scalar, got array {array:?}"), } } From 20470b9c921d7e1e14fd6705c6b4ab9bb369d320 Mon Sep 17 00:00:00 2001 From: grorge Date: Sat, 7 Nov 2026 15:14:54 +0800 Subject: [PATCH 3/3] fix: keep the list element field of a dictionary field from a scalar struct A dictionary field now converts the value its key points at through the same path as other fields, so a list value keeps its element field. The SQL tests now cover array, struct and map fields, and a field of a null scalar struct, and check that the field access runs natively. The Rust tests check the list values and null structs. Assisted-by: Claude Code (claude-opus-5-5) --- .../src/struct_funcs/get_struct_field.rs | 128 ++++++++++++++++-- .../expressions/struct/get_struct_field.sql | 27 +++- 2 files changed, 144 insertions(+), 11 deletions(-) diff --git a/native/spark-expr/src/struct_funcs/get_struct_field.rs b/native/spark-expr/src/struct_funcs/get_struct_field.rs index 311d891ce97..cdae3c94daa 100644 --- a/native/spark-expr/src/struct_funcs/get_struct_field.rs +++ b/native/spark-expr/src/struct_funcs/get_struct_field.rs @@ -127,7 +127,9 @@ impl PhysicalExpr for GetStructField { /// The scalar that a one-row array holds. A list wraps the array as is: `try_from_array` rebuilds /// a list's element field from its name and nullability, which drops metadata such as a Variant -/// element's extension type, so the scalar's type would no longer match `data_type`. +/// element's extension type, so the scalar's type would no longer match `data_type`. A dictionary +/// converts the value its key points at here for the same reason, since `try_from_array` would +/// convert it with itself. fn single_row_scalar(array: ArrayRef) -> DataFusionResult { Ok(match array.data_type() { DataType::List(_) => ScalarValue::List(Arc::new(array.as_list::().clone())), @@ -141,6 +143,15 @@ fn single_row_scalar(array: ArrayRef) -> DataFusionResult { DataType::LargeListView(_) => { ScalarValue::LargeListView(Arc::new(array.as_list_view::().clone())) } + DataType::Dictionary(key_type, _) => { + let dict = array.as_any_dictionary(); + let value = if dict.keys().is_valid(0) { + single_row_scalar(dict.values().slice(dict.normalized_keys()[0], 1))? + } else { + ScalarValue::try_from(dict.values().data_type())? + }; + ScalarValue::Dictionary(key_type.clone(), Box::new(value)) + } _ => ScalarValue::try_from_array(&array, 0)?, }) } @@ -158,7 +169,10 @@ impl Display for GetStructField { #[cfg(test)] mod tests { use super::*; - use arrow::array::{Int64Array, LargeListViewArray, ListArray, ListViewArray, NullArray}; + use arrow::array::{ + DictionaryArray, FixedSizeListArray, Int32Array, Int64Array, LargeListArray, + LargeListViewArray, ListArray, ListViewArray, NullArray, + }; use arrow::buffer::{NullBuffer, OffsetBuffer, ScalarBuffer}; use arrow::datatypes::Fields; use datafusion::physical_expr::expressions::{Column, Literal}; @@ -215,8 +229,9 @@ mod tests { } } - // A list field of a scalar struct keeps its element field, metadata included, so the scalar's - // type is the one `data_type` declares. + // A list field of a scalar struct is the struct's list, and a null struct gives a null list. + // Either way it keeps its element field, metadata included, so the scalar's type is the one + // `data_type` declares. #[test] fn list_field_of_scalar_struct_keeps_its_type() { let item = @@ -231,6 +246,18 @@ mod tests { Arc::clone(&values), None, )), + Arc::new(LargeListArray::new( + Arc::clone(&item), + OffsetBuffer::from_lengths([2]), + Arc::clone(&values), + None, + )), + Arc::new(FixedSizeListArray::new( + Arc::clone(&item), + 2, + Arc::clone(&values), + None, + )), Arc::new(ListViewArray::new( Arc::clone(&item), ScalarBuffer::from(vec![0_i32]), @@ -255,13 +282,96 @@ mod tests { for list in lists { let fields = Fields::from(vec![Field::new("l", list.data_type().clone(), true)]); - let scalar = ScalarValue::Struct(Arc::new(StructArray::new(fields, vec![list], None))); - let expr = GetStructField::new(Arc::new(Literal::new(scalar)), 0); - match expr.evaluate(&batch).unwrap() { - ColumnarValue::Scalar(value) => { - assert_eq!(value.data_type(), expr.data_type(&schema).unwrap()) + // The second struct is null although its child slot holds the list. + for struct_nulls in [None, Some(NullBuffer::new_null(1))] { + let is_null = struct_nulls.is_some(); + let scalar = ScalarValue::Struct(Arc::new(StructArray::new( + fields.clone(), + vec![Arc::clone(&list)], + struct_nulls, + ))); + let expr = GetStructField::new(Arc::new(Literal::new(scalar)), 0); + let value = match expr.evaluate(&batch).unwrap() { + ColumnarValue::Scalar(value) => value, + ColumnarValue::Array(array) => panic!("expected a scalar, got array {array:?}"), + }; + let kind = list.data_type(); + assert_eq!( + value.data_type(), + expr.data_type(&schema).unwrap(), + "{kind}" + ); + assert_eq!(value.is_null(), is_null, "{kind}, null struct: {is_null}"); + if !is_null { + assert_eq!(value.to_array().unwrap().as_ref(), list.as_ref(), "{kind}"); } + } + } + } + + // A dictionary field of a scalar struct is the value its key points at, still dictionary + // encoded with the same key type. A list value keeps its element field, so the field broadcasts + // into a column of the type `data_type` declares. + #[test] + fn dictionary_field_of_scalar_struct_keeps_its_type() { + let item = + Arc::new(Field::new("item", DataType::Int64, true).with_metadata( + [("ARROW:extension:name".to_string(), "test.ext".to_string())].into(), + )); + let lists = ListArray::new( + Arc::clone(&item), + OffsetBuffer::from_lengths([1, 2]), + Arc::new(Int64Array::from(vec![0_i64, 1, 2])), + None, + ); + let dict = |key: Option| -> ArrayRef { + Arc::new(DictionaryArray::new( + Int32Array::from(vec![key]), + Arc::new(lists.clone()), + )) + }; + let schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![Arc::new(Int64Array::from(vec![0_i64, 1, 2]))], + ) + .unwrap(); + + // The key points at the second list, a null key gives a null, and so does a null struct + // whose child slot still holds a key. + for (field, struct_nulls, expected) in [ + (dict(Some(1)), None, Some(lists.slice(1, 1))), + (dict(None), None, None), + (dict(Some(1)), Some(NullBuffer::new_null(1)), None), + ] { + let fields = Fields::from(vec![Field::new("d", field.data_type().clone(), true)]); + let scalar = ScalarValue::Struct(Arc::new(StructArray::new( + fields, + vec![field], + struct_nulls, + ))); + let expr = GetStructField::new(Arc::new(Literal::new(scalar)), 0); + let value = match expr.evaluate(&batch).unwrap() { + ColumnarValue::Scalar(value) => value, ColumnarValue::Array(array) => panic!("expected a scalar, got array {array:?}"), + }; + let declared = Arc::new(Schema::new(vec![Field::new( + "d", + expr.data_type(&schema).unwrap(), + true, + )])); + RecordBatch::try_new( + declared, + vec![value.to_array_of_size(batch.num_rows()).unwrap()], + ) + .unwrap(); + let ScalarValue::Dictionary(key_type, inner) = value else { + panic!("expected a dictionary scalar, got {value:?}") + }; + assert_eq!(*key_type, DataType::Int32); + match expected { + Some(list) => assert_eq!(*inner, ScalarValue::List(Arc::new(list))), + None => assert!(inner.is_null(), "expected a null, got {inner:?}"), } } } diff --git a/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql b/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql index 84acd379fe7..7f7c04d4e49 100644 --- a/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql +++ b/spark/src/test/resources/sql-tests/expressions/struct/get_struct_field.sql @@ -15,6 +15,11 @@ -- specific language governing permissions and limitations -- under the License. +-- PushFoldableIntoBranches is excluded for the null struct query at the end, which it would +-- rewrite so that native no longer reads a field of the struct. The other queries keep a native +-- GetStructField either way. +-- ExcludeRules: org.apache.spark.sql.catalyst.optimizer.PushFoldableIntoBranches + statement CREATE TABLE test_struct(s struct) USING parquet @@ -41,12 +46,30 @@ SELECT s.x, s.n FROM ( -- A field of a struct that native evaluates as a scalar (a cast of named_struct over a scalar -- subquery) is one value for every row. The table is one file, so at the default batch size its --- four rows share a batch. +-- four rows share a batch. expect_native checks that the field access itself reaches native, so +-- that an optimizer rewrite cannot turn these into tests of something else. statement CREATE TABLE test_struct_scalar(a int) USING parquet statement INSERT INTO test_struct_scalar SELECT /*+ REPARTITION(1) */ * FROM VALUES (1), (-1), (NULL), (4) AS v(a) -query +query expect_native(getstructfield) SELECT a, CAST(named_struct('x', (SELECT max(a) FROM test_struct_scalar)) AS STRUCT).x, CAST(named_struct('x', (SELECT max(a) FROM test_struct_scalar WHERE a > 100)) AS STRUCT).x FROM test_struct_scalar + +-- The same holds for an array, a struct and a map field. +query expect_native(getstructfield) +SELECT a, CAST(named_struct('l', array((SELECT max(a) FROM test_struct_scalar))) AS STRUCT>).l FROM test_struct_scalar + +query expect_native(getstructfield) +SELECT a, CAST(named_struct('s', named_struct('y', (SELECT max(a) FROM test_struct_scalar))) AS STRUCT>).s FROM test_struct_scalar + +-- The map comes from map_from_arrays, which runs natively; map() over a scalar subquery would run +-- in Spark. +query expect_native(getstructfield, map_from_arrays) +SELECT a, CAST(named_struct('m', map_from_arrays(array('k'), array((SELECT max(a) FROM test_struct_scalar)))) AS STRUCT>).m FROM test_struct_scalar + +-- A field of a null struct that native evaluates as a scalar (an IF over a scalar subquery whose +-- other branch is NULL) is NULL for every row, next to the same IF selecting the struct. +query expect_native(getstructfield) +SELECT a, IF((SELECT max(a) FROM test_struct_scalar) > 100, named_struct('x', (SELECT max(a) FROM test_struct_scalar)), NULL).x, IF((SELECT max(a) FROM test_struct_scalar) > 0, named_struct('x', (SELECT max(a) FROM test_struct_scalar)), NULL).x FROM test_struct_scalar