comphead commented on PR #7302: URL: https://github.com/apache/datafusion-comet/pull/7302#issuecomment-6474584872
Thanks @andygrove. All three are addressed in 04cdf52d63. **Tests per site.** The single writer test is replaced by one test per site, each running the query from `toRdd` with AQE off: - the native shuffle writer - a native plan (`CometExecRDD`) in a map stage feeding the JVM shuffle - `LIMIT` (`CometCollectLimitExec`, which goes through `getNativeLimitRDD`) - `ORDER BY ... LIMIT` (`CometTakeOrderedAndProjectExec`, top-K and final) Each one checks the line `CometExecIterator` logs when it creates a plan, and asserts that every native plan resolved the session's `greedy_unified` rather than the default `fair_unified`. A batch count only covered the plan fused into the writer. This check covers every native plan in the job. I ran the same query shapes in `spark-shell`: before the PR, each shape had map-stage plans on `fair_unified` (for `LIMIT` and top-K, 6 of 7 plans). With the PR, all of them resolve `greedy_unified`. **Shuffle reads outside the wrappers.** Confirmed, they could run unwrapped. When a Comet shuffle read has no native plan above it in its stage, it read the defaults. An example is an intermediate stage where Spark's aggregate or project consumes a `CometColumnarExchange`. I checked it with `spark.comet.shuffle.readBufferSize=0`, which fails only where an executor reads the session's value. That stage failed under `collect()` and passed under `toRdd`. `CometShuffledBatchRDD` now captures the configs and applies them in `compute` and `computeAsShuffleBlockIterator`, and a test covers that case. Since shuffle reads now apply the configs themselves, the wrap in `CometNativeShuffleInputRDD.compute` became redundant (its only Comet config reads were those shuffle reads), so I removed it. That RDD now only carries the captured configs to the writer. The JVM columnar writers are tracked in #7303, with a repro, and linked from the description. **Defaults.** The `NativeShuffleSpec` default was already gone. `sqlConf` on `CometNativeShuffleInputRDD` is now required, and both call sites pass `CapturedSQLConf.capture()`. `CometExecRDD` and `CometShuffledBatchRDD` capture in their constructors. An RDD can only be constructed on the driver, since it needs the `SparkContext`, so that stays driver-side. I haven't run the new tests locally. They will run in this PR's CI. -- 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]
