jayshrivastava commented on PR #26593:
URL: https://github.com/apache/datafusion/pull/26593#issuecomment-6697040472
@mbutrovich I've thought about partitioning satisfaction and layouts a
little bit. I this is the main issue to address before merging.
### Background
- `symmetric_join_output_partitioning` sets the output partitioning of the
join based on the join type only (does not care about split points,
distribution etc.)
- `EnforceDistribution` optimizer rule
- The join declares `input_distribution_requirements`, so we assert the
inputs are key partitioned. However, we don't check split points, bounds, or
partition count. We insert a `RepartitionExec` if this failes
- `enforce_distribution_relationships` check that all join inputs are
co-partitioned
- `adapt` - Imo this should be called `adapt_schema` and modify the
schema temporarily to check co-partitioned ness. I have reverted the previous
change were we strip the bounds.
- `has_same_layout` - This is the main thing that checks partition
count, split points etc. If the layout is not the same, we insert a
`RepartitionExec`
So the question is - now that there's bounds, are there soundness /
correctness issues?
### Join Example
For the most part, everything in vanilla datafusion is correct. The
`symmetric_join_output_partitioning` does the heavy lifting, choosing the right
bounds as needed.
The only suspicious case is this:
```
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0)]
RepartitionExec: partitioning=Range([a@0 ASC], [(10), (20), (30)], 4),
input_partitions=1
DataSourceExec: partitions=1, partition_sizes=[1]
DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
Left partitioning: Range([a@0 ASC], [(10), (20), (30)], 4)
Right partitioning: Range([a@0 ASC], [(10), (20), (30)], 4, start=(0),
end=(35))
Output partitioning: Range([a@1 ASC], [(10), (20), (30)], 4, start=(0),
end=(35))
```
We repartition **without bounds** into a join that **has bounds**. In this
scenario, because the operator is a join, **it's expected that the join drops
rows outside the bounds**. The key thing to note is bounds are not considered
by the optimizer rule
- `input_distribution_requirements` enforces distributions, which are not
aware of partition counts, split points, nor bounds
- `has_same_layout` - considers everything but bounds
The `ExecutionPlan` must truncate to the bounds.
## Custom Plan
Say I create a custom plan which enforces an input distribution, I can
introduce an issue where we emit rows outside the bounds:
```
Output partitioning: Range([a@0 ASC], [(20)], 2, start=(10), end=(30))
Input partitioning: Hash([a@0], 2)
TestProjectionExec // configures the `input_distribution_requirements` to be
key partitioned on `a`
RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
DataSourceExec: partitions=2, partition_sizes=[1, 1]
```
This emits rows outside of [10, 30]. This is a real problem.
## Ideas
My two best ideas are as follows:
1. Leave this as is, defer to the operator to handle bounds. If an operator
declares `Partitioning::Range` with bounds and its input doesn't have bounds,
the operator should handle this. This is reasonable because the plan node has
to **opt in**. There wouldn't be a correctness issue without that. This is
currently true for all joins and works fine.
2. We could make `has_same_layout` consider bounds, but for joins, the
operator handles rows outside the bounds already. Updating `has_same_layout` to
return false when the input has no bounds and the output is bounded would
insert repartitions which are unnecessary. For example here, we don't need to
repartition the left
```
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0)]
DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
Left partitioning: Range([a@0 ASC], [(10), (20), (30)], 4)
Right partitioning: Range([a@0 ASC], [(10), (20), (30)], 4, start=(0),
end=(35))
Output partitioning: Range([a@1 ASC], [(10), (20), (30)], 4, start=(0),
end=(35))
```
If I were to make `has_same_layout` return false on a bounds mismatch, I
would add a `FilterExec` like so just to be safe in lieu of a repartition (an
in some cases you would need both a repartition and filter).
```
FilterExec: 0 <= a AND a < 30
```
I wouldn't mind putting that in this PR if it makes sense.
3. This one is related to https://github.com/apache/datafusion/issues/25483
and is a bigger change. I would say we can do this whenever work on
https://github.com/apache/datafusion/issues/25483 begins and not put it in this
PR.
The solution there would be to probably add partitions to
`input_distribution_requirements` (or `RangeLayoutRelationshipKind` as the
issue proposes).
This change feels like it may be overdue. I'm a new contributor, but I feel
like the distributions were added for things like aggregations, where you only
care that there's a distinct value in each partition but you don't care about
anything else. We're kind of piggybacking on that mechanism for strong
partitioning guarantees, like for co-partitioned joins where the actual set of
values in each partition matters.
I'm going to push a commit with some tests which show the above examples.
Please let me know what you think!
--
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]
