andygrove commented on code in PR #7336:
URL: https://github.com/apache/datafusion-comet/pull/7336#discussion_r4473618223
##########
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:
Done in bfc14c7671.
`CometSparkSessionExtensions.unsupportedEnvironmentReason` now holds the
big-endian, INT96 and native-library checks. `isCometLoaded` and
`maybeSetShuffleManager` both call it, so the two cannot drift. One difference:
if the library fails to load and `spark.comet.nativeLoadRequired` is true,
`isCometLoaded` throws, but the plugin treats that as supported and still sets
the manager, so the failure shows up where it did before rather than at
startup. `CometDriverPluginShuffleManagerSuite` has a test with
`spark.sql.parquet.int96TimestampConversion=true`. The other two conditions
can't be set in a unit test. The check only sees the startup value of the INT96
setting, since a session can still change it later.
##########
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:
Added to both `pr_build_linux.yml` and `pr_build_macos.yml` in bfc14c7671.
##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -57,6 +57,15 @@ Treat setting one of these keys as a temporary measure. If
you find you cannot s
legacy behavior, please open an issue describing your use case so it can be
considered before the
key is removed.
+## Upgrading to Comet 1.3.0
+
+### Shuffle Manager Set by the Plugin on Spark 4.0 and Later
Review Comment:
The 1.3.0 section now says up front that it needs no `spark.comet.legacy.*`
key, as 1.2.0 does (bfc14c7671).
--
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]
