andygrove commented on PR #6095: URL: https://github.com/apache/datafusion-comet/pull/6095#issuecomment-5804684625
Both P2s were real, and both are fixed. **Nondeterministic projections.** `replaysRowsInOrder` now admits a `CometProjectExec` only if every expression in its project list is deterministic, and a `CometFilterExec` only if its condition is. The reasoning was "projections and filters are row-wise". That holds for what the operator does, but a nondeterministic expression can still return different values on a retry, so it can reorder or re-filter rows. `CometNativePositionalRoundRobinSuite` checks a nondeterministic UDF projection and a `rand(42)` filter, and both now fall back to `HashAll`. It also checks that the same UDF without `asNondeterministic` is still admitted, so the test shows the determinism flag is what excludes it and not a fallback somewhere else in the plan. **Group size not frozen.** The group size is now resolved on the driver, in `CometShuffleExchangeExec.resolvePositionalGroupRows`, and stored as a concrete value in the `PositionalRoundRobin` carried by `NativeShuffleSpec`, so it is frozen with the dependency. The native auto-resolution is gone (`AUTO_GROUP_ROWS`, `resolve_group_rows`), and the planner rejects a group size that is not positive, so the value can no longer come from the executor. The new test follows your repro: it runs a repartition at `spark.comet.batchSize=8192`, drops the map output through `MapOutputTrackerMaster.unregisterAllMapAndMergeOutput`, halves the batch size, re-runs the same executed plan, and requires every partition to hold the same rows in the same order. To check that the test catches the bug, I temporarily moved the resolution back into the writer, using the task's batch size. The test failed: the re-run tasks saw 4096 and placed rows differently. **Performance qualifications.** Both are taken: - The PR description now says the benchmark covers placement and the gather only. It leaves out encoding, compression, the file write and the reduce side, so it shows where the write path saves time, not how much faster a whole shuffle gets. - "Wraps once per batch" is now qualified in the config doc, in `native_shuffle.md` and in the description. Because of the 64-row floor, past `batch_size / 64` output partitions a task covers only `batch_size / 64` partitions per batch, so small map tasks can still leave reducers empty. I ran your 1,000-reducer case through `positionalStartPartition` and the resolved group size and got the same 311 empty reducers, so the docs use that figure. I also updated the review skill's positional item: a reviewer should now check that admitted expressions are deterministic and that the group size is still resolved on the driver. -- 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]
