comphead commented on PR #5449:
URL:
https://github.com/apache/datafusion-comet/pull/5449#issuecomment-5400143979
@andygrove @sunchao need your brain on the below:
Although the performance is great, we need to think if this PR addresses
task retry correctly, namely:
Spark requires round-robin shuffle to be deterministic under task retry.
If a retried map
task places rows on different reducer partitions than the original
attempt, and downstream
reducers have already consumed the original output, the job silently loses
or duplicates
rows. This is
[SPARK-23207](https://issues.apache.org/jira/browse/SPARK-23207).
Spark's round-robin assignment is **positional, not content-based**
(`CometShuffleExchangeExec.scala:969-975`, copied from
`ShuffleExchangeExec`):
```scala
var position = new XORShiftRandom(partitionId).nextInt(numPartitions)
(_: InternalRow) => { position += 1; position } // then HashPartitioner
does the mod
```
A counter seeded per map task, bumped per row. Placement depends purely on
a row's *index*
in the input iterator. Nothing about the row itself is inspected.
`sortBeforeRepartition=true` therefore wraps the input in a local sort by
binary `UnsafeRow`
(`:999-1035`, `UnsafeExternalRowSorter` with `RecordBinaryComparator`)
before the counter
runs. The sort canonicalizes arrival order, which makes index a function
of content, which
makes placement deterministic.
### Why determinism is required
A map task can be retried after downstream reducers have already consumed
the original
attempt's output (fetch failure, executor loss, speculation). If attempt 2
assigns rows to
different reducers than attempt 1, reducers that already fetched keep
attempt-1 rows while
reducers fetching after the retry get attempt-2 rows. The union is not the
input: rows are
lost and duplicated, silently, with no failure. That is SPARK-23207, and
it is why the sort
is on by default despite costing a full external sort per map task.
### How this PR provides determinism
`RoundRobinStrategy::WholeBatch` is **structurally the same design as
Spark's**, one level
coarser: `round_robin_batch_seq % n` per batch instead of `position % n`
per row. Both are
positional. Both are deterministic exactly when input order is.
The difference is where the order guarantee comes from:
| | Spark | `HashAll` (Comet default) | `WholeBatch` (this PR) |
|---|---|---|---|
| Placement key | row index | `pmod(murmur3(row), n)` | batch index |
| Order-sensitive | yes | no | yes |
| Determinism from | forced local sort | row content | assumed
order-preserving upstream |
| Cost | external sort/task | hash every column of every row | one modulo
per batch |
So the PR provides determinism **by assumption, not by construction**. It
takes Spark's
positional design and drops the sort that makes it safe, betting that
Comet's Parquet scan
already emits identical batches in identical order on retry, which for a
plain scan it does.
The bet is expressed as an opt-in flag defaulting to `false`, prose on
`RoundRobinStrategy`
and the `CometConf` entry, and
`test_round_robin_batch_granular_retry_deterministic` asserting
byte-identical output across
two runs.
What it does not have is Spark's enforcement. Spark does not trust its
upstream, it sorts.
`WholeBatch` trusts its upstream and nothing checks that the trust is
warranted:
`CometNativeShuffleWriter.scala:332` sets the flag regardless of what the
child is, and the
support check at `CometShuffleExchangeExec.scala:515` tests only whether
round-robin is
enabled. Note also that Comet's native shuffle path ignores
`sortBeforeRepartition`
entirely, since the sort lives in `prepareJVMShuffleDependency` and not in
`prepareNativeShuffleDependency` (`:764`). For `HashAll` that is correct.
For `WholeBatch`
it removes the one mechanism Spark relies on.
--
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]
