Repository navigation
feat(datetime): support date_bin literal origins for range locality - #27135
Conversation
github-actions
Bot
added
physical-expr
|
cc @NGA-TRAN @jayzhan211 Another small extension, thanks! |
NGA-TRAN
left a comment
NGA-TRAN
left a comment
There was a problem hiding this comment.
Looks great. Thanks Ben
| 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, 0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp,Utf8("1970-01-01 00:00:00"))], 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)] |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
@benbellick thanks for the PR. On skewed tables this change makes three-argument date_bin slower: when one file holds 97% of 16M rows and groups are mostly unique, the new SinglePartitioned plan takes 2.3x the shuffle plan's time at target_partitions=4 and 4.1x at 8, because all that grouping runs in one partition on one thread.
This doesn't need to block the PR. It only applies with preserve_file_partitions enabled, sorted files gain up to 5x, and the two-argument call, which this PR doesn't change, shows the same ratio (4.15x). It's fine to handle in a follow-up: could you open an issue for taking row-count skew into account when skipping the shuffle?
New plan time ÷ shuffle plan time (16M rows, 4 files, 9 rounds, medians; below 1 means the new plan is faster)
| Input | 3-arg, target_partitions=8 |
3-arg, target_partitions=4 |
2-arg, target_partitions=8 |
|---|---|---|---|
| Sorted, mostly unique groups | 0.21 | 0.29 | 0.20 |
| Sorted, low cardinality | 0.60 | 0.87 | 0.51 |
| Balanced, low cardinality | 1.37 | 0.95 | 1.46 |
| Balanced, mostly unique groups | 1.10 | 0.93 | 1.13 |
| 97% skew, low cardinality | 3.07 | 1.03 | 3.03 |
| 97% skew, mostly unique groups | 4.11 | 2.35 | 4.15 |
Same binary for both plans, using a wrapper UDF that keeps the default-deny locality hook, on a busy laptop. With round-robin repartitioning off at target_partitions=8, balanced low-cardinality is 1.00, low-cardinality skew is 1.04 and mostly-unique skew is 2.29. At default config the plan doesn't change.
|
Thanks everyone |
|
Thanks for benchmarking this @jayzhan211! Opened #27268 to track the follow-up. |
Which issue does this PR close?
Closes #26913.
Rationale for this change
An explicit origin currently prevents
date_binfrom reusing range partitioning for aggregation, even when its buckets align with every range split.What changes are included in this PR?
Admit non-null literal origins while retaining timezone-less nanosecond timestamps and positive fixed-duration, month-free strides. A fixed origin shifts the buckets without changing their nondecreasing behavior. The existing boundary proof still rejects crossing buckets and evaluation errors.
What is the testing strategy for this PR?
SQLLogicTests check plans and results for explicit epoch, shifted-aligned, and crossing origins. Unit tests cover rejected arguments, ascending/descending splits, and fail-closed boundary errors.
Are there any user-facing changes?
More aligned
date_binaggregations can avoid a hash shuffle. No function-result or public API changes.Note: This PR was developed with AI assistance. All changes have been reviewed, and I take full responsibility for this contribution.