Skip to content

Commit 5c846af

Browse files
authored
fix: Streaming aggregation produces wrong result with grouping sets (#24422)
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #24421 ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. Please explain the problem you are trying to solve in terms of the user-visible behavior, rather than the implementation. For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of the implementation. "COUNT(DISTINCT) returns wrong results when the column contains nulls" is the user-visible problem. --> The aggregation planning will enable ordering optimization when the grouping sets is available, however they're not compatible now. I'm also not sure if we want to enable ordering optimization for grouping sets in the future, since this seems requires extra implementation complexity. This PR disables ordering/streaming aggregation optimization when grouping set is present. Reference for ordering optimization in aggregation: https://github.com/apache/datafusion/blob/710e74e6764275675b117823d26d4944115f6ba2/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs#L62 ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here, but it is sometimes worth providing a summary of the individual changes in this PR. --> - 80cb8ee: reproducer that fails without the fix - 5673f3e: fix. It disables ordering optimization when grouping set is present ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> slt ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. If there are any breaking changes to public APIs, please add the `api change` label. --> no
1 parent ebea069 commit 5c846af

3 files changed

Lines changed: 130 additions & 13 deletions

File tree

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2235,7 +2235,7 @@ fn test_pushdown_grouping_sets_filter_on_common_column() {
22352235
- DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
22362236
output:
22372237
Ok:
2238-
- AggregateExec: mode=Final, gby=[(a@0 as a, b@1 as b), (NULL as a, b@1 as b)], aggr=[cnt], ordering_mode=PartiallySorted([1])
2238+
- AggregateExec: mode=Final, gby=[(a@0 as a, b@1 as b), (NULL as a, b@1 as b)], aggr=[cnt]
22392239
- DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=b@1 = bar
22402240
"
22412241
);
@@ -2521,7 +2521,7 @@ fn test_pushdown_through_aggregate_grouping_sets_with_reordered_input() {
25212521
- DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
25222522
output:
25232523
Ok:
2524-
- AggregateExec: mode=Final, gby=[(a@1 as a, b@2 as b), (NULL as a, b@2 as b)], aggr=[cnt], ordering_mode=PartiallySorted([1])
2524+
- AggregateExec: mode=Final, gby=[(a@1 as a, b@2 as b), (NULL as a, b@2 as b)], aggr=[cnt]
25252525
- ProjectionExec: expr=[c@2 as c, a@0 as a, b@1 as b]
25262526
- DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=b@1 = bar
25272527
"

datafusion/physical-plan/src/aggregates/mod.rs

Lines changed: 37 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -170,7 +170,7 @@ use crate::statistics::{ChildStats, StatisticsArgs};
170170
use crate::{ChildrenPropertiesMode, ReplaceChildrenOptions, validate_child_count};
171171
use crate::{
172172
DisplayFormatType, Distribution, ExecutionPlan, InputDistributionRequirements,
173-
InputOrderMode, SendableRecordBatchStream, Statistics,
173+
InputOrderMode, Partitioning, SendableRecordBatchStream, Statistics,
174174
};
175175
use datafusion_common::config::ConfigOptions;
176176
use parking_lot::Mutex;
@@ -1015,7 +1015,7 @@ impl AggregateExec {
10151015
.filter(|idx| group_by.groups.iter().all(|group| !group[*idx]))
10161016
.collect();
10171017

1018-
let input_order_mode = if indices.len() == groupby_exprs.len()
1018+
let mut input_order_mode = if indices.len() == groupby_exprs.len()
10191019
&& !indices.is_empty()
10201020
&& group_by.groups.len() == 1
10211021
{
@@ -1026,19 +1026,29 @@ impl AggregateExec {
10261026
InputOrderMode::Linear
10271027
};
10281028

1029+
// To keep the ordering optimization simple, grouping sets are not supported.
1030+
// See [`OrderedPartialAggregateStream`] for more details about this optimization.
1031+
if group_by.has_grouping_set() {
1032+
input_order_mode = InputOrderMode::Linear;
1033+
}
1034+
10291035
// construct a map from the input expression to the output expression of the Aggregation group by
10301036
let group_expr_mapping =
10311037
ProjectionMapping::try_new(group_by.expr.clone(), &input.schema())?;
10321038

1033-
let cache = Self::compute_properties(
1034-
&input,
1035-
Arc::clone(&schema),
1036-
&group_expr_mapping,
1037-
group_by.is_true_no_grouping(),
1038-
&mode,
1039-
&input_order_mode,
1040-
aggr_expr.as_ref(),
1041-
)?;
1039+
let cache = if group_by.has_grouping_set() {
1040+
Self::compute_grouping_set_properties(&input, Arc::clone(&schema))
1041+
} else {
1042+
Self::compute_properties(
1043+
&input,
1044+
Arc::clone(&schema),
1045+
&group_expr_mapping,
1046+
group_by.is_true_no_grouping(),
1047+
&mode,
1048+
&input_order_mode,
1049+
aggr_expr.as_ref(),
1050+
)?
1051+
};
10421052

10431053
let mut exec = AggregateExec {
10441054
mode,
@@ -1438,6 +1448,22 @@ impl AggregateExec {
14381448
))
14391449
}
14401450

1451+
fn compute_grouping_set_properties(
1452+
input: &Arc<dyn ExecutionPlan>,
1453+
schema: SchemaRef,
1454+
) -> PlanProperties {
1455+
// Grouping-set expansion can replace group keys with nulls and adds a
1456+
// grouping ID, so input properties do not project through unchanged.
1457+
PlanProperties::new(
1458+
EquivalenceProperties::new(schema),
1459+
Partitioning::UnknownPartitioning(
1460+
input.output_partitioning().partition_count(),
1461+
),
1462+
EmissionType::Final,
1463+
input.boundedness(),
1464+
)
1465+
}
1466+
14411467
pub fn input_order_mode(&self) -> &InputOrderMode {
14421468
&self.input_order_mode
14431469
}

datafusion/sqllogictest/test_files/grouping.slt

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,97 @@ NULL A 1 0 2 1
4848
NULL B 1 0 2 1
4949
NULL NULL 1 1 3 3
5050

51+
# Grouping sets must not use the ordered aggregation optimization even when the
52+
# input is ordered by a grouping expression that is present in every set.
53+
statement ok
54+
SET datafusion.execution.target_partitions = 1;
55+
56+
statement ok
57+
SET datafusion.execution.enable_migration_aggregate = false;
58+
59+
statement ok
60+
CREATE EXTERNAL TABLE grouping_set_ordered (
61+
a0 INTEGER,
62+
a INTEGER,
63+
b INTEGER,
64+
c INTEGER,
65+
d INTEGER
66+
)
67+
STORED AS CSV
68+
WITH ORDER (c)
69+
LOCATION '../core/tests/data/window_2.csv'
70+
OPTIONS ('format.has_header' 'true');
71+
72+
# An unordered AggregateExec omits `ordering_mode`; its absence below ensures
73+
# that grouping sets do not select ordered/streaming aggregation.
74+
query TT
75+
EXPLAIN
76+
SELECT b, c, grouping(b, c) AS gid
77+
FROM grouping_set_ordered
78+
GROUP BY GROUPING SETS ((b, c), (c));
79+
----
80+
logical_plan
81+
01)Projection: grouping_set_ordered.b, grouping_set_ordered.c, CAST(__grouping_id & UInt8(3) AS Int32) AS gid
82+
02)--Aggregate: groupBy=[[GROUPING SETS ((grouping_set_ordered.b, grouping_set_ordered.c), (grouping_set_ordered.c))]], aggr=[[]]
83+
03)----TableScan: grouping_set_ordered projection=[b, c]
84+
physical_plan
85+
01)ProjectionExec: expr=[b@0 as b, c@1 as c, CAST(__grouping_id@2 & 3 AS Int32) as gid]
86+
02)--AggregateExec: mode=Final, gby=[b@0 as b, c@1 as c, __grouping_id@2 as __grouping_id], aggr=[]
87+
03)----AggregateExec: mode=Partial, gby=[(b@0 as b, c@1 as c), (NULL as b, c@1 as c)], aggr=[]
88+
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c], output_ordering=[c@1 ASC NULLS LAST], file_type=csv, has_header=true
89+
90+
# The grouping sets produce two rows for each value of c.
91+
query I
92+
SELECT c
93+
FROM (
94+
SELECT b, c, grouping(b, c) AS gid
95+
FROM grouping_set_ordered
96+
GROUP BY GROUPING SETS ((b, c), (c))
97+
)
98+
ORDER BY c
99+
LIMIT 10;
100+
----
101+
0
102+
0
103+
1
104+
1
105+
2
106+
2
107+
3
108+
3
109+
4
110+
4
111+
112+
statement ok
113+
SET datafusion.execution.enable_migration_aggregate = true;
114+
115+
query I
116+
SELECT c
117+
FROM (
118+
SELECT b, c, grouping(b, c) AS gid
119+
FROM grouping_set_ordered
120+
GROUP BY GROUPING SETS ((b, c), (c))
121+
)
122+
ORDER BY c
123+
LIMIT 10;
124+
----
125+
0
126+
0
127+
1
128+
1
129+
2
130+
2
131+
3
132+
3
133+
4
134+
4
135+
136+
statement ok
137+
RESET datafusion.execution.enable_migration_aggregate;
138+
139+
statement ok
140+
SET datafusion.execution.target_partitions = 4;
141+
51142
# grouping_with_cube
52143
query TTIIII
53144
select

0 commit comments

Comments
 (0)