comphead commented on code in PR #7302:
URL: https://github.com/apache/datafusion-comet/pull/7302#discussion_r4405707089
##########
spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala:
##########
@@ -1983,4 +1984,98 @@ class CometNativeShuffleSuite extends CometTestBase with
AdaptiveSparkPlanHelper
}
}
}
+
+ // https://github.com/apache/datafusion-comet/issues/7301: a job launched
from a query's RDD
+ // runs outside a SQL execution, so Spark passes no configs to its map tasks.
+ private def withSessionConfsOutsideSqlExecution(f: String => Unit): Unit = {
+ withTempPath { dir =>
+ spark
+ .range(0, 1000, 1, 4)
+ .selectExpr("id % 10 as k", "id as v")
+ .write
+ .parquet(dir.toString)
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ CometConf.COMET_OFFHEAP_MEMORY_POOL_TYPE.key -> "greedy_unified") {
+ f(dir.toString)
+ }
+ }
+ }
+
+ /** Runs `df` from its RDD and checks that every native plan got the
session's memory pool. */
+ private def checkNativePlansUseSessionConfs(df: DataFrame): Unit = {
+ val appender = new LogAppender("native plan memory pools")
Review Comment:
Good catch, thanks. `LogAppender` keeps INFO and above by default, so the
check saw no lines. That is very likely why the `[shuffle]` job failed. Fixed
in bcf48626d5 with `appender.setThreshold(Level.DEBUG)`. Locally the five tests
pass, and all five fail with `CapturedSQLConf.withConf` turned into a no-op.
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala:
##########
@@ -95,7 +95,8 @@ class CometNativeShuffleWriter[K, V](
override def write(inputs: Iterator[Product2[K, V]]): Unit = {
try {
- writeInternal(inputs)
+ val input = nativeShuffleInput(inputs)
+ input.sqlConf.withConf(writeInternal(input))
Review Comment:
Right, `getWriter` read `spark.comet.shuffle.rss.maxFrameBytes` from the
session before the writer applied anything. Fixed in bcf48626d5: the captured
configs now live on `NativeShuffleSpec`, which `getWriter` reaches through the
dependency, and the frame-size read runs inside `withConf`. The writer takes
them from the same spec. The other two limits in
`CelebornShufflePusherFactory.create` (frame and in-flight bytes) come from the
executor's `SparkConf`, not the session, so they are unchanged. I did not add a
Celeborn test for this read.
--
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]
