andygrove commented on PR #5449: URL: https://github.com/apache/datafusion-comet/pull/5449#issuecomment-5768393056
Coming back to this after reading the two new defences together, and I think they cancel each other out at the defaults, which is the thing to settle before the placement question. `failIfRetryingPositionalRoundRobin` throws when `attemptNumber > 0` or `stageAttemptNumber > 0`. But every path the `INDETERMINATE` declaration exists to enable goes through a stage resubmission, and `DAGScheduler.submitMissingTasks` calls `stage.makeNewStageAttempt()` on every submission, so every rolled-back task carries `stageAttemptNumber >= 1` and hits the throw before it does any work. The rollback that `CometNativeShuffleInputRDD.getOutputDeterministicLevel` asks for can never complete while `failOnRetry` is at its default, so the fallback sentence in the config doc describes a path you can only reach by turning the flag off. The operational side is worse than the design side: an executor lost mid-stage re-enqueues its map tasks with a fresh `attemptNumber`, an executor lost after the stage completes produces a fetch failure and a resubmit, both throw, both exhaust `spark.task.maxFailures`, and the job dies. Speculation adds a spurious failure on every long-tail task. I don't think we can ask anyone to enable this on the kind of cluster where the 47x is worth having. The deeper reason the flag is there is that `round_robin_batch_seq` keys placement on batch framing, and framing is the one property no Spark contract covers. `DETERMINATE` promises the same rows in the same order and says nothing about how a downstream operator chunks them, so a spilling operator between the scan and the writer can reframe under different memory pressure while still satisfying the contract. `failOnRetry` is covering exactly that uncovered gap. Which is the argument I'd now make for the row-counter variant from my earlier comment, and it's a better one than the performance argument: `partition = (seed + rows_seen / B) % n` closes the gap, so the residual assumption becomes exactly the one Spark's own round robin makes. That's the assumption the determinism level already describes and the DAGScheduler already knows how to act on, so rollback becomes load-bearing instead of decorative and `failOnRetry` can go away entirely. Skew also stops depending on how the reader happened to frame things — a task emitting 8 batches into 200 output partitions touches 8 reducers today, which the trade-offs section concedes. One thing the override can't see, separately from all that. `CometNativeShuffleInputRDD`'s dependencies are `ctx.inputs` (`CometShuffleExchangeExec.scala:126`), the leaf RDDs of the inlined native subtree. Everything between those leaves and the writer runs inside the same `CometExecIterator` and never appears in the RDD graph, so a spilling native aggregate under a Parquet scan still reports `DETERMINATE`. Spark is blind in the same way for `MapPartitionsRDD`, which is why `sortBeforeRepartition` defaults to true — the determinism level is Spark's backstop for people who turn the sort off, not its primary defence. So what we have restores parity with `sortBeforeRepartition=false`, not with the default. Could we close that with a plan-level check on the native subtree instead? With the row counter the predicate shrinks from "preserves order and framing" to "preserves order", which is short enough to be checkable — scan, project, filter, not aggregate or join or anything that spills — and much more defensible than the allowlist I was skeptical of in August. And when the check fails I'd rather fall back to `HashAll` at plan time than commit to positional placement and then fail the job from inside a task, which is the opposite of how we handle every other unsupported case. On the two things I said I wanted to check before committing to the run-based index, both turn out to be answerable from arrow-rs rather than from a benchmark. `ArrayData::slice` propagates the slice into struct children (`arrow-data-59.3.0/src/data.rs:630`) and the IPC writer truncates per type — `get_or_truncate_buffer` for numeric and temporal, `reencode_offsets` for the byte arrays, `get_list_array_buffers` for lists and maps, `bit_slice` for booleans — so a sliced nested struct writes only its window and the write-amplification worry doesn't apply. The exception is `Utf8View`/`BinaryView`, where the views buffer is truncated but every variadic data buffer is written in full, which we already know about at `rss_partition_writer.rs:744`. So the test worth writing is narrow and specific to view columns. The wrinkle is that `arrow::compute::concat` short-circuits at one input with `array.slice(0, array.len())` (`arrow-select-59.3.0/src/concat.rs:506`), so a chunk that is a single strict sub-range comes back as a slice rather than a fresh batch and lands on that path. Worth deciding deliberately rather than letting it fall out. On sizing `B`, I'd now measure the default rather than derive it from the formula I suggested. `clamp(batch_size / num_partitions, 64, batch_size)` gives 64 at the default partition count, which is ~128 runs per 8192-row batch and, on a 194-column nested schema, tens of thousands of small bulk copies per batch. Still far better than a per-row gather, but it also eliminates the zero-copy path completely, where `B = batch_size` keeps it whenever framing is aligned and degenerates to exactly what this PR does today. The two ends behave differently enough that I wouldn't guess. Last thing, and it belongs in its own issue rather than here: `HashAll` isn't a good fallback either. `pmod(murmur3(whole row), n)` sends identical rows to the same partition, so `repartition(200)` over low-cardinality data collapses onto a handful of reducers where Spark's round robin spreads it evenly, and `maxHashColumns` makes that strictly worse. That's an independent reason to get positional placement right rather than treating the hash path as the safe harbour. -- 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]
