andygrove commented on PR #6095: URL: https://github.com/apache/datafusion-comet/pull/6095#issuecomment-5796689936
Thanks, this is a careful read, and the first point is one I had backwards in a way that matters. `sortBeforeRepartition` defaults to true, so Spark's round robin only asks a retry for the same rows, while positional placement asks for the same rows in the same order. I had described the two assumptions as identical, and I repeated it on the planner thread when I said `groupRows = 1` places rows where Spark's would, which only holds with the sort off. The consequence is the one you name: `replaysRowsInOrder` is the whole safety argument, not one of two mechanisms agreeing with Spark. I have rewritten it that way everywhere, which after your simplification point is one place. `native_shuffle.md` now opens its round robin section with what Spark actually does in both modes and carries the full argument, and the rustdoc, proto, config docs, scaladoc and the review skill each say what they need locally and point there. The RDD scaladoc now calls `isOrderSensitive` the non-default bran ch and describes the check as defence in depth that cannot fire under today's allowlist. I confirmed that last part in `buildNativeContext` rather than take it on trust: a native scan leaf adds no input RDD. The spill gap was real. `check_spill_metrics_count_input_buffers` and `heterogeneous_spill_metrics_do_not_depend_on_input_batching` now run the run-indexed shape alongside the row-indexed one, the second with one-row groups so every spill slices and concatenates across the view, dictionary and list columns. A new `positional_placement_survives_spilling` pushes ragged framing through both spill triggers, the buffer limit and a pool that refuses to grow, and requires the result to equal the unspilled placement row for row, each partition in order. Resetting `row_seq` in `spill()` fails it, which is the bug it exists to catch. On Celeborn I have taken the gate rather than the argument. `positionalRoundRobinSpec` returns `None` under the Celeborn manager, with a test in `CometCelebornShufflePlanningSuite` over a bare native scan, so the manager is the only thing ruling it out, and removing the check fails it. Making the case for push shuffle properly means testing the `maxFrameBytes` reservation over slices and walking through Celeborn's stage-rerun path, which feels like its own PR. The smaller ones are all in. `usesPositionalRoundRobin` is a package-private method on the exec now, with `shuffleType` folded into the decision, and the public companion predicate is gone. The view-type fallback keeps `maxHashColumns`: `RowGroups` carries it, and the fallback is a small function with its own test. On `RunIterator`, a buffered batch can never exceed `batch_size` because `insert_batch` slices to it, so the zero-copy branch's `>=` could only ever match at equality. It is `==` now and both comments say the same thing. The timer is `interleave_time` again to match the metric, and the metrics doc describes it as the gather either way. A missing `TaskContext` throws instead of starting every task at the same partition. You were right about the fixture, and it went further than the nested one, since the flat fixture in the partitioning bench was eight clones too. Both now build each batch from where the previous one left off. I ran the cloned fixture back to back with the new one on the same machine, and the clone reproduces the numbers first posted here almost exactly, so the difference is the fixture. Distinct batches make the nested gathers 2.4x to 4.8x slower, the per-row interleave most of all, since it now reads from 55 MB of batches instead of one that stays in cache. Nested `HashAll` `place+gather` goes from 37.7 ms to 106 ms. The flat schema moves under 1.5x because its eight batches fit in cache either way, and the zero-copy arm barely moves. The upshot is that `RowGroups(auto)` is about 5x `HashAll` end to end on nested rather than 4.5x. The cheap-hash middle ground also looks worse than your `place`-only numbers suggested: hashing one column is 21x cheaper to place but only 1.1x faste r end to end, because it leaves the per-row gather in place. The updated table is in the description, with the caveat that the machine was not idle and the gathers moved by up to 14% between two runs. I'd rather not take `#[cfg(feature = "bench")]`, though. The crate is `publish = false`, so the only thing it ships in is libcomet, where an unreferenced module is dropped at link time. `required-features` on the bench target would also make `cargo clippy --all-targets` in `rust-test` and `cargo check --benches` in the benchmark check skip the bench silently, and not compiling it is how the `OnceLock` panic sat on `main` unnoticed. Would you be happy leaving it ungated, or does the gate buy something I'm not seeing? On scope, folding in the bench fix was deliberate, since without it nothing on this branch can be measured. I agree it and the bench seam are separable from the placement change, though, and I'm happy to split them back out if that makes this easier to review. I've applied `run-spark-4.1-tests`, since as you say `PartitionIndices` changes the flush path for every partitioning, and `run-benchmark-check` for the bench changes. On simplification, `PositionalRun` is gone. `positional_runs` returns an iterator of `(partition, rows)` that is consumed straight into `BufferedRun`s, which drops the scratch field and the second pass, and `empty_like` takes no argument. The three subsumed Rust tests are removed, and `positional_placement_begins_at_the_start_partition` went with them, since the walk test now asserts every group's partition, a wrapping start included. The two arithmetic Scala tests are replaced by one that runs ten real map tasks of 5,000 rows into 200 reducers through the shuffle and asserts none is empty. With the start reverted to the bare map partition id it fails with exactly 112 empty, the number @mbutrovich's simulation predicted. -- 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]
