NGA-TRAN commented on code in PR #26835:
URL: https://github.com/apache/datafusion/pull/26835#discussion_r4407510542
##########
datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt:
##########
@@ -152,6 +135,68 @@ k2 2024-01-01T00:30:00 7
k2 2024-01-01T01:30:00 30
k2 2024-01-01T01:45:00 5
+##########
+# The 90-minute bin [00:00, 01:30) crosses the scan's 01:00 partition split.
+# Keep the hash shuffle even though the date_bin parameters are admitted.
+##########
+
+query TT
+EXPLAIN SELECT key, date_bin(INTERVAL '90 minutes', timestamp) AS time_bin,
sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY key, time_bin;
+----
+physical_plan
+01)ProjectionExec: expr=[key@0 as key,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 5400000000000 }"),range_sorted_time_bin.timestamp)@1 as time_bin,
sum(range_sorted_time_bin.value)@2 as sum(range_sorted_time_bin.value)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 5400000000000 }"),range_sorted_time_bin.timestamp)@1 as
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 5400000000000 }"),range_sorted_time_bin.timestamp)],
aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
+03)----RepartitionExec: partitioning=Hash([key@0,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 5400000000000 }"),range_sorted_time_bin.timestamp)@1], 2),
input_partitions=2, preserve_order=true, sort_exprs=key@0 ASC,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 5400000000000 }"),range_sorted_time_bin.timestamp)@1 ASC
+04)------AggregateExec: mode=Partial, gby=[key@0 as key,
date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 5400000000000
}, timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano {
months: 0, days: 0, nanoseconds: 5400000000000
}"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)],
group_clustering_mode=Full
+05)--------FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
+06)----------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]},
projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC,
timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC],
[(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a,
pruning_predicate=col4_null_count@2 != row_count@3 AND col4_min@0 <= a AND a <=
col4_max@1, required_guarantees=[col4 in (a)]
+
+query TPI
+SELECT key, date_bin(INTERVAL '90 minutes', timestamp) AS time_bin, sum(value)
+FROM range_sorted_time_bin
+WHERE col4 = 'a'
+GROUP BY key, time_bin
+ORDER BY key, time_bin;
+----
+k1 2024-01-01T00:00:00 33
+k2 2024-01-01T00:00:00 7
+k2 2024-01-01T01:30:00 35
+
+##########
+# An explicit origin remains outside the audited slice, even when it is
+# the epoch and produces the same bins as the two-argument expression.
+##########
Review Comment:
Do do think this comment will be clearer?
```suggestion
# This PR enables shuffle removal only for two-argument date_bin calls.
# A three-argument call therefore retains the shuffle, even when its
# explicit epoch origin produces identical bins and would be safe.
##########
```
Do you plan to support this sometime; or maybe open a ticket for it?
##########
datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt:
##########
@@ -102,11 +90,9 @@ GROUP BY key, time_bin;
----
physical_plan
01)ProjectionExec: expr=[key@0 as key,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as time_bin,
sum(range_sorted_time_bin.value)@2 as sum(range_sorted_time_bin.value)]
-02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)],
aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
-03)----RepartitionExec: partitioning=Hash([key@0,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1], 2),
input_partitions=2, preserve_order=true, sort_exprs=key@0 ASC,
date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0,
nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 ASC
-04)------AggregateExec: mode=Partial, gby=[key@0 as key,
date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 },
timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months:
0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)],
aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
-05)--------FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
-06)----------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]},
projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC,
timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC],
[(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a,
pruning_predicate=col4_null_count@2 != row_count@3 AND col4_min@0 <= a AND a <=
col4_max@1, required_guarantees=[col4 in (a)]
+02)--AggregateExec: mode=SinglePartitioned, gby=[key@0 as key,
date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 },
timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months:
0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)],
aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
+03)----FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
+04)------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]},
projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC,
timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC],
[(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a,
pruning_predicate=col4_null_count@2 != row_count@3 AND col4_min@0 <= a AND a <=
col4_max@1, required_guarantees=[col4 in (a)]
Review Comment:
🎉
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
