From 8460a80fb8bc288d52385fe9e8f577cdecc29be6 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sun, 10 May 2026 23:43:29 +0530 Subject: [PATCH 01/14] added NullHandling Enum matching Spark explode_outer --- datafusion/common/src/lib.rs | 2 +- datafusion/common/src/unnest.rs | 93 +++++++++-- datafusion/core/tests/dataframe/mod.rs | 24 +++ datafusion/physical-plan/src/unnest.rs | 158 ++++++++++++++++-- datafusion/proto/proto/datafusion.proto | 17 +- datafusion/proto/src/generated/pbjson.rs | 110 ++++++++++-- datafusion/proto/src/generated/prost.rs | 50 +++++- .../proto/src/logical_plan/from_proto.rs | 14 +- datafusion/proto/src/logical_plan/to_proto.rs | 11 +- datafusion/sql/src/unparser/plan.rs | 2 +- 10 files changed, 425 insertions(+), 56 deletions(-) diff --git a/datafusion/common/src/lib.rs b/datafusion/common/src/lib.rs index 9be0941b5d575..f735add7d53a9 100644 --- a/datafusion/common/src/lib.rs +++ b/datafusion/common/src/lib.rs @@ -95,7 +95,7 @@ pub use schema_reference::SchemaReference; pub use spans::{Location, Span, Spans}; pub use stats::{ColumnStatistics, Statistics}; pub use table_reference::{ResolvedTableReference, TableReference}; -pub use unnest::{RecursionUnnestOption, UnnestOptions}; +pub use unnest::{NullHandling, RecursionUnnestOption, UnnestOptions}; pub use utils::project_schema; // These are hidden from docs purely to avoid polluting the public view of what this crate exports. diff --git a/datafusion/common/src/unnest.rs b/datafusion/common/src/unnest.rs index db48edd061605..3306e79a6f691 100644 --- a/datafusion/common/src/unnest.rs +++ b/datafusion/common/src/unnest.rs @@ -19,23 +19,38 @@ use crate::Column; +/// How [`UnnestOptions`] handles `NULL` and empty list values in the input column. +/// +/// The variants enumerate the three observable behaviors so that callers do +/// not have to compose multiple boolean flags to express what they want. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Hash)] +pub enum NullHandling { + /// Drop rows where the input list is `NULL` or empty. Matches the + /// default behavior of systems such as DuckDB and ClickHouse. + Drop, + /// Preserve `NULL` input rows as a single output row containing `NULL`. + /// Empty lists still produce zero output rows. This is the default and + /// matches DataFusion's historical `preserve_nulls = true` behavior. + #[default] + Preserve, + /// Like [`Self::Preserve`], and additionally treat an empty list + /// identically to a `NULL` list, producing a single output row + /// containing `NULL`. Matches Spark's `explode_outer` semantics. + PreserveAndExpandEmpty, +} + /// Options for unnesting a column that contains a list type, /// replicating values in the other, non nested rows. /// /// Conceptually this operation is like joining each row with all the /// values in the list column. /// -/// If `preserve_nulls` is false, nulls and empty lists -/// from the input column are not carried through to the output. This -/// is the default behavior for other systems such as ClickHouse and -/// DuckDB -/// -/// If `preserve_nulls` is true (the default), nulls from the input -/// column are carried through to the output. +/// The behavior with `NULL` and empty input lists is controlled by +/// [`NullHandling`]. See its variants for full details. /// /// # Examples /// -/// ## `Unnest(c1)`, preserve_nulls: false +/// ## `Unnest(c1)`, null_handling: NullHandling::Drop /// ```text /// ┌─────────┐ ┌─────┐ ┌─────────┐ ┌─────┐ /// │ {1, 2} │ │ A │ Unnest │ 1 │ │ A │ @@ -49,7 +64,7 @@ use crate::Column; /// c1 c2 /// ``` /// -/// ## `Unnest(c1)`, preserve_nulls: true +/// ## `Unnest(c1)`, null_handling: NullHandling::Preserve /// ```text /// ┌─────────┐ ┌─────┐ ┌─────────┐ ┌─────┐ /// │ {1, 2} │ │ A │ Unnest │ 1 │ │ A │ @@ -63,13 +78,30 @@ use crate::Column; /// c1 c2 c1 c2 /// ``` /// +/// ## `Unnest(c1)`, null_handling: NullHandling::PreserveAndExpandEmpty +/// ```text +/// ┌─────────┐ ┌─────┐ ┌─────────┐ ┌─────┐ +/// │ {1, 2} │ │ A │ Unnest │ 1 │ │ A │ +/// ├─────────┤ ├─────┤ ├─────────┤ ├─────┤ +/// │ null │ │ B │ │ 2 │ │ A │ +/// ├─────────┤ ├─────┤ ────────────▶ ├─────────┤ ├─────┤ +/// │ {} │ │ D │ │ null │ │ B │ +/// ├─────────┤ ├─────┤ ├─────────┤ ├─────┤ +/// │ {3} │ │ E │ │ null │ │ D │ +/// └─────────┘ └─────┘ ├─────────┤ ├─────┤ +/// c1 c2 │ 3 │ │ E │ +/// └─────────┘ └─────┘ +/// c1 c2 +/// ``` +/// /// `recursions` instruct how a column should be unnested (e.g unnesting a column multiple /// time, with depth = 1 and depth = 2). Any unnested column not being mentioned inside this /// options is inferred to be unnested with depth = 1 #[derive(Debug, Clone, PartialEq, PartialOrd, Hash, Eq)] pub struct UnnestOptions { - /// Should nulls in the input be preserved? Defaults to true - pub preserve_nulls: bool, + /// How to handle `NULL` and empty list values in the input column. + /// Defaults to [`NullHandling::Preserve`]. + pub null_handling: NullHandling, /// If specific columns need to be unnested multiple times (e.g at different depth), /// declare them here. Any unnested columns not being mentioned inside this option /// will be unnested with depth = 1 @@ -88,8 +120,7 @@ pub struct RecursionUnnestOption { impl Default for UnnestOptions { fn default() -> Self { Self { - // default to true to maintain backwards compatible behavior - preserve_nulls: true, + null_handling: NullHandling::Preserve, recursions: vec![], } } @@ -101,13 +132,41 @@ impl UnnestOptions { Default::default() } - /// Set the behavior with nulls in the input as described on - /// [`Self`] - pub fn with_preserve_nulls(mut self, preserve_nulls: bool) -> Self { - self.preserve_nulls = preserve_nulls; + /// Set the [`NullHandling`] mode used when unnesting `NULL` or empty + /// input lists. + pub fn with_null_handling(mut self, null_handling: NullHandling) -> Self { + self.null_handling = null_handling; self } + /// Backward-compatible setter that maps the previous boolean + /// `preserve_nulls` flag onto [`NullHandling`]. + /// + /// `true` maps to [`NullHandling::Preserve`]; `false` maps to + /// [`NullHandling::Drop`]. To opt into Spark `explode_outer` + /// semantics, call [`Self::with_null_handling`] directly with + /// [`NullHandling::PreserveAndExpandEmpty`]. + pub fn with_preserve_nulls(self, preserve_nulls: bool) -> Self { + let null_handling = if preserve_nulls { + NullHandling::Preserve + } else { + NullHandling::Drop + }; + self.with_null_handling(null_handling) + } + + /// Returns true if `NULL` input rows produce a single output row + /// containing `NULL`. + pub fn preserve_nulls(&self) -> bool { + !matches!(self.null_handling, NullHandling::Drop) + } + + /// Returns true if empty input lists should produce a single + /// output row containing `NULL` (Spark `explode_outer` semantics). + pub fn expand_empty_as_null(&self) -> bool { + matches!(self.null_handling, NullHandling::PreserveAndExpandEmpty) + } + /// Set the recursions for the unnest operation pub fn with_recursions(mut self, recursion: RecursionUnnestOption) -> Self { self.recursions.push(recursion); diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs index 1418dcad1150e..6c1b7cd62b48b 100644 --- a/datafusion/core/tests/dataframe/mod.rs +++ b/datafusion/core/tests/dataframe/mod.rs @@ -4357,6 +4357,7 @@ async fn unnest_column_nulls() -> Result<()> { let options = UnnestOptions::new().with_preserve_nulls(false); let results = df + .clone() .unnest_columns_with_options(&["list"], options)? .collect() .await?; @@ -4373,6 +4374,29 @@ async fn unnest_column_nulls() -> Result<()> { " ); + // Spark `explode_outer` semantics: NULL and empty lists both produce a + // single output row containing NULL. + let options = UnnestOptions::new() + .with_null_handling(datafusion_common::NullHandling::PreserveAndExpandEmpty); + let results = df + .unnest_columns_with_options(&["list"], options)? + .collect() + .await?; + assert_snapshot!( + batches_to_string(&results), + @r" + +------+----+ + | list | id | + +------+----+ + | 1 | A | + | 2 | A | + | | B | + | | C | + | 3 | D | + +------+----+ + " + ); + Ok(()) } diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs index c774ff09af33c..6eeb4ea0fd617 100644 --- a/datafusion/physical-plan/src/unnest.rs +++ b/datafusion/physical-plan/src/unnest.rs @@ -759,14 +759,22 @@ fn build_batch( /// l2: [4,5], [], null, [6, 7] /// ``` /// -/// If `preserve_nulls` is false, the longest length array will be: +/// With [`NullHandling::Drop`], the longest length array will be: /// /// ```ignore /// longest_length: [3, 0, 0, 2] /// ``` /// -/// whereas if `preserve_nulls` is true, the longest length array will be: +/// With [`NullHandling::Preserve`] (the default), the longest length array +/// will be: /// +/// ```ignore +/// longest_length: [3, 1, 1, 2] +/// ``` +/// +/// With [`NullHandling::PreserveAndExpandEmpty`], empty input lists are +/// also bumped to length 1 so they produce a single `NULL` row, matching +/// Spark `explode_outer` semantics: /// /// ```ignore /// longest_length: [3, 1, 1, 2] @@ -775,12 +783,16 @@ fn find_longest_length( list_arrays: &[ArrayRef], options: &UnnestOptions, ) -> Result { - // The length of a NULL list - let null_length = if options.preserve_nulls { + // The length to substitute for a NULL input list. + let null_length = if options.preserve_nulls() { Scalar::new(Int64Array::from_value(1, 1)) } else { Scalar::new(Int64Array::from_value(0, 1)) }; + let expand_empty = options.expand_empty_as_null(); + // Reused scalars for the empty-list rewrite when expand_empty is set. + let zero = Scalar::new(Int64Array::from_value(0, 1)); + let one = Scalar::new(Int64Array::from_value(1, 1)); let list_lengths: Vec = list_arrays .iter() .map(|list_array| { @@ -789,6 +801,12 @@ fn find_longest_length( length_array = cast(&length_array, &DataType::Int64)?; length_array = zip(&is_not_null(&length_array)?, &length_array, &null_length)?; + if expand_empty { + // Bump empty lists (length 0) to length 1 so they + // produce a single output row padded with NULL. + let is_zero = arrow_ord::cmp::eq(&length_array, &zero)?; + length_array = zip(&is_zero, &one, &length_array)?; + } Ok(length_array) }) .collect::>()?; @@ -1058,6 +1076,7 @@ mod tests { }; use arrow::buffer::{NullBuffer, OffsetBuffer}; use arrow::datatypes::{Field, Int32Type}; + use datafusion_common::NullHandling; use datafusion_common::test_util::batches_to_string; use insta::assert_snapshot; @@ -1240,7 +1259,7 @@ mod tests { list_type_columns.as_ref(), &HashSet::default(), &UnnestOptions { - preserve_nulls: true, + null_handling: NullHandling::Preserve, recursions: vec![], }, )? @@ -1276,6 +1295,71 @@ mod tests { Ok(()) } + #[test] + fn test_build_batch_preserve_and_expand_empty() -> Result<()> { + // c1: [A, B, C], [], NULL, [D], NULL, [NULL, F] c2: 1, 2, 3, 4, 5, 6 + // Expected for `NullHandling::PreserveAndExpandEmpty` (Spark explode_outer): + // [A, B, C] -> three rows with c2 = 1, 1, 1 + // [] -> one row with c2 = 2 and unnested value NULL + // NULL -> one row with c2 = 3 and unnested value NULL + // [D] -> one row with c2 = 4 + // NULL -> one row with c2 = 5 and unnested value NULL + // [NULL, F] -> two rows with c2 = 6, 6 + let list_array = Arc::new(make_generic_array::()) as ArrayRef; + let other = + Arc::new(arrow::array::Int32Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef; + let in_schema = Arc::new(Schema::new(vec![ + Field::new( + "c1", + DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true))), + true, + ), + Field::new("c2", DataType::Int32, true), + ])); + let out_schema = Arc::new(Schema::new(vec![ + Field::new("c1_unnested", DataType::Utf8, true), + Field::new("c2", DataType::Int32, true), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&in_schema), + vec![Arc::clone(&list_array), Arc::clone(&other)], + )?; + let list_type_columns = vec![ListUnnest { + index_in_input_schema: 0, + depth: 1, + }]; + + let ret = build_batch( + &batch, + &out_schema, + &list_type_columns, + &HashSet::default(), + &UnnestOptions { + null_handling: NullHandling::PreserveAndExpandEmpty, + recursions: vec![], + }, + )? + .unwrap(); + + assert_snapshot!(batches_to_string(&[ret]), + @r" + +-------------+----+ + | c1_unnested | c2 | + +-------------+----+ + | A | 1 | + | B | 1 | + | C | 1 | + | | 2 | + | | 3 | + | D | 4 | + | | 5 | + | | 6 | + | F | 6 | + +-------------+----+ + "); + Ok(()) + } + #[test] fn test_unnest_list_array() -> Result<()> { // [A, B, C], [], NULL, [D], NULL, [NULL, F] @@ -1323,11 +1407,11 @@ mod tests { fn verify_longest_length( list_arrays: &[ArrayRef], - preserve_nulls: bool, + null_handling: NullHandling, expected: Vec, ) -> Result<()> { let options = UnnestOptions { - preserve_nulls, + null_handling, recursions: vec![], }; let longest_length = find_longest_length(list_arrays, &options)?; @@ -1347,20 +1431,55 @@ mod tests { // Test with single ListArray // [A, B, C], [], NULL, [D], NULL, [NULL, F] let list_array = Arc::new(make_generic_array::()) as ArrayRef; - verify_longest_length(&[Arc::clone(&list_array)], false, vec![3, 0, 0, 1, 0, 2])?; - verify_longest_length(&[Arc::clone(&list_array)], true, vec![3, 0, 1, 1, 1, 2])?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::Drop, + vec![3, 0, 0, 1, 0, 2], + )?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::Preserve, + vec![3, 0, 1, 1, 1, 2], + )?; + // PreserveAndExpandEmpty also treats empty lists as a NULL row. + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::PreserveAndExpandEmpty, + vec![3, 1, 1, 1, 1, 2], + )?; // Test with single LargeListArray // [A, B, C], [], NULL, [D], NULL, [NULL, F] let list_array = Arc::new(make_generic_array::()) as ArrayRef; - verify_longest_length(&[Arc::clone(&list_array)], false, vec![3, 0, 0, 1, 0, 2])?; - verify_longest_length(&[Arc::clone(&list_array)], true, vec![3, 0, 1, 1, 1, 2])?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::Drop, + vec![3, 0, 0, 1, 0, 2], + )?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::Preserve, + vec![3, 0, 1, 1, 1, 2], + )?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::PreserveAndExpandEmpty, + vec![3, 1, 1, 1, 1, 2], + )?; // Test with single FixedSizeListArray // [A, B], NULL, [C, D], NULL, [NULL, F], [NULL, NULL] let list_array = Arc::new(make_fixed_list()) as ArrayRef; - verify_longest_length(&[Arc::clone(&list_array)], false, vec![2, 0, 2, 0, 2, 2])?; - verify_longest_length(&[Arc::clone(&list_array)], true, vec![2, 1, 2, 1, 2, 2])?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::Drop, + vec![2, 0, 2, 0, 2, 2], + )?; + verify_longest_length( + &[Arc::clone(&list_array)], + NullHandling::Preserve, + vec![2, 1, 2, 1, 2, 2], + )?; // Test with multiple list arrays // [A, B, C], [], NULL, [D], NULL, [NULL, F] @@ -1368,8 +1487,17 @@ mod tests { let list1 = Arc::new(make_generic_array::()) as ArrayRef; let list2 = Arc::new(make_fixed_list()) as ArrayRef; let list_arrays = vec![Arc::clone(&list1), Arc::clone(&list2)]; - verify_longest_length(&list_arrays, false, vec![3, 0, 2, 1, 2, 2])?; - verify_longest_length(&list_arrays, true, vec![3, 1, 2, 1, 2, 2])?; + verify_longest_length(&list_arrays, NullHandling::Drop, vec![3, 0, 2, 1, 2, 2])?; + verify_longest_length( + &list_arrays, + NullHandling::Preserve, + vec![3, 1, 2, 1, 2, 2], + )?; + verify_longest_length( + &list_arrays, + NullHandling::PreserveAndExpandEmpty, + vec![3, 1, 2, 1, 2, 2], + )?; Ok(()) } diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto index 865887d41e111..d9265b9c25ea8 100644 --- a/datafusion/proto/proto/datafusion.proto +++ b/datafusion/proto/proto/datafusion.proto @@ -331,7 +331,22 @@ message ColumnUnnestListRecursion { } message UnnestOptions { - bool preserve_nulls = 1; + // Reserved for the historical `bool preserve_nulls = 1;` field. + // Use `null_handling` instead. + reserved 1; + reserved "preserve_nulls"; + + enum NullHandling { + // Preserve nulls; empty lists produce no rows. The historical default. + PRESERVE = 0; + // Drop both null and empty lists from the output. + DROP = 1; + // Preserve nulls, and additionally expand empty lists into a single + // NULL output row (Spark `explode_outer` semantics). + PRESERVE_AND_EXPAND_EMPTY = 2; + } + + NullHandling null_handling = 3; repeated RecursionUnnestOption recursions = 2; } diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs index b8639afd04a89..6f6685526582a 100644 --- a/datafusion/proto/src/generated/pbjson.rs +++ b/datafusion/proto/src/generated/pbjson.rs @@ -25005,19 +25005,21 @@ impl serde::Serialize for UnnestOptions { { use serde::ser::SerializeStruct; let mut len = 0; - if self.preserve_nulls { + if !self.recursions.is_empty() { len += 1; } - if !self.recursions.is_empty() { + if self.null_handling != 0 { len += 1; } let mut struct_ser = serializer.serialize_struct("datafusion.UnnestOptions", len)?; - if self.preserve_nulls { - struct_ser.serialize_field("preserveNulls", &self.preserve_nulls)?; - } if !self.recursions.is_empty() { struct_ser.serialize_field("recursions", &self.recursions)?; } + if self.null_handling != 0 { + let v = unnest_options::NullHandling::try_from(self.null_handling) + .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.null_handling)))?; + struct_ser.serialize_field("nullHandling", &v)?; + } struct_ser.end() } } @@ -25028,15 +25030,15 @@ impl<'de> serde::Deserialize<'de> for UnnestOptions { D: serde::Deserializer<'de>, { const FIELDS: &[&str] = &[ - "preserve_nulls", - "preserveNulls", "recursions", + "null_handling", + "nullHandling", ]; #[allow(clippy::enum_variant_names)] enum GeneratedField { - PreserveNulls, Recursions, + NullHandling, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -25058,8 +25060,8 @@ impl<'de> serde::Deserialize<'de> for UnnestOptions { E: serde::de::Error, { match value { - "preserveNulls" | "preserve_nulls" => Ok(GeneratedField::PreserveNulls), "recursions" => Ok(GeneratedField::Recursions), + "nullHandling" | "null_handling" => Ok(GeneratedField::NullHandling), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -25079,33 +25081,107 @@ impl<'de> serde::Deserialize<'de> for UnnestOptions { where V: serde::de::MapAccess<'de>, { - let mut preserve_nulls__ = None; let mut recursions__ = None; + let mut null_handling__ = None; while let Some(k) = map_.next_key()? { match k { - GeneratedField::PreserveNulls => { - if preserve_nulls__.is_some() { - return Err(serde::de::Error::duplicate_field("preserveNulls")); - } - preserve_nulls__ = Some(map_.next_value()?); - } GeneratedField::Recursions => { if recursions__.is_some() { return Err(serde::de::Error::duplicate_field("recursions")); } recursions__ = Some(map_.next_value()?); } + GeneratedField::NullHandling => { + if null_handling__.is_some() { + return Err(serde::de::Error::duplicate_field("nullHandling")); + } + null_handling__ = Some(map_.next_value::()? as i32); + } } } Ok(UnnestOptions { - preserve_nulls: preserve_nulls__.unwrap_or_default(), recursions: recursions__.unwrap_or_default(), + null_handling: null_handling__.unwrap_or_default(), }) } } deserializer.deserialize_struct("datafusion.UnnestOptions", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for unnest_options::NullHandling { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + let variant = match self { + Self::Preserve => "PRESERVE", + Self::Drop => "DROP", + Self::PreserveAndExpandEmpty => "PRESERVE_AND_EXPAND_EMPTY", + }; + serializer.serialize_str(variant) + } +} +impl<'de> serde::Deserialize<'de> for unnest_options::NullHandling { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + "PRESERVE", + "DROP", + "PRESERVE_AND_EXPAND_EMPTY", + ]; + + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = unnest_options::NullHandling; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + fn visit_i64(self, v: i64) -> std::result::Result + where + E: serde::de::Error, + { + i32::try_from(v) + .ok() + .and_then(|x| x.try_into().ok()) + .ok_or_else(|| { + serde::de::Error::invalid_value(serde::de::Unexpected::Signed(v), &self) + }) + } + + fn visit_u64(self, v: u64) -> std::result::Result + where + E: serde::de::Error, + { + i32::try_from(v) + .ok() + .and_then(|x| x.try_into().ok()) + .ok_or_else(|| { + serde::de::Error::invalid_value(serde::de::Unexpected::Unsigned(v), &self) + }) + } + + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + match value { + "PRESERVE" => Ok(unnest_options::NullHandling::Preserve), + "DROP" => Ok(unnest_options::NullHandling::Drop), + "PRESERVE_AND_EXPAND_EMPTY" => Ok(unnest_options::NullHandling::PreserveAndExpandEmpty), + _ => Err(serde::de::Error::unknown_variant(value, FIELDS)), + } + } + } + deserializer.deserialize_any(GeneratedVisitor) + } +} impl serde::Serialize for ValuesNode { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs index b742e82ea24ec..16056a709e17b 100644 --- a/datafusion/proto/src/generated/prost.rs +++ b/datafusion/proto/src/generated/prost.rs @@ -528,10 +528,56 @@ pub struct ColumnUnnestListRecursion { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct UnnestOptions { - #[prost(bool, tag = "1")] - pub preserve_nulls: bool, #[prost(message, repeated, tag = "2")] pub recursions: ::prost::alloc::vec::Vec, + #[prost(enumeration = "unnest_options::NullHandling", tag = "3")] + pub null_handling: i32, +} +/// Nested message and enum types in `UnnestOptions`. +pub mod unnest_options { + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] + #[repr(i32)] + pub enum NullHandling { + /// Preserve nulls; empty lists produce no rows. The historical default. + Preserve = 0, + /// Drop both null and empty lists from the output. + Drop = 1, + /// Preserve nulls, and additionally expand empty lists into a single + /// NULL output row (Spark `explode_outer` semantics). + PreserveAndExpandEmpty = 2, + } + impl NullHandling { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Preserve => "PRESERVE", + Self::Drop => "DROP", + Self::PreserveAndExpandEmpty => "PRESERVE_AND_EXPAND_EMPTY", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "PRESERVE" => Some(Self::Preserve), + "DROP" => Some(Self::Drop), + "PRESERVE_AND_EXPAND_EMPTY" => Some(Self::PreserveAndExpandEmpty), + _ => None, + } + } + } } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct RecursionUnnestOption { diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs index 78ffd362c8e48..2c7ecd6da27db 100644 --- a/datafusion/proto/src/logical_plan/from_proto.rs +++ b/datafusion/proto/src/logical_plan/from_proto.rs @@ -58,8 +58,20 @@ use super::{AsLogicalPlan, LogicalExtensionCodec}; impl From<&protobuf::UnnestOptions> for UnnestOptions { fn from(opts: &protobuf::UnnestOptions) -> Self { + use datafusion_common::NullHandling; + use protobuf::unnest_options::NullHandling as ProtoNullHandling; + let null_handling = match ProtoNullHandling::try_from(opts.null_handling) { + Ok(ProtoNullHandling::Preserve) => NullHandling::Preserve, + Ok(ProtoNullHandling::Drop) => NullHandling::Drop, + Ok(ProtoNullHandling::PreserveAndExpandEmpty) => { + NullHandling::PreserveAndExpandEmpty + } + // Unknown enum values fall back to the default (Preserve), which + // matches DataFusion's historical behavior. + Err(_) => NullHandling::Preserve, + }; Self { - preserve_nulls: opts.preserve_nulls, + null_handling, recursions: opts .recursions .iter() diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs index d79107d1d0f2b..d2be1db26761d 100644 --- a/datafusion/proto/src/logical_plan/to_proto.rs +++ b/datafusion/proto/src/logical_plan/to_proto.rs @@ -54,8 +54,17 @@ use crate::protobuf::LogicalPlanNode; impl From<&UnnestOptions> for protobuf::UnnestOptions { fn from(opts: &UnnestOptions) -> Self { + use datafusion_common::NullHandling; + use protobuf::unnest_options::NullHandling as ProtoNullHandling; + let null_handling = match opts.null_handling { + NullHandling::Preserve => ProtoNullHandling::Preserve, + NullHandling::Drop => ProtoNullHandling::Drop, + NullHandling::PreserveAndExpandEmpty => { + ProtoNullHandling::PreserveAndExpandEmpty + } + } as i32; Self { - preserve_nulls: opts.preserve_nulls, + null_handling, recursions: opts .recursions .iter() diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs index 2c36fe0b2c98a..a700837ae51a7 100644 --- a/datafusion/sql/src/unparser/plan.rs +++ b/datafusion/sql/src/unparser/plan.rs @@ -1530,7 +1530,7 @@ impl Unparser<'_> { let mut flatten = FlattenRelationBuilder::default(); flatten.input_expr(input_expr); - flatten.outer(unnest.options.preserve_nulls); + flatten.outer(unnest.options.preserve_nulls()); Ok(Some(flatten)) } From 5c160171335c11a909e2c8b9858bc0cd27f9e0e2 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Mon, 11 May 2026 00:04:20 +0530 Subject: [PATCH 02/14] proto fix --- datafusion/proto/src/generated/pbjson.rs | 32 ++++++++++++------------ datafusion/proto/src/generated/prost.rs | 4 +-- 2 files changed, 18 insertions(+), 18 deletions(-) diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs index 6f6685526582a..1676725f52593 100644 --- a/datafusion/proto/src/generated/pbjson.rs +++ b/datafusion/proto/src/generated/pbjson.rs @@ -25005,21 +25005,21 @@ impl serde::Serialize for UnnestOptions { { use serde::ser::SerializeStruct; let mut len = 0; - if !self.recursions.is_empty() { - len += 1; - } if self.null_handling != 0 { len += 1; } - let mut struct_ser = serializer.serialize_struct("datafusion.UnnestOptions", len)?; if !self.recursions.is_empty() { - struct_ser.serialize_field("recursions", &self.recursions)?; + len += 1; } + let mut struct_ser = serializer.serialize_struct("datafusion.UnnestOptions", len)?; if self.null_handling != 0 { let v = unnest_options::NullHandling::try_from(self.null_handling) .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.null_handling)))?; struct_ser.serialize_field("nullHandling", &v)?; } + if !self.recursions.is_empty() { + struct_ser.serialize_field("recursions", &self.recursions)?; + } struct_ser.end() } } @@ -25030,15 +25030,15 @@ impl<'de> serde::Deserialize<'de> for UnnestOptions { D: serde::Deserializer<'de>, { const FIELDS: &[&str] = &[ - "recursions", "null_handling", "nullHandling", + "recursions", ]; #[allow(clippy::enum_variant_names)] enum GeneratedField { - Recursions, NullHandling, + Recursions, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -25060,8 +25060,8 @@ impl<'de> serde::Deserialize<'de> for UnnestOptions { E: serde::de::Error, { match value { - "recursions" => Ok(GeneratedField::Recursions), "nullHandling" | "null_handling" => Ok(GeneratedField::NullHandling), + "recursions" => Ok(GeneratedField::Recursions), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -25081,27 +25081,27 @@ impl<'de> serde::Deserialize<'de> for UnnestOptions { where V: serde::de::MapAccess<'de>, { - let mut recursions__ = None; let mut null_handling__ = None; + let mut recursions__ = None; while let Some(k) = map_.next_key()? { match k { - GeneratedField::Recursions => { - if recursions__.is_some() { - return Err(serde::de::Error::duplicate_field("recursions")); - } - recursions__ = Some(map_.next_value()?); - } GeneratedField::NullHandling => { if null_handling__.is_some() { return Err(serde::de::Error::duplicate_field("nullHandling")); } null_handling__ = Some(map_.next_value::()? as i32); } + GeneratedField::Recursions => { + if recursions__.is_some() { + return Err(serde::de::Error::duplicate_field("recursions")); + } + recursions__ = Some(map_.next_value()?); + } } } Ok(UnnestOptions { - recursions: recursions__.unwrap_or_default(), null_handling: null_handling__.unwrap_or_default(), + recursions: recursions__.unwrap_or_default(), }) } } diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs index 16056a709e17b..ea235f39a55bc 100644 --- a/datafusion/proto/src/generated/prost.rs +++ b/datafusion/proto/src/generated/prost.rs @@ -528,10 +528,10 @@ pub struct ColumnUnnestListRecursion { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct UnnestOptions { - #[prost(message, repeated, tag = "2")] - pub recursions: ::prost::alloc::vec::Vec, #[prost(enumeration = "unnest_options::NullHandling", tag = "3")] pub null_handling: i32, + #[prost(message, repeated, tag = "2")] + pub recursions: ::prost::alloc::vec::Vec, } /// Nested message and enum types in `UnnestOptions`. pub mod unnest_options { From 6783655c21421d9f5bb4f46706aa658e6b9d703a Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Mon, 11 May 2026 17:43:38 +0530 Subject: [PATCH 03/14] fixed rustdoc references --- datafusion/physical-plan/src/unnest.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs index 6eeb4ea0fd617..4ba7e9d1946f7 100644 --- a/datafusion/physical-plan/src/unnest.rs +++ b/datafusion/physical-plan/src/unnest.rs @@ -759,20 +759,20 @@ fn build_batch( /// l2: [4,5], [], null, [6, 7] /// ``` /// -/// With [`NullHandling::Drop`], the longest length array will be: +/// With [`datafusion_common::NullHandling::Drop`], the longest length array will be: /// /// ```ignore /// longest_length: [3, 0, 0, 2] /// ``` /// -/// With [`NullHandling::Preserve`] (the default), the longest length array +/// With [`datafusion_common::NullHandling::Preserve`] (the default), the longest length array /// will be: /// /// ```ignore /// longest_length: [3, 1, 1, 2] /// ``` /// -/// With [`NullHandling::PreserveAndExpandEmpty`], empty input lists are +/// With [`datafusion_common::NullHandling::PreserveAndExpandEmpty`], empty input lists are /// also bumped to length 1 so they produce a single `NULL` row, matching /// Spark `explode_outer` semantics: /// From 745e6418a29e8fca90939762feedc1a220c724f5 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Wed, 13 May 2026 23:04:41 +0530 Subject: [PATCH 04/14] added test cases --- datafusion/core/tests/dataframe/mod.rs | 263 ++++++++++++++++++++++ datafusion/physical-plan/src/unnest.rs | 298 +++++++++++++++++++++++++ 2 files changed, 561 insertions(+) diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs index 6c1b7cd62b48b..2e79773bc4854 100644 --- a/datafusion/core/tests/dataframe/mod.rs +++ b/datafusion/core/tests/dataframe/mod.rs @@ -4400,6 +4400,269 @@ async fn unnest_column_nulls() -> Result<()> { Ok(()) } +/// Spark `explode_outer` semantics on a string list. Each input row is one +/// of: a list with values, a list with values *including a NULL element*, +/// an empty list, or a NULL list. The new `PreserveAndExpandEmpty` mode +/// must preserve inner NULL elements (they are different from "empty"), +/// preserve NULL input rows as a single NULL output row, *and* also turn +/// each empty list into a single NULL output row. +#[tokio::test] +async fn unnest_explode_outer_string_list_with_inner_nulls() -> Result<()> { + let mut list_builder = ListBuilder::new(StringBuilder::new()); + let mut id_builder = StringBuilder::new(); + + // ['x', 'y'], id = A + list_builder.values().append_value("x"); + list_builder.values().append_value("y"); + list_builder.append(true); + id_builder.append_value("A"); + + // ['p', NULL, 'q'], id = B (inner NULL must survive) + list_builder.values().append_value("p"); + list_builder.values().append_null(); + list_builder.values().append_value("q"); + list_builder.append(true); + id_builder.append_value("B"); + + // [], id = C (empty list) + list_builder.append(true); + id_builder.append_value("C"); + + // NULL, id = D (null list) + list_builder.append(false); + id_builder.append_value("D"); + + let batch = RecordBatch::try_from_iter(vec![ + ("tag", Arc::new(list_builder.finish()) as ArrayRef), + ("id", Arc::new(id_builder.finish()) as ArrayRef), + ])?; + + let ctx = SessionContext::new(); + ctx.register_batch("tags", batch)?; + let df = ctx.table("tags").await?; + + let options = UnnestOptions::new() + .with_null_handling(datafusion_common::NullHandling::PreserveAndExpandEmpty); + let results = df + .unnest_columns_with_options(&["tag"], options)? + .collect() + .await?; + assert_snapshot!( + batches_to_string(&results), + @r" + +-----+----+ + | tag | id | + +-----+----+ + | x | A | + | y | A | + | p | B | + | | B | + | q | B | + | | C | + | | D | + +-----+----+ + " + ); + + Ok(()) +} + +/// Spark `explode_outer` on a list-of-struct column. Verifies that +/// (a) struct elements unnest into flattened sub-columns and +/// (b) NULL and empty lists both still produce a single output row whose +/// struct sub-columns are all NULL. +#[tokio::test] +async fn unnest_explode_outer_list_of_struct() -> Result<()> { + use arrow::array::{Int32Array, StructArray}; + + // Per-row sub-list lengths: 2, 1, 0 (empty), 0 (null) + let names = StringArray::from(vec!["alice", "bob", "carol"]); + let ages = Int32Array::from(vec![30, 40, 50]); + let struct_values = StructArray::from(vec![ + ( + Arc::new(Field::new("name", DataType::Utf8, true)), + Arc::new(names) as ArrayRef, + ), + ( + Arc::new(Field::new("age", DataType::Int32, true)), + Arc::new(ages) as ArrayRef, + ), + ]); + let struct_field = + Arc::new(Field::new("item", struct_values.data_type().clone(), true)); + let offsets = arrow::buffer::OffsetBuffer::::from_lengths([2, 1, 0, 0]); + let validity = arrow::buffer::NullBuffer::from(vec![true, true, true, false]); + let people = ListArray::new( + struct_field, + offsets, + Arc::new(struct_values), + Some(validity), + ); + let group = Int32Array::from(vec![1, 2, 3, 4]); + + let batch = RecordBatch::try_from_iter(vec![ + ("people", Arc::new(people) as ArrayRef), + ("group", Arc::new(group) as ArrayRef), + ])?; + + let ctx = SessionContext::new(); + ctx.register_batch("teams", batch)?; + let df = ctx.table("teams").await?; + + let options = UnnestOptions::new() + .with_null_handling(datafusion_common::NullHandling::PreserveAndExpandEmpty); + let results = df + // Unnest the list, then expand the resulting struct rows into columns. + .unnest_columns_with_options(&["people"], options.clone())? + .unnest_columns_with_options(&["people"], options)? + .collect() + .await?; + assert_snapshot!( + batches_to_string(&results), + @r" + +-------------+------------+-------+ + | people.name | people.age | group | + +-------------+------------+-------+ + | alice | 30 | 1 | + | bob | 40 | 1 | + | carol | 50 | 2 | + | | | 3 | + | | | 4 | + +-------------+------------+-------+ + " + ); + + Ok(()) +} + +/// Spark `explode_outer` semantics applied to a `FixedSizeList` column. +/// For fixed-size lists, every non-null row has the fixed length, so +/// "empty" never occurs — `PreserveAndExpandEmpty` should behave identically +/// to `Preserve` here. The test pins that equivalence so we notice if it +/// ever diverges. +#[tokio::test] +async fn unnest_explode_outer_fixed_size_list() -> Result<()> { + let batch = get_fixed_list_batch()?; + let ctx = SessionContext::new(); + ctx.register_batch("shapes", batch)?; + let df = ctx.table("shapes").await?; + + let preserve_results = df + .clone() + .unnest_columns_with_options( + &["tags"], + UnnestOptions::new().with_preserve_nulls(true), + )? + .collect() + .await?; + let explode_outer_results = df + .unnest_columns_with_options( + &["tags"], + UnnestOptions::new().with_null_handling( + datafusion_common::NullHandling::PreserveAndExpandEmpty, + ), + )? + .collect() + .await?; + assert_eq!( + batches_to_sort_string(&preserve_results), + batches_to_sort_string(&explode_outer_results), + "FixedSizeList has no empty case, so PreserveAndExpandEmpty must \ + match Preserve exactly" + ); + + // And the snapshot itself, to make the expected shape explicit. + assert_snapshot!( + batches_to_sort_string(&explode_outer_results), + @r" + +----------+-------+ + | shape_id | tags | + +----------+-------+ + | 1 | | + | 2 | tag21 | + | 2 | tag22 | + | 3 | tag31 | + | 3 | tag32 | + | 4 | | + | 5 | tag51 | + | 5 | tag52 | + | 6 | tag61 | + | 6 | tag62 | + +----------+-------+ + " + ); + + Ok(()) +} + +/// Chain `explode` (drop nulls + empties) followed by `explode_outer` +/// (preserve nulls + empties) on a `list>` input, exercising the +/// "mixed modes in one query plan" path: each `UnnestExec` carries its own +/// `UnnestOptions`. The outer NULL row is dropped by the first stage, then +/// any NULL inner sub-list produced by the first stage is *preserved* by +/// the second stage as a NULL output row. +#[tokio::test] +async fn unnest_explode_then_explode_outer() -> Result<()> { + // outer: [[1,2,3], null, [4,5]], [[7], []], null + let inner = ListArray::from_iter_primitive::(vec![ + Some(vec![Some(1), Some(2), Some(3)]), + None, + Some(vec![Some(4), Some(5)]), + Some(vec![Some(7)]), + Some(vec![]), + ]); + let inner_ref = Arc::new(inner) as ArrayRef; + let offsets = arrow::buffer::OffsetBuffer::::from_lengths([3, 2, 0]); + let validity = arrow::buffer::NullBuffer::from(vec![true, true, false]); + let outer_field = Arc::new(Field::new("item", inner_ref.data_type().clone(), true)); + let outer = ListArray::new(outer_field, offsets, inner_ref, Some(validity)); + let id = Int32Array::from(vec![100, 200, 300]); + + let batch = RecordBatch::try_from_iter(vec![ + ("col", Arc::new(outer) as ArrayRef), + ("id", Arc::new(id) as ArrayRef), + ])?; + let ctx = SessionContext::new(); + ctx.register_batch("t", batch)?; + let df = ctx.table("t").await?; + + // Stage 1: explode — outer NULL row is dropped, empty inner row also dropped. + // Stage 2: explode_outer — any NULL produced by stage 1 (the null sub-list + // from row 0) is kept as a NULL output row. + let results = df + .unnest_columns_with_options( + &["col"], + UnnestOptions::new().with_preserve_nulls(false), + )? + .unnest_columns_with_options( + &["col"], + UnnestOptions::new().with_null_handling( + datafusion_common::NullHandling::PreserveAndExpandEmpty, + ), + )? + .collect() + .await?; + assert_snapshot!( + batches_to_string(&results), + @r" + +-----+-----+ + | col | id | + +-----+-----+ + | 1 | 100 | + | 2 | 100 | + | 3 | 100 | + | | 100 | + | 4 | 100 | + | 5 | 100 | + | 7 | 200 | + | | 200 | + +-----+-----+ + " + ); + + Ok(()) +} + #[tokio::test] async fn unnest_fixed_list() -> Result<()> { let batch = get_fixed_list_batch()?; diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs index 4ba7e9d1946f7..ee928e5e14978 100644 --- a/datafusion/physical-plan/src/unnest.rs +++ b/datafusion/physical-plan/src/unnest.rs @@ -1360,6 +1360,304 @@ mod tests { Ok(()) } + // PreserveAndExpandEmpty must work for LargeListArray (i64 offsets) too, + // not just the i32-offset ListArray exercised above. + #[test] + fn test_build_batch_preserve_and_expand_empty_largelist() -> Result<()> { + let list_array = Arc::new(make_generic_array::()) as ArrayRef; + let other = + Arc::new(arrow::array::Int32Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef; + let in_schema = Arc::new(Schema::new(vec![ + Field::new( + "c1", + DataType::LargeList(Arc::new(Field::new_list_field( + DataType::Utf8, + true, + ))), + true, + ), + Field::new("c2", DataType::Int32, true), + ])); + let out_schema = Arc::new(Schema::new(vec![ + Field::new("c1_unnested", DataType::Utf8, true), + Field::new("c2", DataType::Int32, true), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&in_schema), + vec![Arc::clone(&list_array), Arc::clone(&other)], + )?; + let list_type_columns = vec![ListUnnest { + index_in_input_schema: 0, + depth: 1, + }]; + + let ret = build_batch( + &batch, + &out_schema, + &list_type_columns, + &HashSet::default(), + &UnnestOptions { + null_handling: NullHandling::PreserveAndExpandEmpty, + recursions: vec![], + }, + )? + .unwrap(); + + // Same expected shape as the ListArray case — exercises the LargeList + // code path in unnest_list_array. + assert_snapshot!(batches_to_string(&[ret]), + @r" + +-------------+----+ + | c1_unnested | c2 | + +-------------+----+ + | A | 1 | + | B | 1 | + | C | 1 | + | | 2 | + | | 3 | + | D | 4 | + | | 5 | + | | 6 | + | F | 6 | + +-------------+----+ + "); + Ok(()) + } + + // When two list columns are unnested together, `find_longest_length` + // takes the per-row max. PreserveAndExpandEmpty must bump zeros to ones + // in each input column independently, then the row-wise max picks up + // the right value. + #[test] + fn test_build_batch_preserve_and_expand_empty_multi_column() -> Result<()> { + // col_a: [1, 2], [], NULL, [3] + // col_b: ['x'], ['y'],['z'], NULL + let col_a = ListArray::from_iter_primitive::(vec![ + Some(vec![Some(1), Some(2)]), + Some(vec![]), + None, + Some(vec![Some(3)]), + ]); + let col_b = { + let mut b = + arrow::array::ListBuilder::new(arrow::array::StringBuilder::new()); + b.values().append_value("x"); + b.append(true); + b.values().append_value("y"); + b.append(true); + b.values().append_value("z"); + b.append(true); + b.append(false); + b.finish() + }; + let id = + Arc::new(arrow::array::Int32Array::from(vec![10, 20, 30, 40])) as ArrayRef; + + let in_schema = Arc::new(Schema::new(vec![ + Field::new( + "a", + DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))), + true, + ), + Field::new( + "b", + DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true))), + true, + ), + Field::new("id", DataType::Int32, true), + ])); + let out_schema = Arc::new(Schema::new(vec![ + Field::new("a_unnested", DataType::Int32, true), + Field::new("b_unnested", DataType::Utf8, true), + Field::new("id", DataType::Int32, true), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&in_schema), + vec![ + Arc::new(col_a) as ArrayRef, + Arc::new(col_b) as ArrayRef, + Arc::clone(&id), + ], + )?; + let list_type_columns = vec![ + ListUnnest { + index_in_input_schema: 0, + depth: 1, + }, + ListUnnest { + index_in_input_schema: 1, + depth: 1, + }, + ]; + + let ret = build_batch( + &batch, + &out_schema, + &list_type_columns, + &HashSet::default(), + &UnnestOptions { + null_handling: NullHandling::PreserveAndExpandEmpty, + recursions: vec![], + }, + )? + .unwrap(); + + // Row 0: longest = max(len([1,2])=2, len(['x'])=1) = 2 → a=[1,2], b=['x',NULL] + // Row 1: a=[] bumped to len 1, b=['y'] len 1 → a=[NULL], b=['y'] + // Row 2: a=NULL bumped to len 1, b=['z'] len 1 → a=[NULL], b=['z'] + // Row 3: a=[3] len 1, b=NULL bumped to len 1 → a=[3], b=[NULL] + assert_snapshot!(batches_to_string(&[ret]), + @r" + +------------+------------+----+ + | a_unnested | b_unnested | id | + +------------+------------+----+ + | 1 | x | 10 | + | 2 | | 10 | + | | y | 20 | + | | z | 30 | + | 3 | | 40 | + +------------+------------+----+ + "); + Ok(()) + } + + // PreserveAndExpandEmpty must propagate through recursive depth-2 + // unnesting: an outer NULL or empty produces one NULL output row at + // each level. Adapted from `test_build_batch_list_arr_recursive`. + #[test] + fn test_build_batch_preserve_and_expand_empty_recursive() -> Result<()> { + // col1 | col2 + // [[1,2,3],null,[4,5]] | ['a','b'] + // [[7,8,9,10], null, [11,12,13]] | ['c','d'] + // null | ['e'] + let list_arr1 = ListArray::from_iter_primitive::(vec![ + Some(vec![Some(1), Some(2), Some(3)]), + None, + Some(vec![Some(4), Some(5)]), + Some(vec![Some(7), Some(8), Some(9), Some(10)]), + None, + Some(vec![Some(11), Some(12), Some(13)]), + ]); + let list_arr1_ref = Arc::new(list_arr1) as ArrayRef; + let offsets = OffsetBuffer::from_lengths([3, 3, 0]); + let mut nulls = NullBufferBuilder::new(3); + nulls.append_non_null(); + nulls.append_non_null(); + nulls.append_null(); + let col1_field = Field::new_list_field( + DataType::List(Arc::new(Field::new_list_field( + list_arr1_ref.data_type().to_owned(), + true, + ))), + true, + ); + let col1 = ListArray::new( + Arc::new(Field::new_list_field( + list_arr1_ref.data_type().to_owned(), + true, + )), + offsets, + list_arr1_ref, + nulls.finish(), + ); + + let list_arr2 = StringArray::from(vec![ + Some("a"), + Some("b"), + Some("c"), + Some("d"), + Some("e"), + ]); + let offsets = OffsetBuffer::from_lengths([2, 2, 1]); + let mut nulls = NullBufferBuilder::new(3); + nulls.append_n_non_nulls(3); + let col2_field = Field::new( + "col2", + DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true))), + true, + ); + let col2 = GenericListArray::::new( + Arc::new(Field::new_list_field(DataType::Utf8, true)), + OffsetBuffer::new(offsets.into()), + Arc::new(list_arr2), + nulls.finish(), + ); + let schema = Arc::new(Schema::new(vec![col1_field, col2_field])); + let out_schema = Arc::new(Schema::new(vec![ + Field::new( + "col1_unnest_placeholder_depth_1", + DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))), + true, + ), + Field::new("col1_unnest_placeholder_depth_2", DataType::Int32, true), + Field::new("col2_unnest_placeholder_depth_1", DataType::Utf8, true), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(col1) as ArrayRef, Arc::new(col2) as ArrayRef], + )?; + let list_type_columns = vec![ + ListUnnest { + index_in_input_schema: 0, + depth: 1, + }, + ListUnnest { + index_in_input_schema: 0, + depth: 2, + }, + ListUnnest { + index_in_input_schema: 1, + depth: 1, + }, + ]; + + let ret = build_batch( + &batch, + &out_schema, + &list_type_columns, + &HashSet::default(), + &UnnestOptions { + null_handling: NullHandling::PreserveAndExpandEmpty, + recursions: vec![], + }, + )? + .unwrap(); + + // The third input row (col1 = null, col2 = ['e']) now produces a + // NULL row for the depth-1 col1 placeholder *and* the depth-2 one, + // instead of being dropped at depth 1 and again at depth 2 the way + // it would be under `Drop`. Inner NULLs inside [...null...] sub- + // lists are still padded with NULL as before. + assert_snapshot!(batches_to_string(&[ret]), + @r" + +---------------------------------+---------------------------------+---------------------------------+ + | col1_unnest_placeholder_depth_1 | col1_unnest_placeholder_depth_2 | col2_unnest_placeholder_depth_1 | + +---------------------------------+---------------------------------+---------------------------------+ + | [1, 2, 3] | 1 | a | + | | 2 | b | + | [4, 5] | 3 | | + | [1, 2, 3] | | a | + | | | b | + | [4, 5] | | | + | [1, 2, 3] | 4 | a | + | | 5 | b | + | [4, 5] | | | + | [7, 8, 9, 10] | 7 | c | + | | 8 | d | + | [11, 12, 13] | 9 | | + | | 10 | | + | [7, 8, 9, 10] | | c | + | | | d | + | [11, 12, 13] | | | + | [7, 8, 9, 10] | 11 | c | + | | 12 | d | + | [11, 12, 13] | 13 | | + | | | e | + +---------------------------------+---------------------------------+---------------------------------+ + "); + Ok(()) + } + #[test] fn test_unnest_list_array() -> Result<()> { // [A, B, C], [], NULL, [D], NULL, [NULL, F] From 71947a11504113a22f51aaa14a2404afb9877950 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Mon, 18 May 2026 23:27:30 +0530 Subject: [PATCH 05/14] added explode_outer slt tests --- datafusion/core/tests/dataframe/mod.rs | 135 ------------ datafusion/expr/src/expr.rs | 61 ++++-- datafusion/expr/src/expr_fn.rs | 3 +- datafusion/expr/src/expr_rewriter/mod.rs | 7 +- datafusion/expr/src/expr_schema.rs | 2 +- datafusion/expr/src/tree_node.rs | 6 +- datafusion/proto/proto/datafusion.proto | 3 + datafusion/proto/src/generated/pbjson.rs | 17 ++ datafusion/proto/src/generated/prost.rs | 4 + .../proto/src/logical_plan/from_proto.rs | 5 +- datafusion/proto/src/logical_plan/to_proto.rs | 3 +- .../tests/cases/roundtrip_logical_plan.rs | 12 + datafusion/sql/src/expr/function.rs | 18 +- datafusion/sql/src/select.rs | 55 ++++- datafusion/sql/src/unparser/expr.rs | 1 + .../spark/generator/explode_outer.slt | 205 ++++++++++++++++++ 16 files changed, 372 insertions(+), 165 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs index 2e79773bc4854..0a4ef37db67cd 100644 --- a/datafusion/core/tests/dataframe/mod.rs +++ b/datafusion/core/tests/dataframe/mod.rs @@ -4400,73 +4400,6 @@ async fn unnest_column_nulls() -> Result<()> { Ok(()) } -/// Spark `explode_outer` semantics on a string list. Each input row is one -/// of: a list with values, a list with values *including a NULL element*, -/// an empty list, or a NULL list. The new `PreserveAndExpandEmpty` mode -/// must preserve inner NULL elements (they are different from "empty"), -/// preserve NULL input rows as a single NULL output row, *and* also turn -/// each empty list into a single NULL output row. -#[tokio::test] -async fn unnest_explode_outer_string_list_with_inner_nulls() -> Result<()> { - let mut list_builder = ListBuilder::new(StringBuilder::new()); - let mut id_builder = StringBuilder::new(); - - // ['x', 'y'], id = A - list_builder.values().append_value("x"); - list_builder.values().append_value("y"); - list_builder.append(true); - id_builder.append_value("A"); - - // ['p', NULL, 'q'], id = B (inner NULL must survive) - list_builder.values().append_value("p"); - list_builder.values().append_null(); - list_builder.values().append_value("q"); - list_builder.append(true); - id_builder.append_value("B"); - - // [], id = C (empty list) - list_builder.append(true); - id_builder.append_value("C"); - - // NULL, id = D (null list) - list_builder.append(false); - id_builder.append_value("D"); - - let batch = RecordBatch::try_from_iter(vec![ - ("tag", Arc::new(list_builder.finish()) as ArrayRef), - ("id", Arc::new(id_builder.finish()) as ArrayRef), - ])?; - - let ctx = SessionContext::new(); - ctx.register_batch("tags", batch)?; - let df = ctx.table("tags").await?; - - let options = UnnestOptions::new() - .with_null_handling(datafusion_common::NullHandling::PreserveAndExpandEmpty); - let results = df - .unnest_columns_with_options(&["tag"], options)? - .collect() - .await?; - assert_snapshot!( - batches_to_string(&results), - @r" - +-----+----+ - | tag | id | - +-----+----+ - | x | A | - | y | A | - | p | B | - | | B | - | q | B | - | | C | - | | D | - +-----+----+ - " - ); - - Ok(()) -} - /// Spark `explode_outer` on a list-of-struct column. Verifies that /// (a) struct elements unnest into flattened sub-columns and /// (b) NULL and empty lists both still produce a single output row whose @@ -4595,74 +4528,6 @@ async fn unnest_explode_outer_fixed_size_list() -> Result<()> { Ok(()) } -/// Chain `explode` (drop nulls + empties) followed by `explode_outer` -/// (preserve nulls + empties) on a `list>` input, exercising the -/// "mixed modes in one query plan" path: each `UnnestExec` carries its own -/// `UnnestOptions`. The outer NULL row is dropped by the first stage, then -/// any NULL inner sub-list produced by the first stage is *preserved* by -/// the second stage as a NULL output row. -#[tokio::test] -async fn unnest_explode_then_explode_outer() -> Result<()> { - // outer: [[1,2,3], null, [4,5]], [[7], []], null - let inner = ListArray::from_iter_primitive::(vec![ - Some(vec![Some(1), Some(2), Some(3)]), - None, - Some(vec![Some(4), Some(5)]), - Some(vec![Some(7)]), - Some(vec![]), - ]); - let inner_ref = Arc::new(inner) as ArrayRef; - let offsets = arrow::buffer::OffsetBuffer::::from_lengths([3, 2, 0]); - let validity = arrow::buffer::NullBuffer::from(vec![true, true, false]); - let outer_field = Arc::new(Field::new("item", inner_ref.data_type().clone(), true)); - let outer = ListArray::new(outer_field, offsets, inner_ref, Some(validity)); - let id = Int32Array::from(vec![100, 200, 300]); - - let batch = RecordBatch::try_from_iter(vec![ - ("col", Arc::new(outer) as ArrayRef), - ("id", Arc::new(id) as ArrayRef), - ])?; - let ctx = SessionContext::new(); - ctx.register_batch("t", batch)?; - let df = ctx.table("t").await?; - - // Stage 1: explode — outer NULL row is dropped, empty inner row also dropped. - // Stage 2: explode_outer — any NULL produced by stage 1 (the null sub-list - // from row 0) is kept as a NULL output row. - let results = df - .unnest_columns_with_options( - &["col"], - UnnestOptions::new().with_preserve_nulls(false), - )? - .unnest_columns_with_options( - &["col"], - UnnestOptions::new().with_null_handling( - datafusion_common::NullHandling::PreserveAndExpandEmpty, - ), - )? - .collect() - .await?; - assert_snapshot!( - batches_to_string(&results), - @r" - +-----+-----+ - | col | id | - +-----+-----+ - | 1 | 100 | - | 2 | 100 | - | 3 | 100 | - | | 100 | - | 4 | 100 | - | 5 | 100 | - | 7 | 200 | - | | 200 | - +-----+-----+ - " - ); - - Ok(()) -} - #[tokio::test] async fn unnest_fixed_list() -> Result<()> { let batch = get_fixed_list_batch()?; diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs index 36ef6cf1f5ba9..b02313aec901d 100644 --- a/datafusion/expr/src/expr.rs +++ b/datafusion/expr/src/expr.rs @@ -650,22 +650,43 @@ pub fn intersect_metadata_for_union<'a>( } /// UNNEST expression. +/// +/// When `outer` is `true`, the unnest should preserve `NULL` and empty input +/// lists by emitting a single `NULL` output row for each, matching Spark +/// `explode_outer` semantics. When `false` (the historical default), the +/// behavior is identical to the plain `UNNEST(col)` SQL form. #[derive(Clone, PartialEq, Eq, PartialOrd, Hash, Debug)] pub struct Unnest { pub expr: Box, + /// Spark `explode_outer`-style behavior: also expand empty input lists + /// into a single `NULL` output row. + pub outer: bool, } impl Unnest { - /// Create a new Unnest expression. + /// Create a new Unnest expression with default (non-outer) semantics. pub fn new(expr: Expr) -> Self { Self { expr: Box::new(expr), + outer: false, } } - /// Create a new Unnest expression. + /// Create a new Unnest expression with default (non-outer) semantics. pub fn new_boxed(boxed: Box) -> Self { - Self { expr: boxed } + Self { + expr: boxed, + outer: false, + } + } + + /// Create a new Unnest expression with `explode_outer` semantics: + /// `NULL` and empty input lists each produce a single `NULL` row. + pub fn new_outer(expr: Expr) -> Self { + Self { + expr: Box::new(expr), + outer: true, + } } } @@ -2354,11 +2375,19 @@ impl NormalizeEq for Expr { | (Expr::IsNotTrue(self_expr), Expr::IsNotTrue(other_expr)) | (Expr::IsNotFalse(self_expr), Expr::IsNotFalse(other_expr)) | (Expr::IsNotUnknown(self_expr), Expr::IsNotUnknown(other_expr)) - | (Expr::Negative(self_expr), Expr::Negative(other_expr)) - | ( - Expr::Unnest(Unnest { expr: self_expr }), - Expr::Unnest(Unnest { expr: other_expr }), - ) => self_expr.normalize_eq(other_expr), + | (Expr::Negative(self_expr), Expr::Negative(other_expr)) => { + self_expr.normalize_eq(other_expr) + } + ( + Expr::Unnest(Unnest { + expr: self_expr, + outer: self_outer, + }), + Expr::Unnest(Unnest { + expr: other_expr, + outer: other_outer, + }), + ) => self_outer == other_outer && self_expr.normalize_eq(other_expr), ( Expr::Between(Between { expr: self_expr, @@ -2806,7 +2835,9 @@ impl HashNode for Expr { field.hash(state); column.hash(state); } - Expr::Unnest(Unnest { expr: _expr }) => {} + Expr::Unnest(Unnest { expr: _expr, outer }) => { + outer.hash(state); + } Expr::HigherOrderFunction(HigherOrderFunction { func, args: _args }) => { func.hash(state); } @@ -3026,8 +3057,9 @@ impl Display for SchemaDisplay<'_> { } Expr::Negative(expr) => write!(f, "(- {})", SchemaDisplay(expr)), Expr::Not(expr) => write!(f, "NOT {}", SchemaDisplay(expr)), - Expr::Unnest(Unnest { expr }) => { - write!(f, "UNNEST({})", SchemaDisplay(expr)) + Expr::Unnest(Unnest { expr, outer }) => { + let name = if *outer { "UNNEST_OUTER" } else { "UNNEST" }; + write!(f, "{name}({})", SchemaDisplay(expr)) } Expr::ScalarFunction(ScalarFunction { func, args }) => { match func.schema_name(args) { @@ -3301,8 +3333,9 @@ impl Display for SqlDisplay<'_> { } Expr::Negative(expr) => write!(f, "(- {})", SqlDisplay(expr)), Expr::Not(expr) => write!(f, "NOT {}", SqlDisplay(expr)), - Expr::Unnest(Unnest { expr }) => { - write!(f, "UNNEST({})", SqlDisplay(expr)) + Expr::Unnest(Unnest { expr, outer }) => { + let name = if *outer { "UNNEST_OUTER" } else { "UNNEST" }; + write!(f, "{name}({})", SqlDisplay(expr)) } Expr::SimilarTo(Like { negated, @@ -3655,7 +3688,7 @@ impl Display for Expr { } }, Expr::Placeholder(Placeholder { id, .. }) => write!(f, "{id}"), - Expr::Unnest(Unnest { expr }) => { + Expr::Unnest(Unnest { expr, .. }) => { write!(f, "{UNNEST_COLUMN_PREFIX}({expr})") } Expr::HigherOrderFunction(fun) => { diff --git a/datafusion/expr/src/expr_fn.rs b/datafusion/expr/src/expr_fn.rs index cec6b7ec0565c..e940d20fee795 100644 --- a/datafusion/expr/src/expr_fn.rs +++ b/datafusion/expr/src/expr_fn.rs @@ -386,10 +386,11 @@ pub fn when(when: Expr, then: Expr) -> CaseBuilder { CaseBuilder::new(None, vec![when], vec![then], None) } -/// Create a Unnest expression +/// Create a Unnest expression with default (non-outer) semantics. pub fn unnest(expr: Expr) -> Expr { Expr::Unnest(Unnest { expr: Box::new(expr), + outer: false, }) } diff --git a/datafusion/expr/src/expr_rewriter/mod.rs b/datafusion/expr/src/expr_rewriter/mod.rs index 32a88ab8cf310..3ce2fd6222acc 100644 --- a/datafusion/expr/src/expr_rewriter/mod.rs +++ b/datafusion/expr/src/expr_rewriter/mod.rs @@ -87,13 +87,16 @@ pub fn normalize_col_with_schemas_and_ambiguity_check( using_columns: &[HashSet], ) -> Result { // Normalize column inside Unnest - if let Expr::Unnest(Unnest { expr }) = expr { + if let Expr::Unnest(Unnest { expr, outer }) = expr { let e = normalize_col_with_schemas_and_ambiguity_check( expr.as_ref().clone(), schemas, using_columns, )?; - return Ok(Expr::Unnest(Unnest { expr: Box::new(e) })); + return Ok(Expr::Unnest(Unnest { + expr: Box::new(e), + outer, + })); } expr.transform(|expr| { diff --git a/datafusion/expr/src/expr_schema.rs b/datafusion/expr/src/expr_schema.rs index c989bab3048ad..a8ec6294da384 100644 --- a/datafusion/expr/src/expr_schema.rs +++ b/datafusion/expr/src/expr_schema.rs @@ -157,7 +157,7 @@ impl ExprSchemable for Expr { Expr::Cast(Cast { field, .. }) | Expr::TryCast(TryCast { field, .. }) => { Ok(field.data_type().clone()) } - Expr::Unnest(Unnest { expr }) => { + Expr::Unnest(Unnest { expr, .. }) => { let arg_data_type = expr.get_type(schema)?; // Unnest's output type is the inner type of the list match arg_data_type { diff --git a/datafusion/expr/src/tree_node.rs b/datafusion/expr/src/tree_node.rs index 010441b5a25d1..941fd22ea179f 100644 --- a/datafusion/expr/src/tree_node.rs +++ b/datafusion/expr/src/tree_node.rs @@ -49,7 +49,7 @@ impl TreeNode for Expr { ) -> Result { match self { Expr::Alias(Alias { expr, .. }) - | Expr::Unnest(Unnest { expr }) + | Expr::Unnest(Unnest { expr, .. }) | Expr::Not(expr) | Expr::IsNotNull(expr) | Expr::IsTrue(expr) @@ -150,9 +150,9 @@ impl TreeNode for Expr { quantifier, }) }), - Expr::Unnest(Unnest { expr, .. }) => expr + Expr::Unnest(Unnest { expr, outer }) => expr .map_elements(f)? - .update_data(|expr| Expr::Unnest(Unnest { expr })), + .update_data(|expr| Expr::Unnest(Unnest { expr, outer })), Expr::Alias(Alias { expr, relation, diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto index d9265b9c25ea8..109fce8ba4622 100644 --- a/datafusion/proto/proto/datafusion.proto +++ b/datafusion/proto/proto/datafusion.proto @@ -557,6 +557,9 @@ message NegativeNode { message Unnest { repeated LogicalExprNode exprs = 1; + // When true, this Unnest expression has Spark `explode_outer` semantics: + // NULL and empty input lists both produce a single NULL output row. + bool outer = 2; } message InListNode { diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs index 1676725f52593..7da91e112af99 100644 --- a/datafusion/proto/src/generated/pbjson.rs +++ b/datafusion/proto/src/generated/pbjson.rs @@ -24550,10 +24550,16 @@ impl serde::Serialize for Unnest { if !self.exprs.is_empty() { len += 1; } + if self.outer { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.Unnest", len)?; if !self.exprs.is_empty() { struct_ser.serialize_field("exprs", &self.exprs)?; } + if self.outer { + struct_ser.serialize_field("outer", &self.outer)?; + } struct_ser.end() } } @@ -24565,11 +24571,13 @@ impl<'de> serde::Deserialize<'de> for Unnest { { const FIELDS: &[&str] = &[ "exprs", + "outer", ]; #[allow(clippy::enum_variant_names)] enum GeneratedField { Exprs, + Outer, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -24592,6 +24600,7 @@ impl<'de> serde::Deserialize<'de> for Unnest { { match value { "exprs" => Ok(GeneratedField::Exprs), + "outer" => Ok(GeneratedField::Outer), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -24612,6 +24621,7 @@ impl<'de> serde::Deserialize<'de> for Unnest { V: serde::de::MapAccess<'de>, { let mut exprs__ = None; + let mut outer__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::Exprs => { @@ -24620,10 +24630,17 @@ impl<'de> serde::Deserialize<'de> for Unnest { } exprs__ = Some(map_.next_value()?); } + GeneratedField::Outer => { + if outer__.is_some() { + return Err(serde::de::Error::duplicate_field("outer")); + } + outer__ = Some(map_.next_value()?); + } } } Ok(Unnest { exprs: exprs__.unwrap_or_default(), + outer: outer__.unwrap_or_default(), }) } } diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs index ea235f39a55bc..8e68c1dd4a9e3 100644 --- a/datafusion/proto/src/generated/prost.rs +++ b/datafusion/proto/src/generated/prost.rs @@ -861,6 +861,10 @@ pub struct NegativeNode { pub struct Unnest { #[prost(message, repeated, tag = "1")] pub exprs: ::prost::alloc::vec::Vec, + /// When true, this Unnest expression has Spark `explode_outer` semantics: + /// NULL and empty input lists both produce a single NULL output row. + #[prost(bool, tag = "2")] + pub outer: bool, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct InListNode { diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs index 2c7ecd6da27db..4dffa14f32bfa 100644 --- a/datafusion/proto/src/logical_plan/from_proto.rs +++ b/datafusion/proto/src/logical_plan/from_proto.rs @@ -570,7 +570,10 @@ pub fn parse_expr( if exprs.len() != 1 { return Err(proto_error("Unnest must have exactly one expression")); } - Ok(Expr::Unnest(Unnest::new(exprs.swap_remove(0)))) + Ok(Expr::Unnest(Unnest { + expr: Box::new(exprs.swap_remove(0)), + outer: unnest.outer, + })) } ExprType::InList(in_list) => Ok(Expr::InList(InList::new( Box::new(parse_required_expr( diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs index d2be1db26761d..b04b77e91f69c 100644 --- a/datafusion/proto/src/logical_plan/to_proto.rs +++ b/datafusion/proto/src/logical_plan/to_proto.rs @@ -562,9 +562,10 @@ pub fn serialize_expr( expr_type: Some(ExprType::Negative(expr)), } } - Expr::Unnest(Unnest { expr }) => { + Expr::Unnest(Unnest { expr, outer }) => { let expr = protobuf::Unnest { exprs: vec![serialize_expr(expr.as_ref(), codec)?], + outer: *outer, }; protobuf::LogicalExprNode { expr_type: Some(ExprType::Unnest(expr)), diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index 3e79ddab723eb..262bcc05e17de 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -2321,6 +2321,18 @@ fn roundtrip_inlist() { fn roundtrip_unnest() { let test_expr = Expr::Unnest(Unnest { expr: Box::new(col("col")), + outer: false, + }); + + let ctx = SessionContext::new(); + roundtrip_expr_test(test_expr, ctx); +} + +#[test] +fn roundtrip_unnest_outer() { + let test_expr = Expr::Unnest(Unnest { + expr: Box::new(col("col")), + outer: true, }); let ctx = SessionContext::new(); diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index 67abb8b822063..f1cf559f3ab08 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -546,15 +546,25 @@ impl SqlToRel<'_, S> { } } - // Build Unnest expression - if name.eq("unnest") { + // Build Unnest expression. + // + // `unnest` is DataFusion's native SQL name. `explode` and + // `explode_outer` are Spark/Hive aliases — `explode` matches plain + // `unnest` (drops NULL and empty input lists), while `explode_outer` + // sets `outer = true` so the downstream planner picks + // `NullHandling::PreserveAndExpandEmpty`. + if name.eq("unnest") || name.eq("explode") || name.eq("explode_outer") { + let outer = name.eq("explode_outer"); let mut exprs = self.function_args_to_expr(args, schema, planner_context)?; if exprs.len() != 1 { - return plan_err!("unnest() requires exactly one argument"); + return plan_err!("{name}() requires exactly one argument"); } let expr = exprs.swap_remove(0); Self::check_unnest_arg(&expr, schema)?; - return Ok(Expr::Unnest(Unnest::new(expr))); + return Ok(Expr::Unnest(Unnest { + expr: Box::new(expr), + outer, + })); } if !order_by.is_empty() && is_function_window { diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index b7f7d80e70815..717db4f628eb5 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -32,9 +32,10 @@ use arrow::datatypes::DataType; use datafusion_common::error::DataFusionErrorBuilder; use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; use datafusion_common::{Column, DFSchema, DFSchemaRef, Result, not_impl_err, plan_err}; -use datafusion_common::{RecursionUnnestOption, UnnestOptions}; +use datafusion_common::{NullHandling, RecursionUnnestOption, UnnestOptions}; use datafusion_expr::ExprSchemable; use datafusion_expr::builder::get_struct_unnested_columns; +use datafusion_expr::expr::Unnest as UnnestExpr; use datafusion_expr::expr::{PlannedReplaceSelectItem, WildcardOptions}; use datafusion_expr::expr_rewriter::{ normalize_col, normalize_col_with_schemas_and_ambiguity_check, normalize_sorts, @@ -502,8 +503,16 @@ impl SqlToRel<'_, S> { } break; } else { - // Set preserve_nulls to false to ensure compatibility with DuckDB and PostgreSQL - let mut unnest_options = UnnestOptions::new().with_preserve_nulls(false); + // The default SQL `UNNEST` matches DuckDB/PostgreSQL: drop + // both NULL and empty input lists. Spark `explode_outer` + // (modelled as `Unnest { outer: true }`) overrides that + // and selects `NullHandling::PreserveAndExpandEmpty`. Mixing + // the two in a single SELECT is a planning error because + // `UnnestOptions` is per-`UnnestExec`, not per-column. + let null_handling = + collect_unnest_null_handling(&intermediate_select_exprs)?; + let mut unnest_options = + UnnestOptions::new().with_null_handling(null_handling); let mut unnest_col_vec = vec![]; for (col, maybe_list_unnest) in unnest_columns.into_iter() { @@ -1275,3 +1284,43 @@ fn has_unnest_expr_recursively(expr: &Expr) -> bool { }); has_unnest } + +/// Walk `select_exprs`, observe every [`Expr::Unnest`] inside them, and +/// derive the [`NullHandling`] mode for the resulting [`UnnestOptions`]. +/// +/// * No unnest with `outer = true` → [`NullHandling::Drop`] (default SQL +/// `UNNEST(...)` semantics, matching DuckDB/PostgreSQL). +/// * Every unnest with `outer = true` → [`NullHandling::PreserveAndExpandEmpty`] +/// (Spark `explode_outer(...)` semantics). +/// * A mix of `outer = true` and `outer = false` in one SELECT → planning +/// error, because `UnnestOptions` applies per `Unnest` plan node, not +/// per output column. +fn collect_unnest_null_handling(select_exprs: &[Expr]) -> Result { + let mut saw_outer = false; + let mut saw_inner = false; + for expr in select_exprs { + expr.apply(|e| { + if let Expr::Unnest(UnnestExpr { outer, .. }) = e { + if *outer { + saw_outer = true; + } else { + saw_inner = true; + } + } + Ok(TreeNodeRecursion::Continue) + })?; + } + if saw_outer && saw_inner { + return plan_err!( + "Cannot mix `unnest(...)` (or `explode(...)`) with \ + `explode_outer(...)` in the same SELECT — the unnest operator \ + carries a single null-handling mode. Split the query so each \ + unnest projection uses one mode." + ); + } + Ok(if saw_outer { + NullHandling::PreserveAndExpandEmpty + } else { + NullHandling::Drop + }) +} diff --git a/datafusion/sql/src/unparser/expr.rs b/datafusion/sql/src/unparser/expr.rs index dde383ea26c39..1e41a5bd9ce5f 100644 --- a/datafusion/sql/src/unparser/expr.rs +++ b/datafusion/sql/src/unparser/expr.rs @@ -2402,6 +2402,7 @@ mod tests { name: "array_col".to_string(), spans: Spans::new(), })), + outer: false, }), r#"UNNEST("table".array_col)"#, ), diff --git a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt new file mode 100644 index 0000000000000..ae51acf954160 --- /dev/null +++ b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt @@ -0,0 +1,205 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +####################################################################### +# Spark `explode_outer` / `explode` Tests +# +# `explode_outer(col)` matches Spark semantics: +# - rows whose input list is NULL or empty produce a single output row +# containing NULL, +# - rows with values are exploded element-by-element. +# `explode(col)` is a Spark alias for plain SQL `unnest(col)` and drops +# both NULL and empty input lists. +####################################################################### + +############################## +# 1. Int list: explode vs explode_outer +############################## + +statement ok +CREATE TABLE int_lists (id INT, xs INT[]) AS VALUES + (1, [10, 20, 30]), + (2, []), + (3, NULL), + (4, [40]), + (5, [NULL, 50]); + +# Plain explode: drops both NULL and empty rows. Inner NULL elements survive. +query II +SELECT id, explode(xs) AS x FROM int_lists ORDER BY id, x; +---- +1 10 +1 20 +1 30 +4 40 +5 50 +5 NULL + +# explode_outer: NULL and empty input lists each produce one NULL row. +query II +SELECT id, explode_outer(xs) AS x FROM int_lists ORDER BY id, x; +---- +1 10 +1 20 +1 30 +2 NULL +3 NULL +4 40 +5 50 +5 NULL + +############################## +# 2. String list with inner NULLs +############################## + +statement ok +CREATE TABLE str_lists (id TEXT, tags TEXT[]) AS VALUES + ('A', ['x', 'y']), + ('B', ['p', NULL, 'q']), + ('C', []), + ('D', NULL); + +# Inner NULL elements must survive (they are not the same as "empty"), +# while NULL and empty input lists become a single NULL output row. +query TT +SELECT id, explode_outer(tags) AS tag FROM str_lists ORDER BY id, tag; +---- +A x +A y +B p +B q +B NULL +C NULL +D NULL + +############################## +# 3. Mixed list lengths — verify row-wise expansion +############################## + +statement ok +CREATE TABLE varied_lists (id INT, xs INT[]) AS VALUES + (1, [1, 2, 3, 4]), + (2, [5]), + (3, []), + (4, NULL); + +query II +SELECT id, explode_outer(xs) AS x FROM varied_lists ORDER BY id, x; +---- +1 1 +1 2 +1 3 +1 4 +2 5 +3 NULL +4 NULL + +############################## +# 4. Aliased output column +############################## + +query II +SELECT id, explode_outer(xs) AS unwrapped FROM int_lists WHERE id = 2; +---- +2 NULL + +############################## +# 5. Mixing `unnest` / `explode` and `explode_outer` in one SELECT is an error +# UnnestOptions is per-UnnestExec, so we refuse to silently pick one mode. +############################## + +statement error DataFusion error: Error during planning: Cannot mix `unnest\(\.\.\.\)` \(or `explode\(\.\.\.\)`\) with `explode_outer\(\.\.\.\)` in the same SELECT +SELECT unnest(xs), explode_outer(xs) FROM int_lists; + +statement error DataFusion error: Error during planning: Cannot mix `unnest\(\.\.\.\)` \(or `explode\(\.\.\.\)`\) with `explode_outer\(\.\.\.\)` in the same SELECT +SELECT explode(xs), explode_outer(xs) FROM int_lists; + +############################## +# 6. Chained explode → explode_outer via subquery. +# `explode(xs)` (inner) drops NULL and empty outer rows, then +# `explode_outer(ys)` (outer) preserves NULL and empty sub-lists from +# the inner explosion. +############################## + +statement ok +CREATE TABLE nested_lists (id INT, xs INT[][]) AS VALUES + (100, [[1, 2, 3], NULL, [4, 5]]), + (200, [[7], []]), + (300, NULL); + +query II +SELECT id, explode_outer(ys) AS y +FROM (SELECT id, explode(xs) AS ys FROM nested_lists) +ORDER BY id, y; +---- +100 1 +100 2 +100 3 +100 4 +100 5 +100 NULL +200 7 +200 NULL + +statement ok +DROP TABLE nested_lists; + +############################## +# 7. `explode_outer` agrees with `explode` when no NULL or empty rows exist. +############################## + +statement ok +CREATE TABLE dense_lists (id INT, xs INT[]) AS VALUES + (1, [10, 20]), + (2, [30]), + (3, [40, 50, 60]); + +query II +SELECT id, explode(xs) AS x FROM dense_lists ORDER BY id, x; +---- +1 10 +1 20 +2 30 +3 40 +3 50 +3 60 + +query II +SELECT id, explode_outer(xs) AS x FROM dense_lists ORDER BY id, x; +---- +1 10 +1 20 +2 30 +3 40 +3 50 +3 60 + +############################## +# Cleanup +############################## + +statement ok +DROP TABLE int_lists; + +statement ok +DROP TABLE str_lists; + +statement ok +DROP TABLE varied_lists; + +statement ok +DROP TABLE dense_lists; From 6ba8e2730e0fdeda9d058fd8829b64733961f73e Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Wed, 27 May 2026 18:58:00 +0530 Subject: [PATCH 06/14] format fix --- datafusion/sql/src/select.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index bd8bbade04627..fc536560828db 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -671,8 +671,7 @@ impl SqlToRel<'_, S> { // `NullHandling::PreserveAndExpandEmpty`. Mixing the two in a // single SELECT is a planning error because `UnnestOptions` is // per-`UnnestExec`, not per-column. - let null_handling = - collect_unnest_null_handling(&intermediate_expr_groups)?; + let null_handling = collect_unnest_null_handling(&intermediate_expr_groups)?; let mut unnest_options = UnnestOptions::new().with_null_handling(null_handling); let mut unnest_col_vec = vec![]; From 7e789a2c45980f01099e18af49b602d5c070f939 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sat, 30 May 2026 16:45:49 +0530 Subject: [PATCH 07/14] adding datafusion.spark.allow_multiple_generators config --- datafusion/common/src/config.rs | 21 ++++++++++++ datafusion/sql/src/select.rs | 34 +++++++++++++++++-- .../test_files/information_schema.slt | 2 ++ .../spark/generator/explode_outer.slt | 34 +++++++++++++++++++ docs/source/user-guide/configs.md | 1 + 5 files changed, 90 insertions(+), 2 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 3e3ab3429a2fb..fbef3f6222aa1 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1483,6 +1483,21 @@ impl<'a> TryFrom<&'a FormatOptions> for arrow::util::display::FormatOptions<'a> } } +config_namespace! { + /// Options controlling DataFusion's Spark-compatibility layer (functions + /// under `datafusion/spark` and Spark-specific planner behavior). Keys + /// here mirror their `spark.sql.*` equivalents in Apache Spark. + pub struct SparkOptions { + /// Whether to allow more than one generator function (`unnest`, `explode`, `explode_outer`, ...) in a single `SELECT` projection. + /// + /// - `true` (default): DataFusion's native behavior — multiple generators are allowed in one `SELECT`. + /// - `false`: Spark-strict behavior — any `SELECT` with more than one generator is rejected with `[UNSUPPORTED_GENERATOR.MULTI_GENERATOR]` at planning time, matching Spark's `AnalysisException`. + /// + /// A `SELECT` that mixes `explode_outer` with `unnest`/`explode` is always rejected regardless of this setting, because the two require different null-handling modes on a single `Unnest` plan node. + pub allow_multiple_generators: bool, default = true + } +} + /// A key value pair, with a corresponding description #[derive(Debug, Clone, Hash, PartialEq, Eq)] pub struct ConfigEntry { @@ -1514,6 +1529,9 @@ pub struct ConfigOptions { pub extensions: Extensions, /// Formatting options when printing batches pub format: FormatOptions, + /// Spark-compatibility options (functions under `datafusion/spark` and + /// Spark-specific planner behavior) + pub spark: SparkOptions, } impl ConfigField for ConfigOptions { @@ -1524,6 +1542,7 @@ impl ConfigField for ConfigOptions { self.explain.visit(v, "datafusion.explain", ""); self.sql_parser.visit(v, "datafusion.sql_parser", ""); self.format.visit(v, "datafusion.format", ""); + self.spark.visit(v, "datafusion.spark", ""); } fn set(&mut self, key: &str, value: &str) -> Result<()> { @@ -1536,6 +1555,7 @@ impl ConfigField for ConfigOptions { "explain" => self.explain.set(rem, value), "sql_parser" => self.sql_parser.set(rem, value), "format" => self.format.set(rem, value), + "spark" => self.spark.set(rem, value), _ => _config_err!("Config value \"{key}\" not found on ConfigOptions"), } } @@ -1575,6 +1595,7 @@ impl ConfigField for ConfigOptions { "explain" => self.explain.reset(rem), "sql_parser" => self.sql_parser.reset(rem), "format" => self.format.reset(rem), + "spark" => self.spark.reset(rem), other => _config_err!("Config value \"{other}\" not found on ConfigOptions"), } } diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index fc536560828db..119dd5454313d 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -671,7 +671,19 @@ impl SqlToRel<'_, S> { // `NullHandling::PreserveAndExpandEmpty`. Mixing the two in a // single SELECT is a planning error because `UnnestOptions` is // per-`UnnestExec`, not per-column. - let null_handling = collect_unnest_null_handling(&intermediate_expr_groups)?; + // + // When `datafusion.spark.allow_multiple_generators = false`, any + // SELECT with more than one generator (even same-mode) is rejected + // to match Spark's `UNSUPPORTED_GENERATOR.MULTI_GENERATOR`. + let allow_multiple_generators = self + .context_provider + .options() + .spark + .allow_multiple_generators; + let null_handling = collect_unnest_null_handling( + &intermediate_expr_groups, + allow_multiple_generators, + )?; let mut unnest_options = UnnestOptions::new().with_null_handling(null_handling); let mut unnest_col_vec = vec![]; @@ -1468,9 +1480,18 @@ fn has_unnest_expr_recursively(expr: &Expr) -> bool { /// * A mix of `outer = true` and `outer = false` in one SELECT → planning /// error, because `UnnestOptions` applies per `Unnest` plan node, not /// per output column. -fn collect_unnest_null_handling(expr_groups: &[Vec]) -> Result { +/// * When `allow_multiple_generators` is `false`, any SELECT with more than +/// one generator is rejected with +/// `[UNSUPPORTED_GENERATOR.MULTI_GENERATOR]` (Spark-strict mode). When +/// `true` (the default), DataFusion's native multi-generator support is +/// preserved. +fn collect_unnest_null_handling( + expr_groups: &[Vec], + allow_multiple_generators: bool, +) -> Result { let mut saw_outer = false; let mut saw_inner = false; + let mut generator_count: usize = 0; for group in expr_groups { for expr in group { expr.apply(|e| { @@ -1480,6 +1501,7 @@ fn collect_unnest_null_handling(expr_groups: &[Vec]) -> Result]) -> Result 1 { + return plan_err!( + "[UNSUPPORTED_GENERATOR.MULTI_GENERATOR] Only one generator \ + function (`unnest`, `explode`, `explode_outer`, ...) is allowed \ + per SELECT under Spark-strict mode \ + (`datafusion.spark.allow_multiple_generators = false`)." + ); + } Ok(if saw_outer { NullHandling::PreserveAndExpandEmpty } else { diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 3bf101f203fbd..71ea98dbeb14d 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -341,6 +341,7 @@ datafusion.runtime.max_temp_directory_size 100G datafusion.runtime.memory_limit unlimited datafusion.runtime.metadata_cache_limit 50M datafusion.runtime.temp_directory NULL +datafusion.spark.allow_multiple_generators true datafusion.sql_parser.collect_spans false datafusion.sql_parser.default_null_ordering nulls_max datafusion.sql_parser.dialect generic @@ -491,6 +492,7 @@ datafusion.runtime.max_temp_directory_size 100G Maximum temporary file directory datafusion.runtime.memory_limit unlimited Maximum memory limit for query execution. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. datafusion.runtime.metadata_cache_limit 50M Maximum memory to use for file metadata cache such as Parquet metadata. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. datafusion.runtime.temp_directory NULL The path to the temporary file directory. +datafusion.spark.allow_multiple_generators true Whether to allow more than one generator function (`unnest`, `explode`, `explode_outer`, ...) in a single `SELECT` projection. - `true` (default): DataFusion's native behavior — multiple generators are allowed in one `SELECT`. - `false`: Spark-strict behavior — any `SELECT` with more than one generator is rejected with `[UNSUPPORTED_GENERATOR.MULTI_GENERATOR]` at planning time, matching Spark's `AnalysisException`. A `SELECT` that mixes `explode_outer` with `unnest`/`explode` is always rejected regardless of this setting, because the two require different null-handling modes on a single `Unnest` plan node. datafusion.sql_parser.collect_spans false When set to true, the source locations relative to the original SQL query (i.e. [`Span`](https://docs.rs/sqlparser/latest/sqlparser/tokenizer/struct.Span.html)) will be collected and recorded in the logical plan nodes. datafusion.sql_parser.default_null_ordering nulls_max Specifies the default null ordering for query results. There are 4 options: - `nulls_max`: Nulls appear last in ascending order. - `nulls_min`: Nulls appear first in ascending order. - `nulls_first`: Nulls always be first in any order. - `nulls_last`: Nulls always be last in any order. By default, `nulls_max` is used to follow Postgres's behavior. postgres rule: datafusion.sql_parser.dialect generic Configure the SQL dialect used by DataFusion's parser; supported values include: Generic, MySQL, PostgreSQL, Hive, SQLite, Snowflake, Redshift, MsSQL, ClickHouse, BigQuery, Ansi, DuckDB and Databricks. diff --git a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt index ae51acf954160..a177de217b3b3 100644 --- a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt +++ b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt @@ -128,6 +128,40 @@ SELECT unnest(xs), explode_outer(xs) FROM int_lists; statement error DataFusion error: Error during planning: Cannot mix `unnest\(\.\.\.\)` \(or `explode\(\.\.\.\)`\) with `explode_outer\(\.\.\.\)` in the same SELECT SELECT explode(xs), explode_outer(xs) FROM int_lists; +############################## +# 5b. Spark-strict mode: `datafusion.spark.allow_multiple_generators = false` +# rejects ANY second generator in one SELECT (matches Spark's +# `UNSUPPORTED_GENERATOR.MULTI_GENERATOR`). Default DataFusion behavior +# (`true`) permits multiple same-mode generators. +############################## + +statement ok +set datafusion.spark.allow_multiple_generators = false; + +# Two same-mode generators in one SELECT — rejected only under strict mode. +# (Aliases sidestep the unique-output-name check so we hit the new error path.) +statement error DataFusion error: Error during planning: \[UNSUPPORTED_GENERATOR\.MULTI_GENERATOR\] Only one generator function +SELECT explode(xs) AS a, explode(xs) AS b FROM int_lists; + +statement error DataFusion error: Error during planning: \[UNSUPPORTED_GENERATOR\.MULTI_GENERATOR\] Only one generator function +SELECT explode_outer(xs) AS a, explode_outer(xs) AS b FROM int_lists; + +# Mixed-mode is rejected regardless of this flag (the mix message wins +# because it's a technical constraint, not a stylistic one). +statement error DataFusion error: Error during planning: Cannot mix `unnest\(\.\.\.\)` \(or `explode\(\.\.\.\)`\) with `explode_outer\(\.\.\.\)` in the same SELECT +SELECT unnest(xs) AS a, explode_outer(xs) AS b FROM int_lists; + +# A single generator still works under strict mode. +query I +SELECT explode_outer(xs) AS x FROM int_lists WHERE id = 1 ORDER BY x; +---- +10 +20 +30 + +statement ok +set datafusion.spark.allow_multiple_generators = true; + ############################## # 6. Chained explode → explode_outer via subquery. # `explode(xs)` (inner) drops NULL and empty outer rows, then diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 9856a13f00306..fd63f62aaa693 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -203,6 +203,7 @@ The following configuration settings are available: | datafusion.format.time_format | %H:%M:%S%.f | Time format for time arrays | | datafusion.format.duration_format | pretty | Duration format. Can be either `"pretty"` or `"ISO8601"` | | datafusion.format.types_info | false | Show types in visual representation batches | +| datafusion.spark.allow_multiple_generators | true | Whether to allow more than one generator function (`unnest`, `explode`, `explode_outer`, ...) in a single `SELECT` projection. - `true` (default): DataFusion's native behavior — multiple generators are allowed in one `SELECT`. - `false`: Spark-strict behavior — any `SELECT` with more than one generator is rejected with `[UNSUPPORTED_GENERATOR.MULTI_GENERATOR]` at planning time, matching Spark's `AnalysisException`. A `SELECT` that mixes `explode_outer` with `unnest`/`explode` is always rejected regardless of this setting, because the two require different null-handling modes on a single `Unnest` plan node. | You can also reset configuration options to default settings via SQL using the `RESET` command. For example, to set and reset `datafusion.execution.batch_size`: From d1dc53f2a7d93aa9fdf2058e0a3f1ccc79881385 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sun, 28 Jun 2026 13:06:19 +0530 Subject: [PATCH 08/14] updated SELECT doc --- docs/source/user-guide/sql/select.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/docs/source/user-guide/sql/select.md b/docs/source/user-guide/sql/select.md index ea96f6ae4528d..feafa3976fb3c 100644 --- a/docs/source/user-guide/sql/select.md +++ b/docs/source/user-guide/sql/select.md @@ -279,6 +279,29 @@ SELECT id, UNNEST(items) FROM orders; items (implicit lateral references such as `FROM orders AS t, UNNEST(t.items)` are not currently supported). +### `explode` and `explode_outer` + +`explode(col)` and `explode_outer(col)` are accepted as aliases for `UNNEST(col)` +for compatibility with Spark and Hive dialects. They differ from plain `UNNEST` +in how `NULL` and empty input lists are handled: + +| Form | `NULL` input list | Empty input list | +| --------------------- | ----------------- | ---------------- | +| `UNNEST(col)` | dropped | dropped | +| `explode(col)` | dropped | dropped | +| `explode_outer(col)` | one `NULL` row | one `NULL` row | + +```sql +SELECT id, explode_outer(tags) AS tag FROM rows; +``` + +An input row with an empty `tags` array or `NULL` `tags` produces one output +row whose `tag` is `NULL`, instead of being dropped. + +`explode_outer` cannot be mixed with `unnest` or `explode` in the same `SELECT` +— the unnest plan node carries a single null-handling mode for all its output +columns, so a mix would be ambiguous. The planner returns an error in that case. + ## WHERE clause ```text From 8c6909f6f59f779938117dd05e5de1bec78c3928 Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sun, 28 Jun 2026 13:10:47 +0530 Subject: [PATCH 09/14] fixed format --- docs/source/user-guide/sql/select.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/source/user-guide/sql/select.md b/docs/source/user-guide/sql/select.md index feafa3976fb3c..317fa5abb189e 100644 --- a/docs/source/user-guide/sql/select.md +++ b/docs/source/user-guide/sql/select.md @@ -285,11 +285,11 @@ are not currently supported). for compatibility with Spark and Hive dialects. They differ from plain `UNNEST` in how `NULL` and empty input lists are handled: -| Form | `NULL` input list | Empty input list | -| --------------------- | ----------------- | ---------------- | -| `UNNEST(col)` | dropped | dropped | -| `explode(col)` | dropped | dropped | -| `explode_outer(col)` | one `NULL` row | one `NULL` row | +| Form | `NULL` input list | Empty input list | +| -------------------- | ----------------- | ---------------- | +| `UNNEST(col)` | dropped | dropped | +| `explode(col)` | dropped | dropped | +| `explode_outer(col)` | one `NULL` row | one `NULL` row | ```sql SELECT id, explode_outer(tags) AS tag FROM rows; From da4160fcf551ad902ca75ebe7adb97b09e61842e Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sun, 28 Jun 2026 14:11:06 +0530 Subject: [PATCH 10/14] removed allow_multiple_generators --- datafusion/sql/src/select.rs | 35 ++----------------- .../spark/generator/explode_outer.slt | 34 ------------------ 2 files changed, 3 insertions(+), 66 deletions(-) diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index 9b2b629a1cae6..1d88eb3921a3e 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -672,19 +672,8 @@ impl SqlToRel<'_, S> { // `NullHandling::PreserveAndExpandEmpty`. Mixing the two in a // single SELECT is a planning error because `UnnestOptions` is // per-`UnnestExec`, not per-column. - // - // When `datafusion.spark.allow_multiple_generators = false`, any - // SELECT with more than one generator (even same-mode) is rejected - // to match Spark's `UNSUPPORTED_GENERATOR.MULTI_GENERATOR`. - let allow_multiple_generators = self - .context_provider - .options() - .spark - .allow_multiple_generators; - let null_handling = collect_unnest_null_handling( - &intermediate_expr_groups, - allow_multiple_generators, - )?; + let null_handling = + collect_unnest_null_handling(&intermediate_expr_groups)?; let mut unnest_options = UnnestOptions::new().with_null_handling(null_handling); let mut unnest_col_vec = vec![]; @@ -1482,18 +1471,9 @@ fn has_unnest_expr_recursively(expr: &Expr) -> bool { /// * A mix of `outer = true` and `outer = false` in one SELECT → planning /// error, because `UnnestOptions` applies per `Unnest` plan node, not /// per output column. -/// * When `allow_multiple_generators` is `false`, any SELECT with more than -/// one generator is rejected with -/// `[UNSUPPORTED_GENERATOR.MULTI_GENERATOR]` (Spark-strict mode). When -/// `true` (the default), DataFusion's native multi-generator support is -/// preserved. -fn collect_unnest_null_handling( - expr_groups: &[Vec], - allow_multiple_generators: bool, -) -> Result { +fn collect_unnest_null_handling(expr_groups: &[Vec]) -> Result { let mut saw_outer = false; let mut saw_inner = false; - let mut generator_count: usize = 0; for group in expr_groups { for expr in group { expr.apply(|e| { @@ -1503,7 +1483,6 @@ fn collect_unnest_null_handling( } else { saw_inner = true; } - generator_count += 1; } Ok(TreeNodeRecursion::Continue) })?; @@ -1517,14 +1496,6 @@ fn collect_unnest_null_handling( unnest projection uses one mode." ); } - if !allow_multiple_generators && generator_count > 1 { - return plan_err!( - "[UNSUPPORTED_GENERATOR.MULTI_GENERATOR] Only one generator \ - function (`unnest`, `explode`, `explode_outer`, ...) is allowed \ - per SELECT under Spark-strict mode \ - (`datafusion.spark.allow_multiple_generators = false`)." - ); - } Ok(if saw_outer { NullHandling::PreserveAndExpandEmpty } else { diff --git a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt index a177de217b3b3..ae51acf954160 100644 --- a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt +++ b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt @@ -128,40 +128,6 @@ SELECT unnest(xs), explode_outer(xs) FROM int_lists; statement error DataFusion error: Error during planning: Cannot mix `unnest\(\.\.\.\)` \(or `explode\(\.\.\.\)`\) with `explode_outer\(\.\.\.\)` in the same SELECT SELECT explode(xs), explode_outer(xs) FROM int_lists; -############################## -# 5b. Spark-strict mode: `datafusion.spark.allow_multiple_generators = false` -# rejects ANY second generator in one SELECT (matches Spark's -# `UNSUPPORTED_GENERATOR.MULTI_GENERATOR`). Default DataFusion behavior -# (`true`) permits multiple same-mode generators. -############################## - -statement ok -set datafusion.spark.allow_multiple_generators = false; - -# Two same-mode generators in one SELECT — rejected only under strict mode. -# (Aliases sidestep the unique-output-name check so we hit the new error path.) -statement error DataFusion error: Error during planning: \[UNSUPPORTED_GENERATOR\.MULTI_GENERATOR\] Only one generator function -SELECT explode(xs) AS a, explode(xs) AS b FROM int_lists; - -statement error DataFusion error: Error during planning: \[UNSUPPORTED_GENERATOR\.MULTI_GENERATOR\] Only one generator function -SELECT explode_outer(xs) AS a, explode_outer(xs) AS b FROM int_lists; - -# Mixed-mode is rejected regardless of this flag (the mix message wins -# because it's a technical constraint, not a stylistic one). -statement error DataFusion error: Error during planning: Cannot mix `unnest\(\.\.\.\)` \(or `explode\(\.\.\.\)`\) with `explode_outer\(\.\.\.\)` in the same SELECT -SELECT unnest(xs) AS a, explode_outer(xs) AS b FROM int_lists; - -# A single generator still works under strict mode. -query I -SELECT explode_outer(xs) AS x FROM int_lists WHERE id = 1 ORDER BY x; ----- -10 -20 -30 - -statement ok -set datafusion.spark.allow_multiple_generators = true; - ############################## # 6. Chained explode → explode_outer via subquery. # `explode(xs)` (inner) drops NULL and empty outer rows, then From 1c0e522ad5313fb3c43093dfa84283987afd2fbc Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sun, 28 Jun 2026 16:14:37 +0530 Subject: [PATCH 11/14] fixed cargo fmt --- datafusion/sql/src/select.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index 1d88eb3921a3e..675c56e0a8a06 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -672,8 +672,7 @@ impl SqlToRel<'_, S> { // `NullHandling::PreserveAndExpandEmpty`. Mixing the two in a // single SELECT is a planning error because `UnnestOptions` is // per-`UnnestExec`, not per-column. - let null_handling = - collect_unnest_null_handling(&intermediate_expr_groups)?; + let null_handling = collect_unnest_null_handling(&intermediate_expr_groups)?; let mut unnest_options = UnnestOptions::new().with_null_handling(null_handling); let mut unnest_col_vec = vec![]; From 999399a2f00a47b0202adec792824a57cb6bffbd Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sun, 5 Jul 2026 16:55:01 +0530 Subject: [PATCH 12/14] fixed slt tests --- .../spark/generator/explode_outer.slt | 45 ++++++++++++------- 1 file changed, 30 insertions(+), 15 deletions(-) diff --git a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt index ae51acf954160..4386643585597 100644 --- a/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt +++ b/datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt @@ -24,6 +24,11 @@ # - rows with values are exploded element-by-element. # `explode(col)` is a Spark alias for plain SQL `unnest(col)` and drops # both NULL and empty input lists. +# +# Column types on the tables below are inferred from the VALUES rows — +# DataFusion's parser does not accept the PostgreSQL `TYPE[]` array-column +# syntax inside `CREATE TABLE ... AS VALUES`, so tables are declared as +# CTAS over a `VALUES` subquery with aliased column names. ####################################################################### ############################## @@ -31,12 +36,14 @@ ############################## statement ok -CREATE TABLE int_lists (id INT, xs INT[]) AS VALUES +CREATE TABLE int_lists AS +SELECT column1 AS id, column2 AS xs FROM (VALUES (1, [10, 20, 30]), - (2, []), - (3, NULL), (4, [40]), - (5, [NULL, 50]); + (5, [NULL, 50]), + (2, arrow_cast(make_array(), 'List(Int64)')), + (3, NULL) +); # Plain explode: drops both NULL and empty rows. Inner NULL elements survive. query II @@ -67,11 +74,13 @@ SELECT id, explode_outer(xs) AS x FROM int_lists ORDER BY id, x; ############################## statement ok -CREATE TABLE str_lists (id TEXT, tags TEXT[]) AS VALUES +CREATE TABLE str_lists AS +SELECT column1 AS id, column2 AS tags FROM (VALUES ('A', ['x', 'y']), ('B', ['p', NULL, 'q']), - ('C', []), - ('D', NULL); + ('C', arrow_cast(make_array(), 'List(Utf8)')), + ('D', NULL) +); # Inner NULL elements must survive (they are not the same as "empty"), # while NULL and empty input lists become a single NULL output row. @@ -91,11 +100,13 @@ D NULL ############################## statement ok -CREATE TABLE varied_lists (id INT, xs INT[]) AS VALUES +CREATE TABLE varied_lists AS +SELECT column1 AS id, column2 AS xs FROM (VALUES (1, [1, 2, 3, 4]), (2, [5]), - (3, []), - (4, NULL); + (3, arrow_cast(make_array(), 'List(Int64)')), + (4, NULL) +); query II SELECT id, explode_outer(xs) AS x FROM varied_lists ORDER BY id, x; @@ -136,10 +147,12 @@ SELECT explode(xs), explode_outer(xs) FROM int_lists; ############################## statement ok -CREATE TABLE nested_lists (id INT, xs INT[][]) AS VALUES +CREATE TABLE nested_lists AS +SELECT column1 AS id, column2 AS xs FROM (VALUES (100, [[1, 2, 3], NULL, [4, 5]]), - (200, [[7], []]), - (300, NULL); + (200, [[7], arrow_cast(make_array(), 'List(Int64)')]), + (300, NULL) +); query II SELECT id, explode_outer(ys) AS y @@ -163,10 +176,12 @@ DROP TABLE nested_lists; ############################## statement ok -CREATE TABLE dense_lists (id INT, xs INT[]) AS VALUES +CREATE TABLE dense_lists AS +SELECT column1 AS id, column2 AS xs FROM (VALUES (1, [10, 20]), (2, [30]), - (3, [40, 50, 60]); + (3, [40, 50, 60]) +); query II SELECT id, explode(xs) AS x FROM dense_lists ORDER BY id, x; From 7d34be07feea6527d68e3e2664796926aacc47ae Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Wed, 22 Jul 2026 20:54:54 +0530 Subject: [PATCH 13/14] removed spark from comments --- datafusion/common/src/unnest.rs | 8 ++++---- datafusion/expr/src/expr.rs | 14 +++++++------- datafusion/physical-plan/src/unnest.rs | 5 ++--- datafusion/sql/src/expr/function.rs | 2 +- datafusion/sql/src/select.rs | 5 +++-- 5 files changed, 17 insertions(+), 17 deletions(-) diff --git a/datafusion/common/src/unnest.rs b/datafusion/common/src/unnest.rs index 3306e79a6f691..58aed390ace78 100644 --- a/datafusion/common/src/unnest.rs +++ b/datafusion/common/src/unnest.rs @@ -35,7 +35,7 @@ pub enum NullHandling { Preserve, /// Like [`Self::Preserve`], and additionally treat an empty list /// identically to a `NULL` list, producing a single output row - /// containing `NULL`. Matches Spark's `explode_outer` semantics. + /// containing `NULL`. PreserveAndExpandEmpty, } @@ -143,8 +143,8 @@ impl UnnestOptions { /// `preserve_nulls` flag onto [`NullHandling`]. /// /// `true` maps to [`NullHandling::Preserve`]; `false` maps to - /// [`NullHandling::Drop`]. To opt into Spark `explode_outer` - /// semantics, call [`Self::with_null_handling`] directly with + /// [`NullHandling::Drop`]. To opt into the new empty-list-preserving + /// mode, call [`Self::with_null_handling`] directly with /// [`NullHandling::PreserveAndExpandEmpty`]. pub fn with_preserve_nulls(self, preserve_nulls: bool) -> Self { let null_handling = if preserve_nulls { @@ -162,7 +162,7 @@ impl UnnestOptions { } /// Returns true if empty input lists should produce a single - /// output row containing `NULL` (Spark `explode_outer` semantics). + /// output row containing `NULL`. pub fn expand_empty_as_null(&self) -> bool { matches!(self.null_handling, NullHandling::PreserveAndExpandEmpty) } diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs index 5d86a06485c48..5affcf55c5f61 100644 --- a/datafusion/expr/src/expr.rs +++ b/datafusion/expr/src/expr.rs @@ -673,14 +673,14 @@ pub fn intersect_metadata_for_union<'a>( /// UNNEST expression. /// /// When `outer` is `true`, the unnest should preserve `NULL` and empty input -/// lists by emitting a single `NULL` output row for each, matching Spark -/// `explode_outer` semantics. When `false` (the historical default), the -/// behavior is identical to the plain `UNNEST(col)` SQL form. +/// lists by emitting a single `NULL` output row for each. When `false` (the +/// historical default), the behavior is identical to the plain `UNNEST(col)` +/// SQL form: `NULL` and empty input lists are dropped from the output. #[derive(Clone, PartialEq, Eq, PartialOrd, Hash, Debug)] pub struct Unnest { pub expr: Box, - /// Spark `explode_outer`-style behavior: also expand empty input lists - /// into a single `NULL` output row. + /// Outer-unnest behavior: also expand empty input lists into a single + /// `NULL` output row (in addition to preserving `NULL` input rows). pub outer: bool, } @@ -701,8 +701,8 @@ impl Unnest { } } - /// Create a new Unnest expression with `explode_outer` semantics: - /// `NULL` and empty input lists each produce a single `NULL` row. + /// Create a new Unnest expression with outer-unnest semantics: `NULL` + /// and empty input lists each produce a single `NULL` output row. pub fn new_outer(expr: Expr) -> Self { Self { expr: Box::new(expr), diff --git a/datafusion/physical-plan/src/unnest.rs b/datafusion/physical-plan/src/unnest.rs index 996b1a32d69ba..c896ea896bb6f 100644 --- a/datafusion/physical-plan/src/unnest.rs +++ b/datafusion/physical-plan/src/unnest.rs @@ -775,8 +775,7 @@ fn build_batch( /// ``` /// /// With [`datafusion_common::NullHandling::PreserveAndExpandEmpty`], empty input lists are -/// also bumped to length 1 so they produce a single `NULL` row, matching -/// Spark `explode_outer` semantics: +/// also bumped to length 1 so they produce a single `NULL` output row: /// /// ```ignore /// longest_length: [3, 1, 1, 2] @@ -1300,7 +1299,7 @@ mod tests { #[test] fn test_build_batch_preserve_and_expand_empty() -> Result<()> { // c1: [A, B, C], [], NULL, [D], NULL, [NULL, F] c2: 1, 2, 3, 4, 5, 6 - // Expected for `NullHandling::PreserveAndExpandEmpty` (Spark explode_outer): + // Expected for `NullHandling::PreserveAndExpandEmpty`: // [A, B, C] -> three rows with c2 = 1, 1, 1 // [] -> one row with c2 = 2 and unnested value NULL // NULL -> one row with c2 = 3 and unnested value NULL diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index 9d686939bbb0b..46f8bdceae89b 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -549,7 +549,7 @@ impl SqlToRel<'_, S> { // Build Unnest expression. // // `unnest` is DataFusion's native SQL name. `explode` and - // `explode_outer` are Spark/Hive aliases — `explode` matches plain + // `explode_outer` are accepted aliases — `explode` matches plain // `unnest` (drops NULL and empty input lists), while `explode_outer` // sets `outer = true` so the downstream planner picks // `NullHandling::PreserveAndExpandEmpty`. diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index 675c56e0a8a06..9ede15457f9ad 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -667,7 +667,7 @@ impl SqlToRel<'_, S> { } // The default SQL `UNNEST` matches DuckDB/PostgreSQL: drop both - // NULL and empty input lists. Spark `explode_outer` (modelled as + // NULL and empty input lists. Outer-unnest (modelled as // `Unnest { outer: true }`) overrides that and selects // `NullHandling::PreserveAndExpandEmpty`. Mixing the two in a // single SELECT is a planning error because `UnnestOptions` is @@ -1466,7 +1466,8 @@ fn has_unnest_expr_recursively(expr: &Expr) -> bool { /// * No unnest with `outer = true` → [`NullHandling::Drop`] (default SQL /// `UNNEST(...)` semantics, matching DuckDB/PostgreSQL). /// * Every unnest with `outer = true` → [`NullHandling::PreserveAndExpandEmpty`] -/// (Spark `explode_outer(...)` semantics). +/// (outer-unnest semantics: `NULL` and empty input lists each produce a +/// single `NULL` output row). /// * A mix of `outer = true` and `outer = false` in one SELECT → planning /// error, because `UnnestOptions` applies per `Unnest` plan node, not /// per output column. From 021e18ed9708a17e9d96ebf8ec973987b49fc36e Mon Sep 17 00:00:00 2001 From: Sudarshan Date: Sat, 25 Jul 2026 16:34:29 +0530 Subject: [PATCH 14/14] tests moved and changed as unnest_outer --- datafusion/core/tests/dataframe/mod.rs | 25 +- .../proto-models/proto/datafusion.proto | 6 +- datafusion/sql/src/expr/function.rs | 14 +- datafusion/sql/src/select.rs | 7 +- .../spark/generator/explode_outer.slt | 220 ------------------ datafusion/sqllogictest/test_files/unnest.slt | 183 +++++++++++++++ docs/source/user-guide/sql/select.md | 28 +-- 7 files changed, 222 insertions(+), 261 deletions(-) delete mode 100644 datafusion/sqllogictest/test_files/spark/generator/explode_outer.slt diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs index 4b39f82a9f419..73a9177ab738a 100644 --- a/datafusion/core/tests/dataframe/mod.rs +++ b/datafusion/core/tests/dataframe/mod.rs @@ -4388,8 +4388,8 @@ async fn unnest_column_nulls() -> Result<()> { " ); - // Spark `explode_outer` semantics: NULL and empty lists both produce a - // single output row containing NULL. + // Outer-unnest semantics: NULL and empty lists both produce a single + // output row containing NULL. let options = UnnestOptions::new() .with_null_handling(datafusion_common::NullHandling::PreserveAndExpandEmpty); let results = df @@ -4414,12 +4414,12 @@ async fn unnest_column_nulls() -> Result<()> { Ok(()) } -/// Spark `explode_outer` on a list-of-struct column. Verifies that +/// Outer-unnest on a list-of-struct column. Verifies that /// (a) struct elements unnest into flattened sub-columns and /// (b) NULL and empty lists both still produce a single output row whose /// struct sub-columns are all NULL. #[tokio::test] -async fn unnest_explode_outer_list_of_struct() -> Result<()> { +async fn unnest_outer_list_of_struct() -> Result<()> { use arrow::array::{Int32Array, StructArray}; // Per-row sub-list lengths: 2, 1, 0 (empty), 0 (null) @@ -4482,13 +4482,12 @@ async fn unnest_explode_outer_list_of_struct() -> Result<()> { Ok(()) } -/// Spark `explode_outer` semantics applied to a `FixedSizeList` column. -/// For fixed-size lists, every non-null row has the fixed length, so -/// "empty" never occurs — `PreserveAndExpandEmpty` should behave identically -/// to `Preserve` here. The test pins that equivalence so we notice if it -/// ever diverges. +/// Outer-unnest applied to a `FixedSizeList` column. For fixed-size lists, +/// every non-null row has the fixed length, so "empty" never occurs — +/// `PreserveAndExpandEmpty` should behave identically to `Preserve` here. +/// The test pins that equivalence so we notice if it ever diverges. #[tokio::test] -async fn unnest_explode_outer_fixed_size_list() -> Result<()> { +async fn unnest_outer_fixed_size_list() -> Result<()> { let batch = get_fixed_list_batch()?; let ctx = SessionContext::new(); ctx.register_batch("shapes", batch)?; @@ -4502,7 +4501,7 @@ async fn unnest_explode_outer_fixed_size_list() -> Result<()> { )? .collect() .await?; - let explode_outer_results = df + let outer_results = df .unnest_columns_with_options( &["tags"], UnnestOptions::new().with_null_handling( @@ -4513,14 +4512,14 @@ async fn unnest_explode_outer_fixed_size_list() -> Result<()> { .await?; assert_eq!( batches_to_sort_string(&preserve_results), - batches_to_sort_string(&explode_outer_results), + batches_to_sort_string(&outer_results), "FixedSizeList has no empty case, so PreserveAndExpandEmpty must \ match Preserve exactly" ); // And the snapshot itself, to make the expected shape explicit. assert_snapshot!( - batches_to_sort_string(&explode_outer_results), + batches_to_sort_string(&outer_results), @r" +----------+-------+ | shape_id | tags | diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index 2cbe0f49ab3c1..84bf69e26276f 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -414,7 +414,7 @@ message UnnestOptions { // Drop both null and empty lists from the output. DROP = 1; // Preserve nulls, and additionally expand empty lists into a single - // NULL output row (Spark `explode_outer` semantics). + // NULL output row (outer-unnest semantics). PRESERVE_AND_EXPAND_EMPTY = 2; } @@ -633,8 +633,8 @@ message NegativeNode { message Unnest { repeated LogicalExprNode exprs = 1; - // When true, this Unnest expression has Spark `explode_outer` semantics: - // NULL and empty input lists both produce a single NULL output row. + // When true, this Unnest expression has outer-unnest semantics: NULL and + // empty input lists both produce a single NULL output row. bool outer = 2; } diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index 46f8bdceae89b..e6bee31fbf106 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -548,13 +548,13 @@ impl SqlToRel<'_, S> { // Build Unnest expression. // - // `unnest` is DataFusion's native SQL name. `explode` and - // `explode_outer` are accepted aliases — `explode` matches plain - // `unnest` (drops NULL and empty input lists), while `explode_outer` - // sets `outer = true` so the downstream planner picks - // `NullHandling::PreserveAndExpandEmpty`. - if name.eq("unnest") || name.eq("explode") || name.eq("explode_outer") { - let outer = name.eq("explode_outer"); + // `unnest(col)` drops `NULL` and empty input lists (default SQL + // semantics, matching DuckDB/PostgreSQL). `unnest_outer(col)` sets + // `outer = true` so the downstream planner picks + // `NullHandling::PreserveAndExpandEmpty`, which preserves `NULL` + // and empty input lists as a single `NULL` output row. + if name.eq("unnest") || name.eq("unnest_outer") { + let outer = name.eq("unnest_outer"); let mut exprs = self.function_args_to_expr(args, schema, planner_context)?; if exprs.len() != 1 { return plan_err!("{name}() requires exactly one argument"); diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs index 9ede15457f9ad..bdab013144462 100644 --- a/datafusion/sql/src/select.rs +++ b/datafusion/sql/src/select.rs @@ -1490,10 +1490,9 @@ fn collect_unnest_null_handling(expr_groups: &[Vec]) -> Result true)`). -`explode_outer` cannot be mixed with `unnest` or `explode` in the same `SELECT` -— the unnest plan node carries a single null-handling mode for all its output -columns, so a mix would be ambiguous. The planner returns an error in that case. +`unnest_outer` cannot be mixed with `unnest` in the same `SELECT` — the +unnest plan node carries a single null-handling mode for all its output +columns, so a mix would be ambiguous. The planner returns an error in that +case. ## WHERE clause