kazuyukitanimura commented on code in PR #7336:
URL: https://github.com/apache/datafusion-comet/pull/7336#discussion_r4459438525
##########
spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala:
##########
@@ -326,6 +327,46 @@ class CometPluginsUnifiedModeSuite extends CometTestBase {
assert(execMemOverhead3 == "1G")
assert(execMemOverhead4 == "1G")
}
+
+ // This suite names no shuffle manager. From Spark 4.0 the plugin chooses
Comet's; before that
+ // Spark has built its own by the time the plugin runs, and Comet disables
itself.
+ test("the plugin sets Comet's shuffle manager from Spark 4.0 and Comet
runs") {
+ val expected = CometSparkSessionExtensions.isSpark40Plus
+ assert(SparkEnv.get.shuffleManager.isInstanceOf[CometShuffleManager] ==
expected)
+ withParquetTable((0 until 100).map(i => (i, (i % 7).toString)), "tbl") {
+ val (_, plan) = checkSparkAnswer("SELECT _2, count(*) FROM tbl GROUP BY
_2")
+ assert(collect(plan) { case op: CometShuffleExchangeExec => op
}.nonEmpty == expected, plan)
+ }
+ }
+}
+
+class CometDriverPluginShuffleManagerSuite extends SparkFunSuite {
Review Comment:
CometDriverPluginShuffleManagerSuite is a new top-level suite. CI runs the
plugin suites from an explicit list passed as -DwildcardSuites in
.github/workflows/pr_build_linux.yml (around line 582) and pr_build_macos.yml
(around line 248). Nothing in either list matches this name, so the
version-gate tests and the keep-the-user's-value tests don't run in CI. Could
you add it to both lists? Once it's registered, the run-all-spark-profiles
label would cover 3.4 and 4.2, which the local runs in the description didn't
include.
##########
spark/src/main/scala/org/apache/spark/Plugins.scala:
##########
@@ -174,6 +176,21 @@ object CometDriverPlugin extends Logging {
/** Spark config key under which the loaded Comet version is exposed at
runtime. */
val COMET_VERSION_CONFIG = "spark.comet.version"
+ // Choose Comet's shuffle manager for an application that did not, so that
Comet does not
+ // disable itself. Only from Spark 4.0 (SPARK-45762) does SparkContext
create the manager after
+ // it initializes the plugins, from the same conf that executors then fetch.
An application's own
+ // choice, even Spark's, is kept: it may depend on that manager, as Celeborn
does.
+ private[apache] def maybeSetShuffleManager(conf: SparkConf): Unit = {
Review Comment:
Thanks for working this out from the Spark 4.0 source. I checked the same
ordering in v4.0.0 and it holds on the driver and the executors. One case
concerns me. maybeSetShuffleManager checks spark.comet.enabled and Comet
shuffle, but isCometLoaded can still disable Comet afterwards. That happens on
a big-endian host, when spark.sql.parquet.int96TimestampConversion=true, or
when the native library fails to load and spark.comet.nativeLoadRequired is
left at false. In those applications this PR swaps SortShuffleManager for
CometShuffleManager, and Comet never runs. Spark's
ShuffleExchangeExec.needToCopyObjectsBeforeShuffle returns true for any manager
that is not a SortShuffleManager, so every Spark exchange then copies each row
before the shuffle. On 1.2 the same application ran on Spark's manager with no
copy. Could the plugin skip setting the manager when it can see one of those
conditions at startup? A small helper shared with isCometLoaded would keep the
two from drifting. A tes
t with one of those settings would lock it in.
--
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]
