jayzhan211 commented on PR #24501:
URL: https://github.com/apache/datafusion/pull/24501#issuecomment-5462446066
**`project` applies the monotonic fallback to non-final range keys, where
the disjointness check does not hold**
`f(split) != f(split-1)` preserves the partition boundary only when the
transformed key is the last range key. For an earlier key, rows above the split
can collapse onto `f(split)` and are then ordered by the remaining keys, which
puts them on the wrong side of the projected split point.
Running this against the branch: projecting `Range([timestamp, key],
[(2024-01-01T01:00, 'k5')])` through `SELECT date_trunc('hour', timestamp) AS
h, key` yields `Range([h@0 ASC, key@1 ASC], [(1704070800000000000, k5)], 2)`.
Actual placement is `(01:00:00, 'k1')` in partition 0 and `(01:30:00, 'k1')`
in partition 1. Both project to `(h = 01:00, key = 'k1')`, so the advertised
split point describes no real boundary. `key_satisfaction` then returns `Exact`
for `GROUP BY h, key`, the aggregate runs `SinglePartitioned`, and that group
is emitted twice.
`range_monotonic_fn_satisfaction` already guards exactly this case with
`range.ordering().len() != 1` (the `#24644` comment). `project` needs the
equivalent restriction:
```diff
) -> Option<Self> {
let mut split_points = self.split_points.clone();
let mut sort_exprs = Vec::with_capacity(self.ordering.len());
+ let last_key_idx = self.ordering.len().checked_sub(1)?;
for (key_idx, sort_expr) in self.ordering.iter().enumerate() {
if let Some(projected) =
input_eq_properties.project_expr(&sort_expr.expr, mapping)
{
sort_exprs.push(PhysicalSortExpr::new(projected,
sort_expr.options));
continue;
}
+ // A non-injective transform preserves the partition boundary
+ // only for the final range key. For an earlier key, rows above
+ // the split can collapse onto `f(split)` and are then ordered
+ // by the trailing keys, crossing the projected boundary.
+ if key_idx != last_key_idx {
+ return None;
+ }
+
let (target, source) =
monotonic_range_key_projection(sort_expr, mapping,
input_eq_properties)
```
A regression test mirroring
`test_range_partitioning_project_compound_through_date_bin`, but with the
transform on key 0 instead of key 1, would pin this.
**`Exact` is reported with split points left in the untransformed domain,
which misleads the co-partitioning check**
`range_monotonic_fn_satisfaction` returns `Exact`, but unlike `project` it
does not translate the split points. `Partitioning::Range` keeps advertising
split `s`, while the effective boundary in required-key space is `f(s)`.
That matters because `compatible_co_partitioning_layout` compares raw split
points:
```rust
// datafusion/physical-plan/src/distribution_requirements.rs:348
(Partitioning::Range(left), Partitioning::Range(right)) => {
left.split_points() == right.split_points() && ...
```
and `co_partitioning_satisfied` calls `satisfaction(..., allow_subset =
false)`, which is the path that reaches `Exact` here.
`check_monotonic_transform` accepts any expression reporting `Ordered`, not
only `date_bin` and `date_trunc`. Consider `t1 JOIN t2 ON t1.ts + INTERVAL '1
hour' = t2.ts`, both range-partitioned on `ts` with identical split points
`[s]`:
- `t1` is `Exact`: the shift is strictly monotone, so the disjointness check
always passes
- `t2` is `Exact` via `key_satisfaction`
- split points compare equal, so the two are treated as co-partitioned and
no repartition is inserted
But `t1` partition 0 holds join keys `< s + 1h` while `t2` partition 0 holds
`< s`. A `t2` row at `s + 30min` sits in partition 1 while its `t1` match
(`t1.ts = s - 30min`) sits in partition 0, so the match is silently lost.
For `date_bin` and `date_trunc` this cannot happen: disjointness at `s`
implies `s` starts its bin, so `f(s) == s`. Making that implicit precondition
explicit closes the gap, and also makes `project_split_points_through_fn` a
no-op:
```diff
let Some(at_split) = evaluate_expr_on_key(fn_expr, range_key,
split_value) else {
return false;
};
+ // Require the split to be a fixed point of `f`, which holds for
+ // flooring transforms such as date_bin and date_trunc. Otherwise
+ // the boundary in the transformed domain is `f(split)`, not
+ // `split`, and callers that compare raw split points are wrong.
+ if at_split != *split_value {
+ return false;
+ }
let Some(below_split) = evaluate_expr_on_key(fn_expr, range_key,
&predecessor)
else {
return false;
};
at_split != below_split
```
Every test in the PR still passes with this guard, since both the 60s
`date_bin` split and the `date_trunc('hour')` split are fixed points. If you
want to keep support for general monotone transforms such as `x * 2`, the
alternative is to teach `compatible_co_partitioning_layout` about the
transformed boundary, but that is a larger change. Either way, a test with two
`Range` inputs joined on a shifted key would pin the behaviour.
--
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]
