sunchao commented on PR #5668:
URL:
https://github.com/apache/datafusion-comet/pull/5668#issuecomment-5535696484
Reviewed head `6f2465a2` against base `4219fc79` across five independent
scopes. **I found one recovery bug and two performance issues worth
addressing.**
1. **[P1] A remote failure can abort the local replacement.**
The [replacement
dependency](https://github.com/apache/datafusion-comet/blob/6f2465a2f3d1cb2915a7c7f5551aa7032719467d/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleDependency.scala#L107)
reuses the remote stage’s exact input RDD. Spark propagates stage failures
using RDD identity, so separate shuffle and stage IDs do not provide sufficient
isolation.
I reproduced this in the PR’s existing tests: the remote stage’s
size-limit failure killed the replacement stage. A separate Spark probe
confirmed that a fresh sibling input RDD avoids this failure. Create an
independent scheduling RDD over the original upstream inputs, or fully retire
the remote job before starting its replacement. A narrow wrapper around the old
RDD is insufficient.
2. **[P2] Non-AQE execution serializes independent shuffle branches.**
The [new RDD constructor
call](https://github.com/apache/datafusion-comet/blob/6f2465a2f3d1cb2915a7c7f5551aa7032719467d/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffledRowRDD.scala#L57)
waits for materialization to finish. Joins and unions construct their inputs
sequentially, so one sibling shuffle now finishes before another starts, even
when neither needs fallback.
In a Spark scheduling probe, two one-second map tasks overlapped for
approximately one second with ordinary scheduling and had **zero overlap** with
eager materialization. Submit independent materializations before awaiting
their selected dependencies, while preserving the fence against downstream
reads.
3. **[P2] Fallback double-counts runtime statistics used by AQE.**
The [replacement reuses the original
metrics](https://github.com/apache/datafusion-comet/blob/6f2465a2f3d1cb2915a7c7f5551aa7032719467d/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleDependency.scala#L119).
Successful remote maps have already contributed to them; recomputing every map
locally adds those contributions again.
An exact-head Spark/native probe returned **2 rows but reported 3 runtime
rows**, after one successful remote map and two local maps. Inflated size
statistics can also affect AQE join selection. Give each destination separate
output statistics and publish the selected destination’s values. Resetting
shared counters remains vulnerable to late remote completions.
The core design is reasonable: typed errors and a driver-owned, immutable
storage decision are useful abstractions. The materialization wrapper has real
cancellation and lifecycle responsibilities. The main improvements should make
attempt ownership explicit and move blocking orchestration out of the RDD
constructor. I found no meaningful new native encoding/copy overhead. The 512
MiB admission default raises potential memory usage without allocating that
amount eagerly.
Validation: native build, **34 RSS tests**, and **2 JNI tests** passed. The
focused Spark 4.1 run passed **9/11 tests**, with both failures reproducing the
recovery race. [CI also has failing fallback tests on Spark 3.5, 4.0, and
4.2](https://github.com/apache/datafusion-comet/actions/runs/33832861909). JVM
tests used the PR’s fake Celeborn client with real Spark/native execution. Full
cluster behavior and whole-query performance were not benchmarked.
--
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]
