mbutrovich commented on PR #26593: URL: https://github.com/apache/datafusion/pull/26593#issuecomment-6631827358
> 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. Thanks, that tells me the difference is run-time cost and DFD code, not a plan the alternative can't express. Since bounds are permanent once they ship, could we measure that cost before settling on them? Each task pays for `N - N/W` empty partitions: two tasks spawned per input partition in [`RepartitionExec`](https://github.com/apache/datafusion/blob/32bd81fb226ccf1ae5df67ff340d77baa3e73668/datafusion/physical-plan/src/repartition/mod.rs#L607-L621), plus an empty stream through each operator above the scan. That grows as `N` per task and `N * W` across a stage, so a sweep up to the largest shapes DFD expects would show what bounds save. For example, 16 partitions per task on 100 tasks is `Range(1600)` with 1584 empty partitions on each task. @stuhood also asked in https://github.com/datafusion-contrib/datafusion-distributed/issues/628#issuecomment-6377085055 whether a way to declare empty partitions for every `Partitioning` would remove that cost for `Hash` too. I don't see that answered yet, and it would avoid a `Range`-only field. > 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 it, the same index argument works in DataFusion under the empty-partition alternative. Every task advertises the same `Range(4, [10, 20, 30])`, so partition `k` covers the same keys on every task, and a task that owns `0, 1` has rows only in those partitions. The overlap in the PR description comes from giving each task its own `Range(2, [10])` or `Range(2, [30])`, which drops the global indexes. I think the more useful comparison with Spark is where it keeps the slice. Its [`RangePartitioning`](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala#L1392-L1404) carries only an ordering and a count. The sampled bounds stay in the exchange's [`RangePartitioner`](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala#L397), and Spark never treats two range layouts as co-partitioned because the bounds are sampled ([`RangeShuffleSpec`](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala#L1919-L1935)). A coalesced read gives each reader a contiguous span of reducer indexes and keeps `RangePartitioning` with a smaller count ([`AQEShuffleRe adExec`](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEShuffleReadExec.scala#L74-L94)). As I read it, that works because a Spark partitioning describes the whole stage, so no task needs to know its own slice. Spark's coalescing also covers a case DFD will hit: partitions too small to be worth a worker each. AQE packs contiguous reducer partitions up to a target size ([`ShufflePartitionsUtil`](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ShufflePartitionsUtil.scala#L237-L262)). The read keeps `RangePartitioning` only while every spec is a contiguous, non-overlapping span ([`isCoalescedRead`](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEShuffleReadExec.scala#L151-L157)). Once it splits a skewed partition, it falls back to `UnknownPartitioning` ([L108-L109](https://github.com/apache/spark/blob/cecbd5d846a2cdd891a4d9f0bda302b58b02092a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEShuffleReadExec.scala#L108-L109)). In DataFusion terms, coalescing a span drops the split points inside it, so the result is still a plain `Range` over a subset of the split points. Should DFD coalesce that way before running a range layout on fewer workers? It doesn't remove the per-task slice question, but it shrinks both sides of the trade-off. Under the empty-partition alternative, each task's empty partitions drop from `N - N/W` to `M - M/W` for the coalesced count `M`. CockroachDB is the other range-partitioned engine I checked. It splits scans into per-node key spans ([`PartitionSpans`](https://github.com/cockroachdb/cockroach/blob/d30c905fff79ef825adc96bcc647f1872a90f2ff/pkg/sql/distsql_physical_planner.go#L963-L971)), but its optimizer has no range-layout property ([`physical.Required`](https://github.com/cockroachdb/cockroach/blob/d30c905fff79ef825adc96bcc647f1872a90f2ff/pkg/sql/opt/props/physical/required.go#L32-L69)). Its key-span [range router](https://github.com/cockroachdb/cockroach/blob/d30c905fff79ef825adc96bcc647f1872a90f2ff/pkg/sql/execinfrapb/data.proto#L178-L192) is the closest match to this PR, including the error for rows outside every span. Bulk jobs such as restore use it, not SQL query plans. So neither system puts a worker's key bounds in a partitioning property that the optimizer compares. DataFusion already goes further than Spark, because `RangePartitioning` carries split points and drives co-partitioning. Each DFD task also runs its own DataFusion plan, so its partitioning can describe only its own slice. Bounds fill that gap with key values. Empty partitions fill it by giving each task the stage-wide layout that Spark has. Bounds are `Range`-only, and every rule that compares range layouts (equality, `can_interleave`, #25483) has to define what they mean. Empty partitions need no new API but cost the partitions above. Could the PR description lay out that trade-off, with the measurement, so we can decide on it? > Because this PR is a bit large already, I can file a follow up issue for this. A follow-up works for interleaving. As I read it, that's optimizer behavior behind the existing `can_interleave` signature. Could you file the issue and link it here? The other half of my question decides what the public API means, so I don't think it can wait for a follow-up: how bounds take part when layouts are compared. Today [`has_same_layout`](https://github.com/apache/datafusion/blob/6e318dd7ecbeee7b44a39c0391c36d1e955aa863/datafusion/physical-expr/src/partitioning.rs#L432-L440) ignores bounds and [`==` includes them](https://github.com/apache/datafusion/blob/6e318dd7ecbeee7b44a39c0391c36d1e955aa863/datafusion/physical-expr/src/partitioning.rs#L228-L230). #25483 will build its relationships on one of those, and its `SourceFiner` case needs bounds across tasks. For example, the last partition of `Range(2, [10], end=20)` maps into the first partition of `Range(2, [20])` only because of `end=20`. If #25483 needs bounds to mean something else, we'd be changing the documented semantics of a released type. The [API health policy](https://github.com/apache/datafusion/blob/c120619dec70f723e9e9f9174cb8194c532a5a75/docs/source/contributor-guide/ap i-health.md?plain=1#L82-L84) asks us to avoid breaking changes and to deprecate instead, but a change to what `==` means can't be deprecated, so downstream code would change behavior without warning. Could we settle in this PR how bounds take part in layout comparison and document it on `has_same_layout` and `PartialEq`? -- 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]
