jayshrivastava commented on PR #26593:
URL: https://github.com/apache/datafusion/pull/26593#issuecomment-6609871719
> Should we settle the alternative @stuhood raised in
https://github.com/datafusion-contrib/datafusion-distributed/issues/628#issuecomment-6346697798
before bounds become part of the public range contract? There, every task
keeps the full Range(N, split_points) and the partitions outside its slice are
empty, which needs no new fields on RangePartitioning.
Stu's approach is valid. I also have a similar one in
https://github.com/datafusion-contrib/datafusion-distributed/pull/782. I
decided to publish this PR because I felt that those two approaches diverge
from our goal of keep DFD as close to DF as possible and as thin of a layer
over the single node implementation. Introducing these bounds reduces the
complexity of datafusion distributed very significantly, and the lines of code
as well.
Ballista opted to implement range shuffles without introducing bounds to DF
and you can see the tradeoffs that needed to be made.
Consider that we want to distributed this across two tasks
```
Global: Range(4, splits=[10, 20, 30])
p0 p1 p2 p3
[-inf, 10) [10, 20) [20, 30) [30, +inf)
\___________________/ \____________________/
Task 0 Task 1
```
In ballista, the `RangeShuffleReaderExec` reports `UnknownPartitioning` (see
[reader
properties](https://github.com/apache/datafusion-ballista/blob/96b5ba93463955c91baae703373961a876b30896/ballista/core/src/execution_plans/range_shuffle_reader.rs#L130-L151),
[task
slicing](https://github.com/apache/datafusion-ballista/blob/96b5ba93463955c91baae703373961a876b30896/ballista/scheduler/src/state/task_builder.rs#L345-L378))
```
Range shuffle
/ \
v v
Task 0 Task 1
RangeShuffleReader RangeShuffleReader
Actual ranges:
[-inf,10) [10,20) [20,30) [30,+inf)
DataFusion property:
UnknownPartitioning(2) UnknownPartitioning(2)
```
The `UnknownPartitioning` makes that within a task, we don't have the range
partitioning property anymore. Now, a custom plan node won't know its
partitioned and may not be able to make certain optimizations.
Your suggestion is to have each task advertise all partitions but mark some
as empty like so
```
p0 p1 p2 p3
Task 0: data data empty empty
Task 1: empty empty data data
Both advertise Range(4, splits=[10,20,30]).
```
That is a reasonable alternative, but those empty partitions have an extra
cost, the `RepartitionExec` spawns a reader task and a completion task for
every input partition.
> Spark describes the same kind of slice by index: a coalesced AQE read
gives each reader a span of reducer partitions and keeps the original
RangePartitioning with a smaller count
([AQEShuffleReadExec](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEShuffleReadExec.scala#L93-L94)).
On Spark, I think the distinction is that its physical `RangePartitioning`
describes ordering and a partition count; it does not contain concrete split
values. You can prove that two tasks have disjoint partitions / ranges because
the indexes are disjoint. Ie. one task gets `0,1` and another gets `2,3`. In
datafusion, you can't prove that partitions are disjoint without bounds.
> As I read your reply in
https://github.com/datafusion-contrib/datafusion-distributed/issues/628#issuecomment-6399210844
and the padding thread here, this PR still pads tasks with empty partitions,
so what the bounds save is the N - N/W idle partitions on each task. In
exchange, every range rule has to account for bounds. For example, equality now
includes them, so
[can_interleave](https://github.com/apache/datafusion/blob/67ad9002074c58ed45962c7adcd67c021feaa2d7/datafusion/physical-plan/src/union.rs#L964-L976)
rejects inputs that differ only in bounds, and the layout relationships in
https://github.com/apache/datafusion/issues/25483 will need to define how
bounds compose. Is the idle-partition overhead the remaining cost of that
approach, or is there a plan shape it can't express?
Regarding idle partitions, the user has full control here. They can choose
to not have them. This PR only introduces the **option** to pad.
On `can_interleave`, I would say the current implementation is correct. We
would not interleave these:
```
Input A: [0,10) [10,20) [20,30) [30,35)
Input B: [0,10) [10,20) [20,30) [30,40)
```
However, we may want to interleave them and use this layout.
```
Output: [0,10) [10,20) [20,30) [30,40)
```
Because this PR is a bit large already, I can file a follow up issue for
this. Because bounds are new, I doubt many users will run into this. When
someone does, that optimization can be implemented fairly easily.
--
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]
