goutamadwant commented on PR #24766:
URL: https://github.com/apache/datafusion/pull/24766#issuecomment-5658940972
> Thanks @goutamadwant, there is a suggestion:
>
> `RangePartitioning::PartialEq` compares only `ordering` + effective
`split_points`, but `samples` is not derivable from `split_points` and is
exactly the state `scale()` reads. So `a == b` does not imply `a.scale(k) ==
b.scale(k)`, and doesn't even imply both succeed — your own test asserts
`try_new_with_samples(ord, [10..90], 4) == try_new(ord, [30,50,70])`, yet the
first scales to 10 and the second errors at 4.
>
> There's already a consumer that turns this into wrong results.
`can_interleave` gates on `partition == *reference`:
>
> ```rust
> // datafusion/physical-plan/src/union.rs:931
> matches!(reference, Partitioning::Hash(_, _) | Partitioning::Range(_))
> && inputs
> .map(|plan| plan.borrow().output_partitioning().clone())
> .all(|partition| partition == *reference)
> ```
>
> and `InterleaveExec::compute_properties` then adopts one input's metadata
for the whole output:
>
> ```rust
> // datafusion/physical-plan/src/union.rs:698
> let output_partitioning = inputs[0].output_partitioning().clone();
> ```
>
> Concretely: input A = `try_new_with_samples(ord, [10,20,...,90], 4)`,
input B = `try_new(ord, [30,50,70])`. They compare equal, interleaving is
allowed (correct — the effective boundaries do match), and the `InterleaveExec`
output now advertises `max_partition_count() == 10`. A distributed planner —
the use case this PR is for — calls `scale(10)` on that output and gets
`[10,20,...,90]`, boundaries B's rows were never placed against. Because range
partitioning is a declared, unvalidated property, that's silently wrong rows
per partition, not an error.
>
> Suggest making equality structural:
>
> ```diff
> -#[derive(Debug, Clone)]
> +#[derive(Debug, Clone, PartialEq)]
> pub struct RangePartitioning {
> ordering: LexOrdering,
> samples: Arc<[SplitPoint]>,
> split_points: Arc<[SplitPoint]>,
> partition_count: usize,
> }
> @@
> -impl PartialEq for RangePartitioning {
> - fn eq(&self, other: &Self) -> bool {
> - self.ordering == other.ordering && self.split_points ==
other.split_points
> - }
> -}
> ```
>
> The one place that genuinely wants effective-boundary comparison already
spells it out and is unaffected:
>
> ```rust
> // datafusion/physical-plan/src/distribution_requirements.rs:348
> (Partitioning::Range(left), Partitioning::Range(right)) => {
> left.split_points() == right.split_points()
> && ...
> }
> ```
>
> so co-partitioned joins keep the permissive behavior; only
`can_interleave` gets stricter, which is the conservative direction.
`test_range_partitioning_equality_uses_effective_split_points` would then need
to flip to `assert_ne!`.
>
> If you'd rather keep permissive equality, the alternative is for
`InterleaveExec::compute_properties` to reduce the output to the coarsest
sample set common to all inputs rather than inheriting `inputs[0]`'s — but
that's more machinery for the same guarantee.
@jayzhan211 Switched RangePartitioning to structural equality and added an
interleave regression for matching effective boundaries with different retained
samples. Both input orders are covered, so interleave cannot inherit
unsupported scaling capacity from either input.
--
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]
