From 30a8284a65def2c28d0169dcffff2437e3c25dc6 Mon Sep 17 00:00:00 2001 From: Simon Vandel Sillesen Date: Tue, 14 Jul 2026 08:33:54 +0200 Subject: [PATCH 1/2] test: show DISTINCT wrongly removed when unique key is downgraded by a join `ReplaceDistinctWithAggregate` removes `Distinct::All` when a functional dependence covers all input fields, but it ignores the dependency mode: after a join a formerly-unique key is downgraded to `Dependency::Multi`, so the input may still contain duplicates. This SLT test records the current wrong behavior: `SELECT DISTINCT u.id` over a LEFT JOIN returns one row per matching order instead of one row per user. --- .../sqllogictest/test_files/group_by.slt | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt index 7b0d8a00d55ce..971bbdf99b838 100644 --- a/datafusion/sqllogictest/test_files/group_by.slt +++ b/datafusion/sqllogictest/test_files/group_by.slt @@ -5641,3 +5641,53 @@ set datafusion.execution.target_partitions = 4; statement count 0 drop table t; + +# DISTINCT must not be removed when a unique key is downgraded to a +# non-unique functional dependency by a join: `u.id` is a primary key, but +# after the LEFT JOIN each `u` row can occur once per matching order. +statement ok +CREATE TABLE users_with_pk (id INT, name VARCHAR, primary key(id)) AS VALUES + (1, 'alice'), + (2, 'bob'); + +statement ok +CREATE TABLE user_orders (user_id INT, amount INT) AS VALUES + (1, 10), + (1, 20), + (2, 30); + +# BUG: the DISTINCT is dropped by ReplaceDistinctWithAggregate, so `id` is +# returned once per matching order instead of once per user. +query I +SELECT DISTINCT u.id + FROM users_with_pk u + LEFT JOIN user_orders o ON u.id = o.user_id + ORDER BY u.id; +---- +1 +1 +2 + +# BUG: no Aggregate (or any other deduplication) appears in the plan. +query TT +EXPLAIN SELECT DISTINCT u.id + FROM users_with_pk u + LEFT JOIN user_orders o ON u.id = o.user_id; +---- +logical_plan +01)Projection: u.id +02)--Left Join: u.id = o.user_id +03)----SubqueryAlias: u +04)------TableScan: users_with_pk projection=[id] +05)----SubqueryAlias: o +06)------TableScan: user_orders projection=[user_id] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[id@0] +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +drop table users_with_pk; + +statement ok +drop table user_orders; From 82ff3a669ba0174ec1adbf1553a30e0b25d7200b Mon Sep 17 00:00:00 2001 From: Simon Vandel Sillesen Date: Tue, 14 Jul 2026 08:36:09 +0200 Subject: [PATCH 2/2] fix: do not remove DISTINCT when a unique key was downgraded by a join `ReplaceDistinctWithAggregate` removes `Distinct::All` when a functional dependence's determinant columns cover all input fields, treating the input as already deduplicated. However it ignored the dependency mode: after a join, `FunctionalDependencies::join` downgrades dependencies to `Dependency::Multi` because a formerly-unique key may now occur in many rows. Removing the DISTINCT in that case returns duplicate rows. For example, with `users.id` as a primary key and several orders per user: SELECT DISTINCT users.id FROM users LEFT JOIN orders ON users.id = orders.user_id returned `id` values repeated once per matching order. Only remove the DISTINCT when the dependency mode is `Single`, i.e. the determinant key is guaranteed to occur at most once. The sqllogictest added in the previous commit now shows the correct deduplicated result and an Aggregate in the plan. --- .../common/src/functional_dependencies.rs | 6 ++-- .../src/replace_distinct_aggregate.rs | 13 ++++++--- .../sqllogictest/test_files/group_by.slt | 29 ++++++++++--------- 3 files changed, 29 insertions(+), 19 deletions(-) diff --git a/datafusion/common/src/functional_dependencies.rs b/datafusion/common/src/functional_dependencies.rs index 24ca33c0c2c90..8b15c49c565f1 100644 --- a/datafusion/common/src/functional_dependencies.rs +++ b/datafusion/common/src/functional_dependencies.rs @@ -151,8 +151,10 @@ pub struct FunctionalDependence { /// Describes functional dependency mode. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Dependency { - Single, // A determinant key may occur only once. - Multi, // A determinant key may occur multiple times (in multiple rows). + /// A determinant key may occur only once. + Single, + /// A determinant key may occur multiple times (in multiple rows). + Multi, } impl FunctionalDependence { diff --git a/datafusion/optimizer/src/replace_distinct_aggregate.rs b/datafusion/optimizer/src/replace_distinct_aggregate.rs index 06df61e766615..cc2616379057a 100644 --- a/datafusion/optimizer/src/replace_distinct_aggregate.rs +++ b/datafusion/optimizer/src/replace_distinct_aggregate.rs @@ -22,7 +22,7 @@ use crate::{OptimizerConfig, OptimizerRule}; use std::sync::Arc; use datafusion_common::tree_node::Transformed; -use datafusion_common::{Column, Result}; +use datafusion_common::{Column, Dependency, Result}; use datafusion_expr::expr_rewriter::normalize_cols; use datafusion_expr::utils::expand_wildcard; use datafusion_expr::{Aggregate, Distinct, DistinctOn, Expr, LogicalPlan}; @@ -101,9 +101,14 @@ impl OptimizerRule for ReplaceDistinctWithAggregate { let field_count = input.schema().fields().len(); for dep in input.schema().functional_dependencies().iter() { - // If distinct is exactly the same with a previous GROUP BY, we can - // simply remove it: - if dep.source_indices.len() >= field_count + // If the input is already unique on all of its columns (e.g. + // it is a GROUP BY over exactly these columns), the DISTINCT + // is a no-op and we can simply remove it. The dependency mode + // must be `Single`: a `Multi` dependence (e.g. a former key + // downgraded by a join) means equal rows may occur multiple + // times, so the DISTINCT still has work to do. + if dep.mode == Dependency::Single + && dep.source_indices.len() >= field_count && dep.source_indices[..field_count] .iter() .enumerate() diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt index 971bbdf99b838..de493d6e4a2b1 100644 --- a/datafusion/sqllogictest/test_files/group_by.slt +++ b/datafusion/sqllogictest/test_files/group_by.slt @@ -5656,8 +5656,6 @@ CREATE TABLE user_orders (user_id INT, amount INT) AS VALUES (1, 20), (2, 30); -# BUG: the DISTINCT is dropped by ReplaceDistinctWithAggregate, so `id` is -# returned once per matching order instead of once per user. query I SELECT DISTINCT u.id FROM users_with_pk u @@ -5665,26 +5663,31 @@ SELECT DISTINCT u.id ORDER BY u.id; ---- 1 -1 2 -# BUG: no Aggregate (or any other deduplication) appears in the plan. +# The DISTINCT must be planned as an Aggregate; it cannot be removed based +# on the (join-downgraded) primary key of `users_with_pk`. query TT EXPLAIN SELECT DISTINCT u.id FROM users_with_pk u LEFT JOIN user_orders o ON u.id = o.user_id; ---- logical_plan -01)Projection: u.id -02)--Left Join: u.id = o.user_id -03)----SubqueryAlias: u -04)------TableScan: users_with_pk projection=[id] -05)----SubqueryAlias: o -06)------TableScan: user_orders projection=[user_id] +01)Aggregate: groupBy=[[u.id]], aggr=[[]] +02)--Projection: u.id +03)----Left Join: u.id = o.user_id +04)------SubqueryAlias: u +05)--------TableScan: users_with_pk projection=[id] +06)------SubqueryAlias: o +07)--------TableScan: user_orders projection=[user_id] physical_plan -01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[id@0] -02)--DataSourceExec: partitions=1, partition_sizes=[1] -03)--DataSourceExec: partitions=1, partition_sizes=[1] +01)AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[] +02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=4 +03)----AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[] +04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +05)--------HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, user_id@0)], projection=[id@0] +06)----------DataSourceExec: partitions=1, partition_sizes=[1] +07)----------DataSourceExec: partitions=1, partition_sizes=[1] statement ok drop table users_with_pk;