You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/14 15:40:15 UTC
(datafusion-comet) 01/01: fix: make Comet's in-memory cache opt-in again (#6963)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6963-98fed43e601bf5ab912dd0953c1d2f54526738ae
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 6d33dd80518afc56fbc889b3232e69794ef9a1bf
Author: Andy Grove <ag...@apache.org>
AuthorDate: Wed Oct 14 15:39:50 2026 +0000
fix: make Comet's in-memory cache opt-in again (#6963)
Set spark.comet.exec.inMemoryCache.enabled back to false on every Spark
version. Since #5634 turned it on from Spark 3.5, sorted cached relations
reach native joins that do not keep the order Spark relies on (#6787,
#6898, #6922), so those wrong results are reachable with default settings.
Keep Comet's cache suites and the 3.5, 4.0, 4.1 and 4.2 Spark SQL diffs
running against Comet's cache format by setting the flag beside the cache
serializer, and restore the opt-in wording in the docs.
---
dev/diffs/3.5.9.diff | 44 +++++++++++++---------
dev/diffs/4.0.4.diff | 44 +++++++++++++---------
dev/diffs/4.1.3.diff | 44 +++++++++++++---------
dev/diffs/4.2.0.diff | 44 +++++++++++++---------
docs/source/contributor-guide/spark-sql-tests.md | 7 ++--
.../user-guide/latest/compatibility/operators.md | 11 +++---
docs/source/user-guide/latest/in-memory-cache.md | 24 ++++++++----
docs/source/user-guide/latest/migration-guide.md | 25 ------------
docs/source/user-guide/latest/operators.md | 2 +-
.../user-guide/latest/understanding-comet-plans.md | 2 +-
.../main/scala/org/apache/comet/CometConf.scala | 10 ++---
.../comet/exec/CometInMemoryCacheSuite.scala | 8 ++--
12 files changed, 139 insertions(+), 126 deletions(-)
diff --git a/dev/diffs/3.5.9.diff b/dev/diffs/3.5.9.diff
index dd71681d4d..64fb818293 100644
--- a/dev/diffs/3.5.9.diff
+++ b/dev/diffs/3.5.9.diff
@@ -38,7 +38,7 @@ index 49eca0f7555..b9e5d69a6c6 100644
</dependencyManagement>
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
-index daa3ebac992..2c721d85961 100644
+index daa3ebac992..8e2147bd670 100644
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@ -459,9 +459,11 @@ object SparkBuild extends PomBuild {
@@ -53,7 +53,7 @@ index daa3ebac992..2c721d85961 100644
enable(SparkConnectCommon.settings)(connectCommon)
enable(SparkConnect.settings)(connect)
-@@ -1198,6 +1200,35 @@ object SQL {
+@@ -1198,6 +1200,37 @@ object SQL {
)
}
@@ -61,10 +61,11 @@ index daa3ebac992..2c721d85961 100644
+ * Comet disables itself when its shuffle manager is not registered. When Comet is enabled, make
+ * that shuffle manager the default for every test SparkConf in the modules that have Comet on
+ * their classpath, so that suites which build their own SparkSession or SparkContext run Comet
-+ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's cache serializer
-+ * is made the default the same way. Spark keeps the first cache serializer it loads for the rest
-+ * of the JVM, so a suite that caches through a session of its own would otherwise leave the
-+ * suites after it in Spark's cache format.
++ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's in-memory
++ * cache, which is disabled by default, is turned on the same way, and its serializer made the
++ * default. Spark keeps the first cache serializer it loads for the rest of the JVM, so a suite
++ * that caches through a session of its own would otherwise leave the suites after it in Spark's
++ * cache format.
+ */
+object CometTestSettings {
+ private val cometEnabled =
@@ -73,6 +74,7 @@ index daa3ebac992..2c721d85961 100644
+ private val defaults = Seq(
+ "spark.shuffle.manager" ->
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
++ "spark.comet.exec.inMemoryCache.enabled" -> "true",
+ "spark.sql.cache.serializer" ->
+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+
@@ -3647,10 +3649,10 @@ index e937173a590..263934fbe7b 100644
spark.internalCreateDataFrame(withoutFilters.execute(), schema)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-index c23bf4204f7..d922f0b3d23 100644
+index c23bf4204f7..b6cded1de26 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-@@ -97,6 +97,24 @@ trait SharedSparkSessionBase
+@@ -97,6 +97,27 @@ trait SharedSparkSessionBase
// this rule may potentially block testing of other optimization rules such as
// ConstantPropagation etc.
.set(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, ConvertToLocalRelation.ruleName)
@@ -3666,11 +3668,14 @@ index c23bf4204f7..d922f0b3d23 100644
+ .set("spark.shuffle.manager",
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
-+ // CometDriverPlugin installs Comet's cache serializer when
-+ // spark.comet.exec.inMemoryCache.enabled is on, as it is by default. These sessions do not
-+ // load the plugin, so install it here.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // Comet's in-memory cache is disabled by default, so turn it on to run these suites
++ // against Comet's cache format. CometDriverPlugin installs Comet's cache serializer when
++ // spark.comet.exec.inMemoryCache.enabled is on, but these sessions do not load the plugin,
++ // so install it here.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
conf.set(
StaticSQLConf.WAREHOUSE_PATH,
@@ -3756,10 +3761,10 @@ index 6160c3e5f6c..bfc0c618a9b 100644
test("SPARK-4963 DataFrame sample on mutable row return wrong result") {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-index 1d646f40b3e..b8a043c6248 100644
+index 1d646f40b3e..800bb0c88cb 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-@@ -53,25 +53,45 @@ object TestHive
+@@ -53,25 +53,48 @@ object TestHive
new SparkContext(
System.getProperty("spark.sql.test.master", "local[1]"),
"TestSQLContext",
@@ -3812,9 +3817,12 @@ index 1d646f40b3e..b8a043c6248 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // As in SharedSparkSession: what CometDriverPlugin would install.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // As in SharedSparkSession: Comet's in-memory cache, and the serializer
++ // CometDriverPlugin would install for it.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
+ conf
diff --git a/dev/diffs/4.0.4.diff b/dev/diffs/4.0.4.diff
index 28d63654b5..ac636fd808 100644
--- a/dev/diffs/4.0.4.diff
+++ b/dev/diffs/4.0.4.diff
@@ -78,7 +78,7 @@ index 046f357ff79..7c6fd199e41 100644
<dependency>
<groupId>org.apache.datasketches</groupId>
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
-index 9c2312ed256..09026e8dc8a 100644
+index 9c2312ed256..6bc184ca650 100644
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@ -414,9 +414,11 @@ object SparkBuild extends PomBuild {
@@ -93,7 +93,7 @@ index 9c2312ed256..09026e8dc8a 100644
enable(HiveThriftServer.settings)(hiveThriftServer)
-@@ -1183,6 +1185,35 @@ object SQL {
+@@ -1183,6 +1185,37 @@ object SQL {
}
}
@@ -101,10 +101,11 @@ index 9c2312ed256..09026e8dc8a 100644
+ * Comet disables itself when its shuffle manager is not registered. When Comet is enabled, make
+ * that shuffle manager the default for every test SparkConf in the modules that have Comet on
+ * their classpath, so that suites which build their own SparkSession or SparkContext run Comet
-+ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's cache serializer
-+ * is made the default the same way. Spark keeps the first cache serializer it loads for the rest
-+ * of the JVM, so a suite that caches through a session of its own would otherwise leave the
-+ * suites after it in Spark's cache format.
++ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's in-memory
++ * cache, which is disabled by default, is turned on the same way, and its serializer made the
++ * default. Spark keeps the first cache serializer it loads for the rest of the JVM, so a suite
++ * that caches through a session of its own would otherwise leave the suites after it in Spark's
++ * cache format.
+ */
+object CometTestSettings {
+ private val cometEnabled =
@@ -113,6 +114,7 @@ index 9c2312ed256..09026e8dc8a 100644
+ private val defaults = Seq(
+ "spark.shuffle.manager" ->
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
++ "spark.comet.exec.inMemoryCache.enabled" -> "true",
+ "spark.sql.cache.serializer" ->
+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+
@@ -4482,10 +4484,10 @@ index f0f3f94b811..b7d18771314 100644
spark.internalCreateDataFrame(withoutFilters.execute(), schema)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-index 720b13b812e..388dad6b4f3 100644
+index 720b13b812e..34950a76bde 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-@@ -98,6 +98,25 @@ trait SharedSparkSessionBase
+@@ -98,6 +98,28 @@ trait SharedSparkSessionBase
// this rule may potentially block testing of other optimization rules such as
// ConstantPropagation etc.
.set(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, ConvertToLocalRelation.ruleName)
@@ -4502,11 +4504,14 @@ index 720b13b812e..388dad6b4f3 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // CometDriverPlugin installs Comet's cache serializer when
-+ // spark.comet.exec.inMemoryCache.enabled is on, as it is by default. These sessions do not
-+ // load the plugin, so install it here.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // Comet's in-memory cache is disabled by default, so turn it on to run these suites
++ // against Comet's cache format. CometDriverPlugin installs Comet's cache serializer when
++ // spark.comet.exec.inMemoryCache.enabled is on, but these sessions do not load the plugin,
++ // so install it here.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
conf.set(
StaticSQLConf.WAREHOUSE_PATH,
@@ -4629,10 +4634,10 @@ index b67370f6eb9..746b3974b29 100644
override def beforeEach(): Unit = {
super.beforeEach()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-index a394d0b7393..9c626b05ab2 100644
+index a394d0b7393..d59a3bf7f0f 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-@@ -53,24 +53,38 @@ object TestHive
+@@ -53,24 +53,41 @@ object TestHive
new SparkContext(
System.getProperty("spark.sql.test.master", "local[1]"),
"TestSQLContext",
@@ -4678,9 +4683,12 @@ index a394d0b7393..9c626b05ab2 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // As in SharedSparkSession: what CometDriverPlugin would install.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // As in SharedSparkSession: Comet's in-memory cache, and the serializer
++ // CometDriverPlugin would install for it.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
+
+ conf
diff --git a/dev/diffs/4.1.3.diff b/dev/diffs/4.1.3.diff
index d26af6706e..563b03f64b 100644
--- a/dev/diffs/4.1.3.diff
+++ b/dev/diffs/4.1.3.diff
@@ -78,7 +78,7 @@ index 370bfa4397f..ccb4398de58 100644
<dependency>
<groupId>org.apache.datasketches</groupId>
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
-index 01aa71047fe..06913c0656b 100644
+index 01aa71047fe..95953a0a668 100644
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@ -443,9 +443,11 @@ object SparkBuild extends PomBuild {
@@ -93,7 +93,7 @@ index 01aa71047fe..06913c0656b 100644
enable(HiveThriftServer.settings)(hiveThriftServer)
-@@ -1347,6 +1349,35 @@ object SQL {
+@@ -1347,6 +1349,37 @@ object SQL {
}
}
@@ -101,10 +101,11 @@ index 01aa71047fe..06913c0656b 100644
+ * Comet disables itself when its shuffle manager is not registered. When Comet is enabled, make
+ * that shuffle manager the default for every test SparkConf in the modules that have Comet on
+ * their classpath, so that suites which build their own SparkSession or SparkContext run Comet
-+ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's cache serializer
-+ * is made the default the same way. Spark keeps the first cache serializer it loads for the rest
-+ * of the JVM, so a suite that caches through a session of its own would otherwise leave the
-+ * suites after it in Spark's cache format.
++ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's in-memory
++ * cache, which is disabled by default, is turned on the same way, and its serializer made the
++ * default. Spark keeps the first cache serializer it loads for the rest of the JVM, so a suite
++ * that caches through a session of its own would otherwise leave the suites after it in Spark's
++ * cache format.
+ */
+object CometTestSettings {
+ private val cometEnabled =
@@ -113,6 +114,7 @@ index 01aa71047fe..06913c0656b 100644
+ private val defaults = Seq(
+ "spark.shuffle.manager" ->
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
++ "spark.comet.exec.inMemoryCache.enabled" -> "true",
+ "spark.sql.cache.serializer" ->
+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+
@@ -4851,10 +4853,10 @@ index f0f3f94b811..b7d18771314 100644
spark.internalCreateDataFrame(withoutFilters.execute(), schema)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-index 720b13b812e..388dad6b4f3 100644
+index 720b13b812e..34950a76bde 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-@@ -98,6 +98,25 @@ trait SharedSparkSessionBase
+@@ -98,6 +98,28 @@ trait SharedSparkSessionBase
// this rule may potentially block testing of other optimization rules such as
// ConstantPropagation etc.
.set(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, ConvertToLocalRelation.ruleName)
@@ -4871,11 +4873,14 @@ index 720b13b812e..388dad6b4f3 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // CometDriverPlugin installs Comet's cache serializer when
-+ // spark.comet.exec.inMemoryCache.enabled is on, as it is by default. These sessions do not
-+ // load the plugin, so install it here.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // Comet's in-memory cache is disabled by default, so turn it on to run these suites
++ // against Comet's cache format. CometDriverPlugin installs Comet's cache serializer when
++ // spark.comet.exec.inMemoryCache.enabled is on, but these sessions do not load the plugin,
++ // so install it here.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
conf.set(
StaticSQLConf.WAREHOUSE_PATH,
@@ -4998,10 +5003,10 @@ index b67370f6eb9..746b3974b29 100644
override def beforeEach(): Unit = {
super.beforeEach()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-index a394d0b7393..9c626b05ab2 100644
+index a394d0b7393..d59a3bf7f0f 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-@@ -53,24 +53,38 @@ object TestHive
+@@ -53,24 +53,41 @@ object TestHive
new SparkContext(
System.getProperty("spark.sql.test.master", "local[1]"),
"TestSQLContext",
@@ -5047,9 +5052,12 @@ index a394d0b7393..9c626b05ab2 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // As in SharedSparkSession: what CometDriverPlugin would install.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // As in SharedSparkSession: Comet's in-memory cache, and the serializer
++ // CometDriverPlugin would install for it.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
+
+ conf
diff --git a/dev/diffs/4.2.0.diff b/dev/diffs/4.2.0.diff
index c827261327..0365143338 100644
--- a/dev/diffs/4.2.0.diff
+++ b/dev/diffs/4.2.0.diff
@@ -78,7 +78,7 @@ index 46558134f41..7fa990be776 100644
<dependency>
<groupId>org.apache.datasketches</groupId>
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
-index 7c5e49f3a69..1091462f8ef 100644
+index 7c5e49f3a69..bfa0bce303d 100644
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@ -455,9 +455,11 @@ object SparkBuild extends PomBuild {
@@ -93,7 +93,7 @@ index 7c5e49f3a69..1091462f8ef 100644
enable(HiveThriftServer.settings)(hiveThriftServer)
-@@ -1426,6 +1428,35 @@ object SQL {
+@@ -1426,6 +1428,37 @@ object SQL {
}
}
@@ -101,10 +101,11 @@ index 7c5e49f3a69..1091462f8ef 100644
+ * Comet disables itself when its shuffle manager is not registered. When Comet is enabled, make
+ * that shuffle manager the default for every test SparkConf in the modules that have Comet on
+ * their classpath, so that suites which build their own SparkSession or SparkContext run Comet
-+ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's cache serializer
-+ * is made the default the same way. Spark keeps the first cache serializer it loads for the rest
-+ * of the JVM, so a suite that caches through a session of its own would otherwise leave the
-+ * suites after it in Spark's cache format.
++ * too, not just the ones that go through SharedSparkSession or TestHive. Comet's in-memory
++ * cache, which is disabled by default, is turned on the same way, and its serializer made the
++ * default. Spark keeps the first cache serializer it loads for the rest of the JVM, so a suite
++ * that caches through a session of its own would otherwise leave the suites after it in Spark's
++ * cache format.
+ */
+object CometTestSettings {
+ private val cometEnabled =
@@ -113,6 +114,7 @@ index 7c5e49f3a69..1091462f8ef 100644
+ private val defaults = Seq(
+ "spark.shuffle.manager" ->
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
++ "spark.comet.exec.inMemoryCache.enabled" -> "true",
+ "spark.sql.cache.serializer" ->
+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+
@@ -4997,10 +4999,10 @@ index e2c74533e7f..a12d55848ea 100644
val tblTargetName = "tbl_target"
val tblSourceQualified = s"default.$tblSourceName"
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-index fb26d3311eb..188b4d439a2 100644
+index fb26d3311eb..52b73da9aaf 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala
-@@ -102,6 +102,25 @@ trait SharedSparkSessionBase
+@@ -102,6 +102,28 @@ trait SharedSparkSessionBase
// this rule may potentially block testing of other optimization rules such as
// ConstantPropagation etc.
.set(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, ConvertToLocalRelation.ruleName)
@@ -5017,11 +5019,14 @@ index fb26d3311eb..188b4d439a2 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // CometDriverPlugin installs Comet's cache serializer when
-+ // spark.comet.exec.inMemoryCache.enabled is on, as it is by default. These sessions do not
-+ // load the plugin, so install it here.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // Comet's in-memory cache is disabled by default, so turn it on to run these suites
++ // against Comet's cache format. CometDriverPlugin installs Comet's cache serializer when
++ // spark.comet.exec.inMemoryCache.enabled is on, but these sessions do not load the plugin,
++ // so install it here.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
conf.set(
StaticSQLConf.WAREHOUSE_PATH,
@@ -5100,10 +5105,10 @@ index 59022deaed7..f9aeacb5a9b 100644
override def beforeEach(): Unit = {
super.beforeEach()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-index 8e7ff526a95..518e1bae289 100644
+index 8e7ff526a95..ec1643e53ee 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
-@@ -54,24 +54,38 @@ object TestHive
+@@ -54,24 +54,41 @@ object TestHive
new SparkContext(
System.getProperty("spark.sql.test.master", "local[1]"),
"TestSQLContext",
@@ -5149,9 +5154,12 @@ index 8e7ff526a95..518e1bae289 100644
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.comet.shuffle.enabled", "true")
+
-+ // As in SharedSparkSession: what CometDriverPlugin would install.
-+ conf.set("spark.sql.cache.serializer",
-+ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
++ // As in SharedSparkSession: Comet's in-memory cache, and the serializer
++ // CometDriverPlugin would install for it.
++ conf
++ .set("spark.comet.exec.inMemoryCache.enabled", "true")
++ .set("spark.sql.cache.serializer",
++ "org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
+ }
+
+ conf
diff --git a/docs/source/contributor-guide/spark-sql-tests.md b/docs/source/contributor-guide/spark-sql-tests.md
index 3918f69ffa..989cd85bf9 100644
--- a/docs/source/contributor-guide/spark-sql-tests.md
+++ b/docs/source/contributor-guide/spark-sql-tests.md
@@ -32,9 +32,10 @@ Here is an overview of the changes that we need to make to Spark:
- Modify `project/SparkBuild.scala` so that, when Comet is enabled, the `sql` and `hive` test JVMs use
`CometShuffleManager` as the default shuffle manager. Comet disables itself when its shuffle manager is not
registered, so without this default the suites that build their own `SparkSession` or `SparkContext` would
- run without Comet. From Spark 3.5, where Comet's cache format is the default, the same JVMs also default
- `spark.sql.cache.serializer` to Comet's `ArrowCachedBatchSerializer`, because Spark keeps the first cache
- serializer it loads for the rest of the JVM
+ run without Comet. From Spark 3.5, the same JVMs also turn on `spark.comet.exec.inMemoryCache.enabled` and
+ default `spark.sql.cache.serializer` to Comet's `ArrowCachedBatchSerializer`, so that Spark's suites test
+ Comet's cache format although it is disabled by default. The serializer has to be a default for the whole JVM
+ because Spark keeps the first cache serializer it loads for the rest of the JVM
Here are the steps involved in running the Spark SQL tests with Comet, using Spark 3.4.3 for this example.
diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md
index 6194f964cd..2f1fb6439f 100644
--- a/docs/source/user-guide/latest/compatibility/operators.md
+++ b/docs/source/user-guide/latest/compatibility/operators.md
@@ -37,12 +37,11 @@ infer the schema from.
## In-Memory Cache
Comet can store cached relations (`df.cache()`, `CACHE TABLE`) in Arrow format and scan them
-natively. This is enabled by default from Spark 3.5 but not on Spark 3.4; see
-[In-Memory Cache](../in-memory-cache.md) for how to turn it off or on. Comet does not replace a
-`spark.sql.cache.serializer` that the application has already set. Relations whose schema Comet's
-Arrow writer does not support are cached in Spark's default format, and their scans fall back to
-Spark. Reads that feed Spark operators rather than Comet operators can be slower than Spark's
-cache.
+natively. This is experimental and disabled by default; see [In-Memory Cache](../in-memory-cache.md)
+for how to enable it. Comet does not replace a `spark.sql.cache.serializer` that the application
+has already set. Relations whose schema Comet's Arrow writer does not support are cached in
+Spark's default format, and their scans fall back to Spark. Reads that feed Spark operators rather
+than Comet operators can be slower than Spark's cache.
With Kryo and `spark.kryo.registrationRequired=true`, Comet needs its Kryo registrator whether or
not the cache is enabled; see [Kryo serialization](../installation.md#kryo-serialization).
diff --git a/docs/source/user-guide/latest/in-memory-cache.md b/docs/source/user-guide/latest/in-memory-cache.md
index f0c2cd6a31..9a15ee39c3 100644
--- a/docs/source/user-guide/latest/in-memory-cache.md
+++ b/docs/source/user-guide/latest/in-memory-cache.md
@@ -25,19 +25,18 @@ format, which Comet operators cannot read. Under Comet's default settings the op
cache scan then run on Spark. With `spark.comet.convert.inMemoryCache.enabled`, a
`CometSparkColumnarToColumnar` above the scan converts each batch for Comet operators instead.
-This feature is **enabled by default** from Spark 3.5. To turn it off, set the config at startup,
-alongside the rest of Comet's configuration:
+This feature is **experimental and disabled by default**; see [Limitations](#limitations). Turn it
+on at startup, alongside the rest of Comet's configuration:
```shell
$SPARK_HOME/bin/spark-shell \
... \
- --conf spark.comet.exec.inMemoryCache.enabled=false
+ --conf spark.comet.exec.inMemoryCache.enabled=true
```
-On Spark 3.4 it is disabled by default, and setting the config to `true` turns it on. Spark 3.4 has
-no hook for the rule through which Comet lets AQE coalesce the shuffle partitions of a union that
-Comet runs, so a union of a shuffle and a relation cached in Comet's format keeps every shuffle
-partition there.
+On Spark 3.4, a union of a shuffle and a relation cached in Comet's format keeps every shuffle
+partition. Spark 3.4 has no hook for the rule through which Comet lets AQE coalesce the shuffle
+partitions of a union that Comet runs.
It has to be set before the `SparkContext` starts. Comet's driver plugin chooses
`spark.sql.cache.serializer` once, while the context is initializing, so an application keeps the
@@ -129,7 +128,7 @@ nowhere to record either that a column is dictionary encoded or the dictionary i
| Config | Default | Description |
| ------------------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `spark.comet.exec.inMemoryCache.enabled` | `true` | Whether to store and scan Spark's in-memory cache in Comet's format. Read at startup. Defaults to `false` on Spark 3.4. |
+| `spark.comet.exec.inMemoryCache.enabled` | `false` | Whether to store and scan Spark's in-memory cache in Comet's format. Read at startup. |
| `spark.comet.exec.inMemoryCache.compression.codec` | `zstd` | Arrow IPC compression codec for cached data: `zstd`, `lz4` or `none`. Affects newly cached data only: a batch records the codec it was written with. |
| `spark.comet.exec.inMemoryCache.compression.zstd.level` | `1` | Compression level when the codec is `zstd`. Ignored otherwise. |
@@ -268,6 +267,15 @@ Spark's cached batch too.
## Limitations
+A relation cached after a sort, such as `sortWithinPartitions`, reports that order to the queries
+that read it, and Spark removes the sorts it makes redundant. Some native joins over such a
+relation do not keep the order, so partitions above them can come back unsorted, and an operator
+that relies on the order, such as a sort-merge join, can return wrong rows. These cases are tracked
+in [#6787](https://github.com/apache/datafusion-comet/issues/6787),
+[#6898](https://github.com/apache/datafusion-comet/issues/6898) and
+[#6922](https://github.com/apache/datafusion-comet/issues/6922), and are why the feature is
+disabled by default.
+
Reads that feed **Spark** operators rather than Comet ones can be slower than Spark's own cache
format, because every cached batch is decoded from Arrow before Spark reads it. How a Spark
operator reads a relation cached in Comet's format depends on the scan below it:
diff --git a/docs/source/user-guide/latest/migration-guide.md b/docs/source/user-guide/latest/migration-guide.md
index a3de441b39..c0aadb4cb9 100644
--- a/docs/source/user-guide/latest/migration-guide.md
+++ b/docs/source/user-guide/latest/migration-guide.md
@@ -62,31 +62,6 @@ key is removed.
Comet `1.2.0` makes no behavior changes that need a `spark.comet.legacy.*` key. The changes below
need none either, but check whether any of them applies to your deployment.
-### In-Memory Cache Enabled by Default
-
-`spark.comet.exec.inMemoryCache.enabled` now defaults to `true` on Spark 3.5 and later. There, an
-application that loads `CometPlugin` now stores what it caches with `CACHE TABLE`, `df.cache()` or
-`df.persist()` in Comet's Arrow format instead of Spark's, and Comet scans it natively. On Spark 3.4
-the default stays `false`; see [In-Memory Cache](in-memory-cache.md) for why. The format does not
-change query results, but it can change performance. Spark's own cache scan can read Comet's format
-more slowly than Spark's, which matters when a session turns Comet or its native execution off
-after caching, and Comet records a fallback reason on such a scan. See
-[In-Memory Cache](in-memory-cache.md#limitations).
-
-The format is chosen once, when the application starts. To keep Spark's format, set
-`spark.comet.exec.inMemoryCache.enabled=false` then. Comet also keeps Spark's format without that
-setting when the application:
-
-- starts with `spark.comet.enabled` or `spark.comet.exec.enabled` set to `false`.
-- leaves Comet shuffle enabled without one of Comet's shuffle managers, so that Comet disables
- itself.
-- uses Kryo with `spark.kryo.registrationRequired=true` and has not registered Comet's cached
- batch, because Kryo would reject it. To use Comet's format, register Comet's classes with
- `spark.kryo.registrator=org.apache.comet.CometKryoRegistrator`; see
- [Kryo](in-memory-cache.md#kryo).
-
-An application that sets `spark.sql.cache.serializer` itself keeps the serializer it chose.
-
### Deprecated and Removed Settings
Using `spark.comet.sparkToColumnar.enabled` and `spark.comet.sparkToColumnar.supportedOperatorList`
diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md
index 9a0a430fc8..ccaa8fbf2c 100644
--- a/docs/source/user-guide/latest/operators.md
+++ b/docs/source/user-guide/latest/operators.md
@@ -66,7 +66,7 @@ omitted from the tables below and may be reconsidered based on demand:
| `LocalTableScanExec` | ⚠️ | Disabled by default; there is no acceleration advantage and this operator is typically only used in test code. Can be opted into via config ([#4393](https://github.com/apache/datafusion-comet/pull/4393)). |
| `EmptyRelationExec` | ✅ | Spark 4.0 and later. See [Empty Relations](compatibility/operators.md#empty-relations) for native-input support, including native Parquet writes. |
| `RangeExec` | ⚠️ | Disabled by default. Set `spark.comet.exec.range.enabled=true` to generate the rows of `spark.range` and SQL `range()` in native code, so the operators above them run natively. It can be slower than Spark when those operators are only cheap expressions, such as a filter, which Spark compiles together with the range into one loop. Ranges whose arithmetic overflows the `Long` range fall back to Spark. |
-| `InMemoryTableScanExec` | ✅ | Enabled by default from Spark 3.5. Comet installs its Arrow cache serializer as the application starts if `spark.comet.exec.inMemoryCache.enabled` is true, its default except on Spark 3.4. Relations with unsupported column types stay in Spark's cache format and fall back. See [In-Memory Cache](in-memory-cache.md). |
+| `InMemoryTableScanExec` | ⚠️ | Experimental, disabled by default. Set `spark.comet.exec.inMemoryCache.enabled=true` before the application starts so Comet installs its Arrow cache serializer. Relations with unsupported column types stay in Spark's cache format and fall back. See [In-Memory Cache](in-memory-cache.md). |
## Projection and filtering
diff --git a/docs/source/user-guide/latest/understanding-comet-plans.md b/docs/source/user-guide/latest/understanding-comet-plans.md
index 70d12b041a..f77cd1b516 100644
--- a/docs/source/user-guide/latest/understanding-comet-plans.md
+++ b/docs/source/user-guide/latest/understanding-comet-plans.md
@@ -298,7 +298,7 @@ by role. Names match what is shown in the plan output.
| `CometNativeScan` | Parquet scan that runs entirely in Rust via DataFusion. |
| `CometIcebergNativeScan` | Iceberg Parquet scan that runs entirely in Rust via DataFusion. |
| `CometCsvNativeScan` | CSV scan that runs entirely in Rust via DataFusion (experimental). |
-| `CometInMemoryTableScan` | JVM-side scan of a table cached in Comet's Arrow format (enabled by default from Spark 3.5). |
+| `CometInMemoryTableScan` | JVM-side scan of a table cached in Comet's Arrow format (experimental, disabled by default). |
| `CometLocalTableScan` | JVM-side scan of a `LocalTableScanExec` (disabled by default). |
| `CometEmptyRelation` | Empty native input that replaces `EmptyRelationExec` (Spark 4.0 and later). |
| `CometRange` | Generates the rows of `spark.range` and SQL `range()` in native code, replacing `RangeExec` (disabled by default). |
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index f844f7e8f8..a54fe24c9d 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -378,9 +378,9 @@ object CometConf extends ShimCometConf {
.category(CATEGORY_EXEC)
.doc(
"Whether to enable Comet native scans and fused Spark reads of in-memory cached tables. " +
- "Requires spark.comet.enabled=true. Defaults to true from Spark 3.5 and to false on " +
- "Spark 3.4, where AQE cannot coalesce the shuffle partitions of a union that Comet " +
- "runs with a cached relation in one branch. At startup, this setting also decides " +
+ "Requires spark.comet.enabled=true. On Spark 3.4, AQE cannot coalesce the shuffle " +
+ "partitions of a union that Comet runs with a cached relation in one branch. At " +
+ "startup, this setting also decides " +
"whether CometDriverPlugin installs Comet's cache serializer, which stores cached " +
"data in Arrow format. The plugin installs it only if spark.comet.enabled and " +
"spark.comet.exec.enabled are also enabled at startup, and only with one of Comet's " +
@@ -403,9 +403,7 @@ object CometConf extends ShimCometConf {
"soon as it is serialized, including the disk half of the default " +
"MEMORY_AND_DISK storage level.")
.booleanConf
- // CometCoalesceShufflePartitions needs a query-stage optimizer rule hook, which Spark 3.4
- // lacks.
- .createWithDefault(CometSparkSessionExtensions.isSpark35Plus)
+ .createWithDefault(false)
val COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_CODEC: ConfigEntry[String] =
conf("spark.comet.exec.inMemoryCache.compression.codec")
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
index b657cf1526..4dd68e6b5f 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
@@ -96,6 +96,7 @@ class CometInMemoryCacheSuite extends CometTestBase {
conf.set("spark.comet.exec.enabled", "true")
conf.set("spark.comet.exec.onHeap.enabled", "true")
conf.set("spark.comet.metrics.enabled", "true")
+ conf.set(CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key, "true")
conf.set(
"spark.sql.cache.serializer",
"org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer")
@@ -1491,10 +1492,9 @@ class CometInMemoryCacheSuite extends CometTestBase {
extraConfs.containsKey(serializerKey)
}
- test("Comet's cache format is the default from Spark 3.5 and opt-in on Spark 3.4") {
- // Spark 3.4 has no hook for CometCoalesceShufflePartitions, so AQE could not coalesce the
- // shuffled branch of a Comet union whose other branch reads a cache in Comet's format.
- assert(CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.defaultValue.get == isSpark35Plus)
+ test("Comet's cache format is opt-in") {
+ // Native joins can lose the order a sorted cached relation reports (#6787, #6898, #6922).
+ assert(!CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.defaultValue.get)
}
test("Comet plugin installs its cache serializer only if Comet can scan the cache natively") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org