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/19 13:50:25 UTC
(datafusion) 01/01: fix: preserve functional dependencies across unnest (#25892)
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-25892-fbf77371a0512765ebf81fdc60e4a6cdceb8101f
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b86f1f209cc062832f5cb048915561572632a0a4
Author: limadog9 <ja...@gmail.com>
AuthorDate: Mon Oct 19 13:49:59 2026 +0000
fix: preserve functional dependencies across unnest (#25892)
## Which issue does this PR close?
- Closes #25844
## Rationale for this change
A list UNNEST can produce several rows for one input key. Retaining the
input's functional dependencies makes a later DISTINCT or GROUP BY
discard unnested values, and can make an ordered window use the wrong
frame. These queries return incorrect results.
## What changes are included in this PR?
- Remap inherited dependency indices to output positions after struct
expansion.
- Downgrade dependencies after list unnest, remove unnested list outputs
from dependency targets, and discard dependencies whose determinants are
unnested.
- Add SQL and plan-level regressions for distinct, grouping, window
frames, and shifted indices.
## What is the testing strategy for this PR?
- cargo fmt --all --check
- cargo test -p datafusion-expr --lib
unnest_remaps_functional_dependencies
- cargo test --profile ci --test sqllogictests --
functional_dependencies.slt
## Are there any user-facing changes?
Yes. A list unnest followed by DISTINCT, GROUP BY, or an ordered window
now returns the expected result.
---
datafusion/expr/src/logical_plan/plan.rs | 119 ++++++++++++++++++++-
.../test_files/functional_dependencies.slt | 87 +++++++++++++++
2 files changed, 203 insertions(+), 3 deletions(-)
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 1844966ec6..a2fee99ec2 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -5261,8 +5261,47 @@ impl Unnest {
let metadata = input_schema.metadata().clone();
let df_schema = DFSchema::new_with_metadata(fields, metadata)?;
- // We can use the existing functional dependencies:
- let deps = input_schema.functional_dependencies().clone();
+ let list_indices: HashSet<_> =
+ list_columns.iter().map(|(index, _)| *index).collect();
+ // Struct unnest may replace one input column with several output columns,
+ // so remap dependencies using the input index of each output column.
+ let remap_indices = |indices: &[usize], exclude_lists: bool| {
+ dependency_indices
+ .iter()
+ .enumerate()
+ .filter_map(|(output_index, input_index)| {
+ (indices.contains(input_index)
+ && (!exclude_lists || !list_indices.contains(input_index)))
+ .then_some(output_index)
+ })
+ .collect()
+ };
+ let deps = FunctionalDependencies::new(
+ input_schema
+ .functional_dependencies()
+ .iter()
+ // A list input no longer determines the rows produced by unnest.
+ // A struct input is also unsafe as a source: a null struct and
+ // a non-null struct with all-null fields have the same outputs.
+ .filter(|dep| {
+ dep.source_indices.iter().all(|index| {
+ !list_indices.contains(index) && !struct_columns.contains(index)
+ })
+ })
+ .map(|dep| FunctionalDependence {
+ source_indices: remap_indices(&dep.source_indices, false),
+ // The unnested values are not determined by the input key.
+ target_indices: remap_indices(&dep.target_indices, true),
+ // One input row can produce multiple output rows.
+ mode: if list_indices.is_empty() {
+ dep.mode
+ } else {
+ Dependency::Multi
+ },
+ ..dep.clone()
+ })
+ .collect(),
+ );
let schema = Arc::new(df_schema.with_functional_dependencies(deps)?);
Ok(Unnest {
@@ -5368,7 +5407,7 @@ mod tests {
use datafusion_common::tree_node::{
TransformedResult, TreeNodeRewriter, TreeNodeVisitor,
};
- use datafusion_common::{Constraint, not_impl_err};
+ use datafusion_common::{Constraint, RecursionUnnestOption, not_impl_err};
use insta::{assert_debug_snapshot, assert_snapshot};
use std::hash::DefaultHasher;
@@ -5448,6 +5487,80 @@ mod tests {
])
}
+ #[test]
+ fn unnest_remaps_functional_dependencies() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new_struct(
+ "s",
+ vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ ],
+ true,
+ ),
+ Field::new("k", DataType::Int32, false),
+ Field::new_list(
+ "xs",
+ Field::new_list(
+ "item",
+ Field::new_list_field(DataType::Int32, true),
+ true,
+ ),
+ true,
+ ),
+ Field::new("payload", DataType::Utf8, true),
+ ]));
+ let constraints = Constraints::new_unverified(vec![
+ Constraint::PrimaryKey(vec![1]),
+ Constraint::Unique(vec![0]),
+ Constraint::Unique(vec![2]),
+ ]);
+ let source: Arc<dyn TableSource> =
+ Arc::new(LogicalTableSource::new(schema).with_constraints(constraints));
+
+ // Expanding a struct preserves row count and remaps keys and targets,
+ // but a dependency sourced from the struct itself cannot be kept.
+ let struct_only = LogicalPlanBuilder::scan("t", Arc::clone(&source), None)?
+ .unnest_column("s")?
+ .build()?;
+ assert_eq!(
+ struct_only.schema().functional_dependencies(),
+ &FunctionalDependencies::new(vec![
+ FunctionalDependence::new(vec![2], vec![0, 1, 2, 3, 4], false)
+ .with_mode(Dependency::Single),
+ FunctionalDependence::new(vec![3], vec![0, 1, 2, 3, 4], true)
+ .with_mode(Dependency::Single),
+ ])
+ );
+
+ // The two recursion outputs both replace the same list input. Neither
+ // is determined by k, and a list-sourced dependency is no longer valid.
+ let options = UnnestOptions::default()
+ .with_recursions(RecursionUnnestOption {
+ input_column: "xs".into(),
+ output_column: "xs_depth_1".into(),
+ depth: 1,
+ })
+ .with_recursions(RecursionUnnestOption {
+ input_column: "xs".into(),
+ output_column: "xs_depth_2".into(),
+ depth: 2,
+ });
+ let plan = LogicalPlanBuilder::scan("t", source, None)?
+ .unnest_columns_with_options(vec!["s".into(), "xs".into()], options)?
+ .build()?;
+ assert_eq!(plan.schema().fields().len(), 6);
+ assert_eq!(
+ plan.schema().functional_dependencies(),
+ &FunctionalDependencies::new(vec![
+ FunctionalDependence::new(vec![2], vec![0, 1, 2, 5], false,)
+ .with_mode(Dependency::Multi)
+ ])
+ );
+
+ Ok(())
+ }
+
#[test]
fn projection_with_leading_computed_column_preserves_pk() -> Result<()> {
let constraints =
diff --git a/datafusion/sqllogictest/test_files/functional_dependencies.slt b/datafusion/sqllogictest/test_files/functional_dependencies.slt
index a407b4cd6d..60ba1f00e6 100644
--- a/datafusion/sqllogictest/test_files/functional_dependencies.slt
+++ b/datafusion/sqllogictest/test_files/functional_dependencies.slt
@@ -466,6 +466,93 @@ SELECT y, count(*) FROM (SELECT x, max(y) AS y FROM t_inner GROUP BY x) GROUP BY
statement ok
drop table t_inner;
+##########
+## List unnest invalidates dependencies on the input row
+##########
+
+statement ok
+CREATE TABLE t_unnest_fd (k VARCHAR, v INT) AS VALUES ('a', 1), ('a', 2), ('b', 3);
+
+# DISTINCT must retain the unnested value in its grouping key.
+query I
+WITH g AS (SELECT k, array_agg(v) AS vs FROM t_unnest_fd GROUP BY k),
+ u AS (SELECT k, unnest(vs) AS v FROM g)
+SELECT count(*) AS n FROM (SELECT DISTINCT k, v FROM u);
+----
+3
+
+# The equivalent GROUP BY must also retain the unnested value.
+query I
+WITH g AS (SELECT k, array_agg(v) AS vs FROM t_unnest_fd GROUP BY k),
+ u AS (SELECT k, unnest(vs) AS v FROM g)
+SELECT count(*) AS n FROM (SELECT k, v FROM u GROUP BY k, v);
+----
+3
+
+# There is no projection between Unnest and DISTINCT to discard a stale target.
+query I
+WITH g AS (SELECT k, array_agg(v) AS vs FROM t_unnest_fd GROUP BY k)
+SELECT count(*) AS n FROM (SELECT DISTINCT * FROM (SELECT k, unnest(vs) FROM g));
+----
+3
+
+# An unnested list of structs also creates multiple values per input key.
+query I
+WITH g AS (
+ SELECT k, array_agg(named_struct('id', v)) AS rs
+ FROM t_unnest_fd GROUP BY k
+), u AS (SELECT k, unnest(rs) AS r FROM g)
+SELECT count(*) AS n FROM (SELECT DISTINCT k, r['id'] FROM u);
+----
+3
+
+# Struct expansion shifts the following list output onto an old dependency
+# target index. DISTINCT must still include the list output.
+query I
+WITH g AS (
+ SELECT k, named_struct('x', min(v), 'y', max(v)) AS s,
+ array_agg(v) AS vs, sum(v) AS n
+ FROM t_unnest_fd GROUP BY k
+)
+SELECT count(*) FROM (
+ SELECT DISTINCT * FROM (SELECT k, unnest(s), unnest(vs), n FROM g)
+);
+----
+3
+
+# ORDER BY k has ties after unnest. Its default RANGE frame includes both rows with k = 'a'.
+query I
+WITH g AS (SELECT k, array_agg(v) AS vs FROM t_unnest_fd GROUP BY k),
+ u AS (SELECT k, unnest(vs) AS v FROM g)
+SELECT count(DISTINCT s) AS n FROM (SELECT sum(v) OVER (ORDER BY k) AS s FROM u);
+----
+2
+
+# Expanding a struct before k shifts the inherited dependency's column indices.
+# The two sy values tie, so their default RANGE frames have the same sum.
+query I
+WITH g AS (
+ SELECT named_struct('x', min(v), 'y', length(k)) AS s,
+ k, sum(v) AS n
+ FROM t_unnest_fd GROUP BY k
+), u(sx, sy, k, n) AS (SELECT unnest(s), k, n FROM g)
+SELECT count(DISTINCT running_sum) FROM (
+ SELECT sum(n) OVER (ORDER BY sy) AS running_sum FROM u
+);
+----
+1
+
+# A row-preserving, single-element unnest remains correct.
+query I
+SELECT count(*) AS n
+FROM (SELECT DISTINCT k, v
+ FROM (SELECT k, unnest(make_array(v)) AS v FROM t_unnest_fd));
+----
+3
+
+statement ok
+drop table t_unnest_fd;
+
##########
## Cleanup
##########
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org