sunchao commented on PR #6095:
URL:
https://github.com/apache/datafusion-comet/pull/6095#issuecomment-5798535020
Reviewed head `dfc5d2b` with **five agents**. I recommend changes for **two
correctness issues** in the opt-in positional path.
1. **[P2] The allowlist accepts projections that change row order on
retry.**
[CometShuffleExchangeExec.scala:381](https://github.com/apache/datafusion-comet/blob/dfc5d2b02430d362bb3c4e3b02dc23fd4de4a839/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala#L381)
Native `.asNondeterministic()` Scala UDFs pass this gate. We reproduced a
projection that swaps two 64-row groups between attempts while preserving
exactly the same 128 rows. Positional placement sends every row to the other
reducer, yet the RDD reports `DETERMINATE`.
Combining original and regenerated partition outputs produced **64
missing and 64 duplicated values**. Content hashing and Spark’s default sorted
round robin preserved placement.
**Fix:** require replayable projection/filter expressions before
admitting positional placement. A conservative determinism check would close
this case.
2. **[P2] Automatic group size is not frozen with the shuffle dependency.**
[CometShuffleExchangeExec.scala:342](https://github.com/apache/datafusion-comet/blob/dfc5d2b02430d362bb3c4e3b02dc23fd4de4a839/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala#L342)
The driver captures `groupRows=0`; native execution resolves it against
the current task’s batch size. We executed a Dataset, changed
`spark.comet.batchSize` from 8192 to 4096, invalidated its map output, and
re-executed that same Dataset. The dependency stayed identical, but effective
groups changed from 1024 to 512 for eight reducers.
The production native writer confirmed changed assignments for unchanged
ordered input. Mixing old and regenerated outputs duplicated and omitted
**1,024 values out of 16,384**.
**Fix:** resolve the effective group size on the driver, or capture its
batch-size input with the dependency.
The **design and abstractions otherwise look justified**: row ordinals
remove batch-framing dependence, and `Rows`/`Runs` represent meaningfully
different copying patterns. No additional memory, spill, or ownership defect
survived verification.
Two performance qualifications matter:
- The quoted speedups measure **placement plus gathering**, excluding
encoding, compression, storage, and reducers.
- The automatic 64-row floor can still leave reducers empty. The config
documentation’s “wraps once per batch” claim needs qualification; ten 8,192-row
maps into 1,000 reducers left **311 empty reducers** in the verified simulation.
Validation included **165 passing native tests**, **480 placement/framing
cases**, **24,768 IPC round trips**, and production-writer probes across all
four codecs with and without spilling. [Required
CI](https://github.com/apache/datafusion-comet/actions/runs/35873988494) and
the [Spark 4.1 SQL
suites](https://github.com/apache/datafusion-comet/actions/runs/35874023698)
passed.
The Spark probes used exact-head Scala overlays with cached surrounding
components. The production native writer was built from this head. Complete
Spark fetch-failure recovery was not induced; old/new outputs were deliberately
combined.
--
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]
