Skip to content

fix: fix index bug and add test to check it #7124

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alamb merged 3 commits into from
Jul 28, 2023
Merged

fix: fix index bug and add test to check it #7124

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2174a51
Fix bug, add test
mustafasrepo Jul 28, 2023
62b4faa
Add new test
mustafasrepo Jul 28, 2023
e4b4ede
Merge branch 'apache_main' into bug_fix/first_last_multi_partition
mustafasrepo Jul 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions datafusion/core/tests/sqllogictests/test_files/groupby.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3348,3 +3348,71 @@ ORDER BY l.sn
2 75 3
3 200 4
4 100 5

# create a table
statement ok
CREATE TABLE FOO (x int, y int) AS VALUES (1, 2), (2, 3), (1, 3);

# make sure that query runs in multi partitions
statement ok
set datafusion.execution.target_partitions = 8;

query I
SELECT LAST_VALUE(x)
FROM FOO;
----
1

query II
SELECT x, LAST_VALUE(x)
FROM FOO
GROUP BY x
ORDER BY x;
----
1 1
2 2

query II
SELECT y, LAST_VALUE(x)
FROM FOO
GROUP BY y
ORDER BY y;
----
2 1
3 1

# plan of the query above should contain partial
# and final aggregation stages
query TT
EXPLAIN SELECT LAST_VALUE(x)
FROM FOO;
----
logical_plan
Aggregate: groupBy=[[]], aggr=[[LAST_VALUE(foo.x)]]
--TableScan: foo projection=[x]
physical_plan
AggregateExec: mode=Final, gby=[], aggr=[LAST_VALUE(foo.x)]
--CoalescePartitionsExec
----AggregateExec: mode=Partial, gby=[], aggr=[LAST_VALUE(foo.x)]
------MemoryExec: partitions=8, partition_sizes=[1, 0, 0, 0, 0, 0, 0, 0]

query I
SELECT FIRST_VALUE(x)
FROM FOO;
----
1

# similarly plan of the above query should
# contain partial and final aggregation stages.
query TT
EXPLAIN SELECT FIRST_VALUE(x)
FROM FOO;
----
logical_plan
Aggregate: groupBy=[[]], aggr=[[FIRST_VALUE(foo.x)]]
--TableScan: foo projection=[x]
physical_plan
AggregateExec: mode=Final, gby=[], aggr=[FIRST_VALUE(foo.x)]
--CoalescePartitionsExec
----AggregateExec: mode=Partial, gby=[], aggr=[FIRST_VALUE(foo.x)]
------MemoryExec: partitions=8, partition_sizes=[1, 0, 0, 0, 0, 0, 0, 0]
4 changes: 2 additions & 2 deletions datafusion/physical-expr/src/aggregate/first_last.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ impl Accumulator for FirstValueAccumulator {
let is_set_flags = &states[last_idx];
let flags = is_set_flags.as_boolean();
let mut filtered_first_vals = vec![];
for state in states.iter().take(last_idx - 1) {
for state in states.iter().take(last_idx) {
filtered_first_vals.push(compute::filter(state, flags)?)
}
self.update_batch(&filtered_first_vals)
Expand Down Expand Up @@ -387,7 +387,7 @@ impl Accumulator for LastValueAccumulator {
let is_set_flags = &states[last_idx];
let flags = is_set_flags.as_boolean();
let mut filtered_first_vals = vec![];
for state in states.iter().take(last_idx - 1) {
for state in states.iter().take(last_idx) {
filtered_first_vals.push(compute::filter(state, flags)?)
}
self.update_batch(&filtered_first_vals)
Expand Down