andygrove commented on PR #5449: URL: https://github.com/apache/datafusion-comet/pull/5449#issuecomment-5430021705
One thing worth pulling in before we go further on enumerating safe upstreams: Spark already has a declarative mechanism for exactly this, and our JVM path already uses it. In `prepareShuffleDependency` we set `isOrderSensitive = isRoundRobin && !SQLConf.get.sortBeforeRepartition` (`CometShuffleExchangeExec.scala:1046`) and thread it into both `mapPartitionsWithIndexInternal` calls. That flag is what lets Spark run safely with `sortBeforeRepartition=false` at all — `MapPartitionsRDD.getOutputDeterministicLevel` returns `INDETERMINATE` when the map function is order-sensitive and the parent is `UNORDERED`, and `RDD.getOutputDeterministicLevel` marks every reduce-side RDD `UNORDERED` because, in Spark's own words, "the arrival order of these shuffle blocks are totally random." So a round-robin repartition downstream of another exchange declares itself indeterminate, and the DAGScheduler rolls the whole stage back rather than re-running a single task into a partially-consumed output. `prepareNativeShuffleDependency` (`:771`) never sets that bit. That's correct today, because `HashAll` places rows by content and is genuinely deterministic regardless of input order. With `WholeBatch` it becomes load-bearing and we'd be the only round-robin path in either engine that is positional and neither sorts nor declares itself order-sensitive. The thin RDD isn't a `MapPartitionsRDD`, so we can't just pass the flag, but I don't think we need to: `CometNativeShuffleInputRDD` declares `OneToOneDependency` on each leaf input RDD, so the determinism level already propagates up from the real parents, and `getOutputDeterministicLevel` is `protected` on `RDD` while the class lives in an `org.apache.spark` package. Could we override it there to return `INDETERMINATE` when the parent level is `UNORDERED` and batch-granular round-robin is enabled? What I like about this over an upstream allowlist is that it doesn't require us to be right about which operators preserve order and framing. A plain scan keeps a `DETERMINATE` parent, stays determinate, and still gets cheap per-task retry. Anything downstream of an exchange goes indeterminate on its own and we get either a correct rollback or a loud job abort instead of silently dropping and duplicating rows. It also composes with the sortBeforeRepartition question rather than replacing it, so we could keep the flag defaulting to false and still not be relying on the default to stay safe. -- 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]
