You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/20 17:31:49 UTC
(datafusion) branch gh-readonly-queue/main/pr-26433-39e2304222600d996f0927f0d2428ebd127e9c3a created (now 79650ec0e2)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26433-39e2304222600d996f0927f0d2428ebd127e9c3a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 79650ec0e2 fix: UNNEST now preserves item fields in expression inference and logical-plan schemas (#26433)
This branch includes the following new commits:
new 79650ec0e2 fix: UNNEST now preserves item fields in expression inference and logical-plan schemas (#26433)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: UNNEST now preserves item fields in expression inference and logical-plan schemas (#26433)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26433-39e2304222600d996f0927f0d2428ebd127e9c3a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 79650ec0e27d1f4dc11a1eeda8c084283b7a7f25
Author: Ethan Tran <et...@quwin.dev>
AuthorDate: Tue Oct 20 17:31:25 2026 +0000
fix: UNNEST now preserves item fields in expression inference and logical-plan schemas (#26433)
## Which issue does this PR close?
- Closes #26331.
## Rationale for this change
In SedonaDB, unnesting a list of GeoArrow geometries currently returns
plain binary values, losing the geometry type and CRS metadata needed by
downstream spatial operations; metadata-aware SQL functions can also
reject the field during planning.
## What changes are included in this PR?
- Preserve list-item fields in expression inference and logical-plan
schemas while retaining output naming and nullability behavior.
## What is the testing strategy for this PR?
- DataFrame regressions cover all five list variants, nested unnesting,
aliases, NULL/empty lists, and metadata-sensitive UDF planning. The
targeted regression run passed all 7 cases.
- The `logical_unnest_plus_1000_columns` benchmark in the optimized
`release-nonlto` profile measured 47.58 ms on base `0d79252f` and 48.75
ms on this branch (+2.5% median). This is within the noise threshold.
- The full extended workspace suite and all-features Clippy were not
rerun; the existing Clippy issue #26181 remains noted.
## Are there any user-facing changes?
UNNEST retains list-item metadata, including Arrow extension metadata,
in inferred fields and result schemas. No public Rust API changes.
---------
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
Co-authored-by: Dmitrii Blaginin <dm...@blaginin.me>
---
datafusion/core/tests/dataframe/mod.rs | 204 +++++++++++++++++++++++++++++++
datafusion/expr/src/expr_schema.rs | 23 +++-
datafusion/expr/src/logical_plan/plan.rs | 43 ++-----
datafusion/expr/src/utils.rs | 22 ++++
4 files changed, 259 insertions(+), 33 deletions(-)
diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs
index d011a25273..1a4f6ffcb4 100644
--- a/datafusion/core/tests/dataframe/mod.rs
+++ b/datafusion/core/tests/dataframe/mod.rs
@@ -4483,6 +4483,210 @@ async fn unnest_dict_encoded_columns() -> Result<()> {
Ok(())
}
+#[rstest]
+#[case::list(DataType::List)]
+#[case::large_list(DataType::LargeList)]
+#[case::fixed_size_list(|field| DataType::FixedSizeList(field, 2))]
+#[case::list_view(DataType::ListView)]
+#[case::large_list_view(DataType::LargeListView)]
+#[tokio::test]
+async fn unnest_list_item_metadata(
+ #[case] list_type: fn(Arc<Field>) -> DataType,
+) -> Result<()> {
+ use arrow::array::BinaryArray;
+ use arrow::buffer::OffsetBuffer;
+ use arrow::compute::cast;
+ use datafusion_common::RecursionUnnestOption;
+
+ let item = Arc::new(Field::new("wkb", DataType::Binary, false).with_metadata(
+ HashMap::from([
+ ("ARROW:extension:name".into(), "geoarrow.wkb".into()),
+ (
+ "ARROW:extension:metadata".into(),
+ r#"{"crs":"EPSG:4326"}"#.into(),
+ ),
+ ("custom".into(), "item".into()),
+ ]),
+ ));
+ let values: ArrayRef =
+ Arc::new(BinaryArray::from_iter_values([b"a", b"b", b"c", b"d"]));
+ let lists = ListArray::new(
+ Arc::clone(&item),
+ OffsetBuffer::new(vec![0, 2, 4].into()),
+ Arc::clone(&values),
+ None,
+ );
+ let lists = cast(&lists, &list_type(Arc::clone(&item)))?;
+ let nested_item = Arc::new(
+ Field::new("lists", lists.data_type().clone(), false)
+ .with_metadata(HashMap::from([("custom".into(), "nested item".into())])),
+ );
+ let nested = ListArray::new(
+ Arc::clone(&nested_item),
+ OffsetBuffer::new(vec![0, 2].into()),
+ Arc::clone(&lists),
+ None,
+ );
+ let nested = cast(&nested, &list_type(Arc::clone(&nested_item)))?;
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("input", nested.data_type().clone(), false)
+ .with_metadata(HashMap::from([("custom".into(), "container".into())])),
+ ]));
+ let ctx = SessionContext::new();
+ let batch = RecordBatch::try_new(schema, vec![nested])?;
+ ctx.register_batch("t", batch.clone())?;
+ let df = ctx.read_batch(batch)?;
+
+ // Each recursion depth must use that element's metadata, rather than the
+ // outer list's metadata. Renaming the result must retain it as well.
+ for (depth, field, expected_values) in
+ [(1, &nested_item, &lists), (2, &item, &values)]
+ {
+ let options = UnnestOptions::new().with_recursions(RecursionUnnestOption {
+ input_column: "input".into(),
+ output_column: "expanded".into(),
+ depth,
+ });
+ let unnested = df
+ .clone()
+ .unnest_columns_with_options(&["input"], options)?
+ .select(vec![col("expanded").alias("output")])?;
+ let expected_field = field
+ .as_ref()
+ .clone()
+ .with_name("output")
+ .with_nullable(true);
+ let sql = if depth == 1 {
+ "SELECT unnest(input) AS output FROM t"
+ } else {
+ "SELECT unnest(unnest(input)) AS output FROM t"
+ };
+ for unnested in [unnested, ctx.sql(sql).await?] {
+ assert_eq!(unnested.schema().field(0).as_ref(), &expected_field);
+ let batches = unnested.collect().await?;
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].schema().field(0), &expected_field);
+ assert_eq!(batches[0].column(0), expected_values);
+ }
+ }
+ Ok(())
+}
+
+#[tokio::test]
+async fn unnest_list_item_metadata_in_udf() -> Result<()> {
+ use arrow::buffer::OffsetBuffer;
+ use datafusion_expr::{
+ ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
+ };
+
+ #[derive(Debug, PartialEq, Eq, Hash)]
+ struct RequireItemMetadata(Signature);
+
+ impl ScalarUDFImpl for RequireItemMetadata {
+ fn name(&self) -> &str {
+ "require_item_metadata"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.0
+ }
+
+ fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
+ Ok(arg_types[0].clone())
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<Arc<Field>> {
+ let field = &args.arg_fields[0];
+ if field.metadata().get("custom").map(String::as_str) != Some("item") {
+ return datafusion_common::plan_err!("Missing item metadata");
+ }
+ Ok(Arc::clone(field))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ Ok(args.args[0].clone())
+ }
+ }
+
+ let item = Arc::new(
+ Field::new("item", DataType::Int32, false)
+ .with_metadata(HashMap::from([("custom".into(), "item".into())])),
+ );
+ let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2]));
+ let list = ListArray::new(
+ Arc::clone(&item),
+ OffsetBuffer::new(vec![0, 2].into()),
+ Arc::clone(&values),
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "input",
+ list.data_type().clone(),
+ false,
+ )]));
+ let ctx = SessionContext::new();
+ ctx.register_batch("t", RecordBatch::try_new(schema, vec![Arc::new(list)])?)?;
+ ctx.register_udf(ScalarUDF::from(RequireItemMetadata(Signature::any(
+ 1,
+ Volatility::Immutable,
+ ))));
+ let expected_field = item
+ .as_ref()
+ .clone()
+ .with_name("output")
+ .with_nullable(true);
+
+ // A UDF can inspect argument metadata before SQL rewrites UNNEST into a plan.
+ for sql in [
+ "SELECT require_item_metadata(unnest(input)) AS output FROM t",
+ "SELECT require_item_metadata(output) AS output \
+ FROM (SELECT unnest(input) AS output FROM t)",
+ ] {
+ let df = ctx.sql(sql).await?;
+ assert_eq!(df.schema().field(0).as_ref(), &expected_field);
+ let batches = df.collect().await?;
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].schema().field(0), &expected_field);
+ assert_eq!(batches[0].column(0), &values);
+ }
+ Ok(())
+}
+
+#[tokio::test]
+async fn unnest_list_item_metadata_with_nulls() -> Result<()> {
+ let item = Arc::new(
+ Field::new("item", DataType::Int32, false)
+ .with_metadata(HashMap::from([("custom".into(), "item".into())])),
+ );
+ let mut builder = ListBuilder::new(Int32Builder::new()).with_field(Arc::clone(&item));
+ builder.values().append_slice(&[1, 2]);
+ builder.append(true);
+ builder.append(true); // Empty list
+ builder.append(false); // NULL list
+ let list = builder.finish();
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "input",
+ list.data_type().clone(),
+ true,
+ )]));
+ let ctx = SessionContext::new();
+ let options = UnnestOptions::new()
+ .with_null_handling(datafusion_common::NullHandling::PreserveAndExpandEmpty);
+ let df = ctx
+ .read_batch(RecordBatch::try_new(schema, vec![Arc::new(list)])?)?
+ .unnest_columns_with_options(&["input"], options)?;
+ let expected_field = item.as_ref().clone().with_name("input").with_nullable(true);
+ assert_eq!(df.schema().field(0).as_ref(), &expected_field);
+ let batches = df.collect().await?;
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].schema().field(0), &expected_field);
+ assert_eq!(
+ batches[0].column(0).as_ref(),
+ &Int32Array::from(vec![Some(1), Some(2), None, None]) as &dyn Array
+ );
+ Ok(())
+}
+
#[tokio::test]
async fn unnest_column_nulls() -> Result<()> {
let df = table_with_lists_and_nulls().await?;
diff --git a/datafusion/expr/src/expr_schema.rs b/datafusion/expr/src/expr_schema.rs
index 8aaca2b72b..2804c736c4 100644
--- a/datafusion/expr/src/expr_schema.rs
+++ b/datafusion/expr/src/expr_schema.rs
@@ -517,6 +517,7 @@ impl ExprSchemable for Expr {
/// - **Column references**: Preserve original field metadata from input schema
/// - **Literals**: Use explicitly provided metadata, otherwise empty
/// - **Aliases**: Merge underlying expr metadata with alias-specific metadata, preferring the alias metadata
+ /// - **List unnesting**: Preserve the element field metadata
/// - **Binary expressions**: field metadata is empty
/// - **Boolean expressions**: field metadata is empty
/// - **Cast expressions**: Type-only casts pass through source metadata (stripping extension
@@ -678,6 +679,27 @@ impl ExprSchemable for Expr {
Expr::LambdaVariable(LambdaVariable {
field: Some(field), ..
}) => Ok(Arc::clone(field).renamed(&schema_name)),
+ Expr::Unnest(Unnest { expr, .. }) => {
+ let nullable = self.nullable(schema)?;
+ match expr.get_type(schema)? {
+ DataType::List(field)
+ | DataType::LargeList(field)
+ | DataType::FixedSizeList(field, _)
+ | DataType::ListView(field)
+ | DataType::LargeListView(field) => Ok(Arc::new(
+ field
+ .as_ref()
+ .clone()
+ .with_name(&schema_name)
+ .with_nullable(nullable),
+ )),
+ _ => Ok(Arc::new(Field::new(
+ &schema_name,
+ self.get_type(schema)?,
+ nullable,
+ ))),
+ }
+ }
Expr::Like(_)
| Expr::SimilarTo(_)
| Expr::Not(_)
@@ -689,7 +711,6 @@ impl ExprSchemable for Expr {
| Expr::Wildcard { .. }
| Expr::GroupingSet(_)
| Expr::Placeholder(_)
- | Expr::Unnest(_)
| Expr::Lambda(_)
| Expr::LambdaVariable(_) => Ok(Arc::new(Field::new(
&schema_name,
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index a2fee99ec2..10cc92d6b5 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -44,8 +44,8 @@ use crate::logical_plan::{DmlStatement, Statement, WriteOp};
use crate::utils::{
check_aggregate_and_window_nesting, check_no_window_functions,
enumerate_grouping_sets, expr_to_columns, exprlist_to_fields,
- find_out_reference_exprs, grouping_set_expr_count, grouping_set_to_exprlist,
- merge_schema, passthrough_field_index, split_conjunction,
+ find_out_reference_exprs, get_unnested_list_field, grouping_set_expr_count,
+ grouping_set_to_exprlist, merge_schema, passthrough_field_index, split_conjunction,
};
use crate::{
BinaryExpr, CreateMemoryTable, CreateView, Execute, Expr, ExprSchemable, GroupingSet,
@@ -5338,13 +5338,15 @@ fn get_unnested_columns(
| DataType::LargeList(_)
| DataType::ListView(_)
| DataType::LargeListView(_) => {
- let data_type = get_unnested_list_datatype_recursive(data_type, depth)?;
- let new_field = Arc::new(Field::new(
- col_name, data_type,
- // Unnesting may produce NULLs even if the list is not null.
- // For example: unnest([1], []) -> 1, null
- true,
- ));
+ let field = get_unnested_list_field(data_type, depth)?;
+ let new_field = Arc::new(
+ field
+ .clone()
+ .with_name(col_name)
+ // Unnesting may produce NULLs even if the list is not null.
+ // For example: unnest([1], []) -> 1, null
+ .with_nullable(true),
+ );
let column = Column::from_name(col_name);
// let column = Column::from((None, &new_field));
qualified_columns.push((column, new_field));
@@ -5369,29 +5371,6 @@ fn get_unnested_columns(
Ok(qualified_columns)
}
-// Get the data type of a multi-dimensional type after unnesting it
-// with a given depth
-fn get_unnested_list_datatype_recursive(
- data_type: &DataType,
- depth: usize,
-) -> Result<DataType> {
- match data_type {
- DataType::List(field)
- | DataType::FixedSizeList(field, _)
- | DataType::LargeList(field)
- | DataType::ListView(field)
- | DataType::LargeListView(field) => {
- if depth == 1 {
- return Ok(field.data_type().clone());
- }
- return get_unnested_list_datatype_recursive(field.data_type(), depth - 1);
- }
- _ => {}
- }
-
- internal_err!("trying to unnest on invalid data type {data_type}")
-}
-
#[cfg(test)]
mod tests {
use super::*;
diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs
index 777e835078..8b9d25e2fb 100644
--- a/datafusion/expr/src/utils.rs
+++ b/datafusion/expr/src/utils.rs
@@ -919,6 +919,28 @@ where
err
}
+/// Returns the element field at the given list unnesting depth.
+pub(crate) fn get_unnested_list_field(
+ data_type: &DataType,
+ depth: usize,
+) -> Result<&Field> {
+ match data_type {
+ DataType::List(field)
+ | DataType::FixedSizeList(field, _)
+ | DataType::LargeList(field)
+ | DataType::ListView(field)
+ | DataType::LargeListView(field) => {
+ if depth == 1 {
+ return Ok(field.as_ref());
+ }
+ return get_unnested_list_field(field.data_type(), depth - 1);
+ }
+ _ => {}
+ }
+
+ internal_err!("trying to unnest on invalid data type {data_type}")
+}
+
/// Create schema fields from an expression list, for use in result set schema construction
///
/// This function converts a list of expressions into a list of complete schema fields,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org