comphead commented on PR #6095:
URL:
https://github.com/apache/datafusion-comet/pull/6095#issuecomment-5784076116
Read the full diff plus the shuffle write path, Spark's
`ShuffleExchangeExec`, and arrow-rs's IPC
writer. The row-ordinal key and the run-indexed flush both look right to me,
and the view-type
fallback checks out against `write_array_data`: arrow-rs truncates for every
other family and
serializes view data buffers whole. Below is what is not already covered in
the thread.
## The Spark-equivalence claim is wrong, and it is load-bearing
`spark.sql.execution.sortBeforeRepartition` defaults to `true`. Spark's
round robin therefore sorts
each partition on the binary `UnsafeRow` form *before* assigning positions,
and sets
`isOrderSensitive = isRoundRobin && !sortBeforeRepartition`, which is
`false` by default. Comet's
JVM path copies both, at `CometShuffleExchangeExec.scala:1088` and `:1128`.
Three consequences:
- **The residual assumption is not the one Spark makes.** Spark's is "a
retried task yields the same
multiset of rows". This PR's is "the same rows in the same order", which
is strictly stronger.
`RowGroups` is the `sortBeforeRepartition=false` variant with no sort, so
`replaysRowsInOrder` is
the only thing standing between it and SPARK-23207, rather than one of two
mechanisms that agree
with Spark.
- **`CometNativeShuffleInputRDD.getOutputDeterministicLevel`'s scaladoc
describes the non-default
branch.** "Spark ... wraps the repartition in a `MapPartitionsRDD` with
`isOrderSensitive = true`
(Comet's own JVM path does this in `prepareJVMShuffleDependency`)" is
false in the default
configuration. There the flag is `false` on both paths and the sort
carries the guarantee.
- **The `+ 1` justification does not follow.** "`groupRows = 1` places rows
exactly where Spark's
round robin would for the same row order" holds only with
`sortBeforeRepartition=false`, or on
input already in binary-sorted order. `native_shuffle.md` states that
caveat for `HashAll` and the
new `RowGroups` section drops it.
No code change implied, but this argument is what the next person widening
`replaysRowsInOrder` will
lean on. Worth correcting in the PR body, `native_shuffle.md`, the RDD
scaladoc, and the skill.
## The `Runs` index shape is never spilled in a test
Every spill test is on `RoundRobinStrategy::default()` or `HashAll {
max_hash_columns: 1 }`.
`positional_repartitioner` builds an unbounded pool with `max_buffer_bytes =
None`, so no new test
reaches `reserve_and_may_spill`'s refusal branch, `partitioned_batches()` ->
`empty_like`, or
`RunIterator` over a partially drained `buffered_batches`. Spill is where
the new bookkeeping is
most likely to be wrong and where `row_seq` continuity actually matters.
Adding the `Runs` shape to
the existing `spill_*` cases, plus one that spills mid-stream and re-asserts
`positional_placement_is_a_partition_of_the_input`, would cover it.
## Celeborn
`create_repartitioner` builds the same `MultiPartitionShuffleRepartitioner`
for
`RssPartitionWriter`, so positional is the first path handing that writer
**sliced** batches. It
looks like it copes, since it slices itself in `push_batch_within_limit` and
`additional_ipc_data_size` rebases offsets, but the `maxFrameBytes`
reservation math over a slice is
untested. Separately, `CometShuffleDependency` and
`CometNativeShuffleWriter` are both shared with
Celeborn and the `INDETERMINATE` rollback path is not obviously the same for
push shuffle. Either
gate positional off when `remoteDestination.isDefined` for now, or say why
it is safe.
## Smaller
- **`getOutputDeterministicLevel` cannot fire today.** `replaysRowsInOrder`
requires a
`CometNativeScanExec` leaf and `foreachUntilCometInput` stops there, so
`ctx.inputs` never carries
a shuffle dependency and `super.getOutputDeterministicLevel` is always
`DETERMINATE`. Only the
hand-built RDD in `CometNativeShuffleInputRDDSuite` reaches the other
branch. Fine as defence in
depth for a wider allowlist later, but "Both run, and positional placement
needs both to agree"
reads as if two gates are live.
- **`usesPositionalRoundRobin` ignores `shuffleType`.** It returns `true`
for a
`CometColumnarShuffle` exchange over a scan, where nothing consults it.
`isPositional` in the new
suite compensates by filtering on `shuffleType`. Folding the check in, or
making the predicate
private and going through `positionalRoundRobinSpec` in the test, drops a
public API that exists
for one call site.
- **The view-type fallback loses `maxHashColumns`.** The planner drops
`rr_partition.max_hash_columns`
on the positional branch and `create_repartitioner` falls back to
`RoundRobinStrategy::default()`,
so `positional.enabled=true` plus `maxHashColumns=4` plus a `Utf8View`
column silently gives `0`.
- **`RunIterator`'s two paths disagree in the comments.** "Chunks stay
`batch_size` rows so that
output block sizes do not depend on how long the runs happen to be" sits
directly under a branch
that returns a whole batch of `>= batch_size` rows.
- **`copy_time` versus `interleave_time`.** The metric is still
`interleave_time` in `metrics.rs`
and in the Spark UI, so grepping the metric name no longer finds its
consumers.
- **`Option(context).map(_.partitionId()).getOrElse(0)`.** If `context` is
ever null on a real path,
every task starts at `XORShiftRandom(0).nextInt(n) + 1` and the whole
stage lands on one
consecutive run of partitions, silently. That is exactly what
`positionalStartPartition` exists to
prevent. If it is only null in tests, throwing would make that explicit.
- **`bench_support` ships in release builds.** `#[doc(hidden)] pub mod`
still compiles into the
published crate. `#[cfg(feature = "bench")]` on the module plus
`required-features = ["bench"]` on
the `[[bench]]` target keeps `benches/` working without it.
- **The nested fixture is eight clones of one batch.** `nested_batches`
returns `vec![batch; count]`,
so all eight share buffers, `count_new_buffers` charges seven of them
zero, and the gather rereads
one cache-resident batch. The headline numbers come off this fixture, so
either offset the fill
per batch or say what it understates.
- **Scope and CI.** The bench fix is an independent correctness fix for
`main` that was already its
own PR, and the `bench_support` seam plus `Fill` plus
`partitioning_benchmark` is a third change.
`PR Benchmark Check` is `skipping` on this run, and so are the Spark SQL
suites, while
`PartitionIndices` changes the flush path for every partitioning and not
only positional. A
`run-spark-4.1-tests` label before queueing looks worth it.
## Simplification
- **The rationale is written out seven times.** Decorrelated starts and
SPARK-21782, the framing
independence argument, and the `groupRows` trade-off each appear in
`RoundRobinStrategy`'s
rustdoc, `partitioning.proto`, `native_shuffle.md`, both `CometConf` doc
strings,
`positionalStartPartition` and `replaysRowsInOrder`, the RDD scaladoc, and
the skill. That is
roughly 200 lines of near-identical prose, and it already cost a three-way
fix earlier in this
thread. Keep the argument once in `native_shuffle.md` and leave one-line
pointers.
- **`PositionalRun` duplicates `BufferedRun`.** Two copy structs differing
only in `partition`
versus `batch`, plus a scratch `Vec` and a second traversal of it in
`buffer_positional_batch_may_spill`. Returning an iterator from
`positional_runs` keeps the
unit-testability the scratch vector is presumably there for and drops the
struct, the
`ScratchSpace` field, and the extra pass.
- **`PartitionIndices::empty_like(&self, num_partitions)`** has one caller,
which passes
`self.num_partitions()`.
- **Three of the seventeen new Rust tests are subsumed.**
`positional_runs_cover_every_row_once_in_order` by
`positional_placement_is_a_partition_of_the_input`,
`positional_runs_wrap_and_offset_by_start_partition` by
`positional_placement_walks_consecutive_partitions_from_its_start`, and
`positional_placement_is_independent_of_batch_framing` by
`positional_placement_survives_reframing`. The end-to-end form is the one
worth keeping in each
pair.
- **Two Scala tests are arithmetic, not shuffle.** `map tasks start on
decorrelated partitions` and
`stage-wide balance needs many more groups per task` exercise the local
`stageSpread` helper and
`XORShiftRandom`, not any Comet shuffle code, yet they run in the shuffle
CI shard. ScalaTest's
`clue` is also by-value, so `spread(64)` and `spread(8192)` are each
evaluated twice and the first
is 780k iterations. Bind once.
- There is no `.slt` harness in the repo, so the `checkSparkAnswer` tests
are already in the right
place.
--
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]
