comphead opened a new issue, #7303:
URL: https://github.com/apache/datafusion-comet/issues/7303
### Describe the bug
Follow-up to #7301. In a job that runs outside a SQL execution
(`QueryExecution.toRdd` with AQE on or off, `Dataset.rdd` with AQE off), the
JVM columnar shuffle writers read every `spark.comet.*` setting at its default
in map stages. #7302 fixes native plans and Comet shuffle reads, but not these
writers.
They read their settings on the executor, from `SQLConf.get`, in their
constructors and while they write:
- `CometDiskBlockWriter`: batch size, codec, zstd level, offsets as lengths,
spill threshold, dictionary ratio
- `CometShuffleExternalSorter`: the same settings
- `CometUnsafeShuffleWriter`, `CometBypassMergeSortShuffleWriter`: tracing
- `SpillWriter`, `SpillSorter`: `spark.comet.shuffle.jvm.batchSize`
- `CometShuffleManager.shouldBypassMergeSort`:
`spark.comet.shuffle.jvm.maxWritersPerExecutor`
### Steps to reproduce
Spark 4.1.1, `local[2]`, with the #7302 head (`04cdf52d63`).
`spark.comet.shuffle.compression.codec=bogus` fails its value check when an
executor reads it, so a job fails only if the writer sees the session's value.
```scala
spark.range(0, 1000, 1, 4).selectExpr("id % 10 as k", "id as
v").write.mode("overwrite").parquet("/tmp/t")
spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.comet.shuffle.mode", "jvm")
spark.conf.set("spark.comet.shuffle.compression.codec", "bogus")
val df = spark.read.parquet("/tmp/t").repartition(2, col("k")) //
CometColumnarExchange over CometNativeScan
df.collect()
// fails: IllegalArgumentException: The value of
spark.comet.shuffle.compression.codec should be one of zstd, lz4, snappy, but
was bogus
df.queryExecution.toRdd.foreach(_ => ())
// succeeds: the map stage's writer used the default codec
```
### Expected behavior
The JVM columnar shuffle writers use the session's Comet settings however
the job is launched.
### Additional context
A possible fix: capture the configs on `CometShuffleDependency` with
`CapturedSQLConf` from #7302, and in `CometShuffleManager.getWriter` build the
JVM writers inside `withConf` and return a writer that also applies it around
`write` and `stop`.
--
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]
