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/04 19:57:02 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6602-55c707051fb6b78c5631e7f5da3dfe2975750f39 created (now bdbc4605d6)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6602-55c707051fb6b78c5631e7f5da3dfe2975750f39
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at bdbc4605d6 feat: give each default Spark-to-Arrow conversion its own spark.comet.convert config (#6602)
This branch includes the following new commits:
new bdbc4605d6 feat: give each default Spark-to-Arrow conversion its own spark.comet.convert config (#6602)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: feat: give each default Spark-to-Arrow conversion its own spark.comet.convert config (#6602)
Posted by gi...@apache.org.
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-6602-55c707051fb6b78c5631e7f5da3dfe2975750f39
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit bdbc4605d6f5814d9f1c9809c6f04d7419d67cc3
Author: Andy Grove <ag...@apache.org>
AuthorDate: Sun Oct 4 19:56:38 2026 +0000
feat: give each default Spark-to-Arrow conversion its own spark.comet.convert config (#6602)
Add spark.comet.convert.range.enabled, spark.comet.convert.inMemoryCache.enabled,
spark.comet.convert.rdd.enabled and spark.comet.convert.oneRowRelation.enabled for the four
operators that spark.comet.sparkToColumnar.supportedOperatorList named by default. All are off
by default.
The list now defaults to empty and stays the way to convert other leaf operators. Using
spark.comet.sparkToColumnar.enabled and the list for the four operators above is deprecated: it
keeps working as before and logs a one-time warning naming the new config. The OneRowRelation
config matches both the RDDScanExec Spark plans before 4.1 and the OneRowRelationExec it plans
from 4.1.
CometTestBase and the suites that set the old keys move to the new ones.
---
docs/source/user-guide/latest/datasources.md | 25 +++-
docs/source/user-guide/latest/migration-guide.md | 23 ++++
.../main/scala/org/apache/comet/CometConf.scala | 57 ++++++++--
.../org/apache/comet/rules/CometExecRule.scala | 54 ++++++++-
.../rules/EliminateRedundantTransitions.scala | 14 +--
.../sql/comet/shims/ShimCometOneRowRelation.scala | 34 ++++++
.../sql/comet/shims/ShimCometOneRowRelation.scala | 34 ++++++
.../sql/comet/shims/ShimCometOneRowRelation.scala | 31 +++++
.../apache/comet/CometArrayExpressionSuite.scala | 5 -
.../comet/CometIcebergWriteDetectionSuite.scala | 8 +-
.../org/apache/comet/CometMapExpressionSuite.scala | 1 -
.../org/apache/comet/exec/CometExecSuite.scala | 11 +-
.../comet/exec/CometInMemoryCacheKryoSuite.scala | 3 +-
.../exec/CometInMemoryCachePruningSuite.scala | 3 +-
.../comet/exec/CometInMemoryCacheSuite.scala | 126 +++++++++------------
.../apache/comet/exec/CometRangeExecSuite.scala | 10 +-
.../apache/comet/rules/CometExecRuleSuite.scala | 113 +++++++++++++++++-
.../scala/org/apache/spark/sql/CometTestBase.scala | 14 ++-
.../benchmark/CometInMemoryCacheBenchmark.scala | 3 +-
.../spark/sql/benchmark/CometRangeBenchmark.scala | 4 +-
.../comet/exec/CometEmptyRelationExecSuite.scala | 6 +-
.../CometEmptyRelationParquetWriterSuite.scala | 2 +-
22 files changed, 448 insertions(+), 133 deletions(-)
diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index c6d9debaf0..fcfb3398de 100644
--- a/docs/source/user-guide/latest/datasources.md
+++ b/docs/source/user-guide/latest/datasources.md
@@ -49,6 +49,25 @@ converted into Arrow format, allowing the Comet pipeline to take over after that
Comet does not provide a Rust-based JSON scan, but when `spark.comet.convert.json.enabled` is enabled, data is immediately
converted into Arrow format, allowing the Comet pipeline to take over after that.
+### Other Spark inputs
+
+Comet can also convert the output of these Spark inputs to Arrow format, so that the operators
+above them can run in Comet. Each conversion is off by default.
+
+- `spark.comet.convert.range.enabled`: `spark.range` and SQL `range()`, for ranges that Comet does
+ not generate natively with `spark.comet.exec.range.enabled`.
+- `spark.comet.convert.inMemoryCache.enabled`: in-memory cached tables that Comet's native cache
+ scan does not read, such as tables cached in Spark's default format.
+- `spark.comet.convert.rdd.enabled`: a DataFrame created from an RDD of rows, for example with
+ `spark.createDataFrame(rdd, schema)`.
+- `spark.comet.convert.oneRowRelation.enabled`: the single row that a query without a `FROM`
+ clause, such as `SELECT 1`, reads.
+
+To convert any other leaf operator, such as the scan of a Data Source V2 connector or of a file
+format other than Parquet, JSON and CSV, set `spark.comet.sparkToColumnar.enabled=true` and name the
+operator in `spark.comet.sparkToColumnar.supportedOperatorList` by its Spark class name without the
+`Exec` suffix, such as `BatchScan` or `FileSourceScan`.
+
### Spark-to-Comet conversion types
Spark-to-Comet conversion supports `ARRAY<STRING>` and `MAP<STRING,STRING>` with binary
@@ -58,10 +77,8 @@ This applies to Spark row and columnar inputs when conversion is enabled for the
Other array element types, other map key/value types, nested collections, and non-binary
string collations remain unsupported at this conversion boundary. Source defaults are unchanged.
-This includes row-backed `ExistingRDD` inputs when
-`spark.comet.sparkToColumnar.enabled=true` and
-`spark.comet.sparkToColumnar.supportedOperatorList` includes `RDDScan`. Spark still produces
-the RDD rows; conversion lets eligible downstream operators execute in Comet.
+This includes row-backed `ExistingRDD` inputs when `spark.comet.convert.rdd.enabled=true`. Spark
+still produces the RDD rows; conversion lets eligible downstream operators execute in Comet.
## Data Catalogs
diff --git a/docs/source/user-guide/latest/migration-guide.md b/docs/source/user-guide/latest/migration-guide.md
index b44dbf9cd7..0a9731a2fa 100644
--- a/docs/source/user-guide/latest/migration-guide.md
+++ b/docs/source/user-guide/latest/migration-guide.md
@@ -57,6 +57,29 @@ 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.2.0
+
+### Deprecated and Removed Settings
+
+Using `spark.comet.sparkToColumnar.enabled` and `spark.comet.sparkToColumnar.supportedOperatorList`
+to convert ranges, in-memory cached tables, RDDs and queries without a `FROM` clause to Arrow is
+deprecated, and will stop working in a future major release. Each of these now has a config of its
+own:
+
+| Operator in the list | Config |
+| -------------------- | -------------------------------------------- |
+| `Range` | `spark.comet.convert.range.enabled` |
+| `InMemoryTableScan` | `spark.comet.convert.inMemoryCache.enabled` |
+| `RDDScan` | `spark.comet.convert.rdd.enabled` |
+| `OneRowRelation` | `spark.comet.convert.oneRowRelation.enabled` |
+
+`spark.comet.sparkToColumnar.supportedOperatorList` now defaults to an empty list. When it is not
+set, `spark.comet.sparkToColumnar.enabled=true` still converts these four, and a list that names one
+of them still converts it, as before. In both cases the driver logs a warning that names the config
+to use instead. Before Spark 4.1, Spark plans a query without a `FROM` clause as an `RDDScan`, so on
+those versions `RDDScan` in the list also converted it. The list remains the way to convert other
+leaf operators, such as the scan of a Data Source V2 connector.
+
## Upgrading to Comet 1.1.0
Comet `1.1.0` makes no behavior changes that need a `spark.comet.legacy.*` key. The changes below
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index ea58da4c02..5c588017ed 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -203,6 +203,42 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_CONVERT_FROM_RANGE_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.range.enabled")
+ .category(CATEGORY_EXEC)
+ .doc(
+ "When enabled, the rows of `spark.range` and SQL `range()` that Comet does not generate " +
+ "natively (see `spark.comet.exec.range.enabled`) will be converted to Arrow format.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.inMemoryCache.enabled")
+ .category(CATEGORY_EXEC)
+ .doc(
+ "When enabled, data from in-memory cached tables that Comet's native cache scan does " +
+ "not read (see `spark.comet.exec.inMemoryCache.enabled`), such as tables cached in " +
+ "Spark's default format, will be converted to Arrow format.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val COMET_CONVERT_FROM_RDD_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.rdd.enabled")
+ .category(CATEGORY_EXEC)
+ .doc("When enabled, data from a DataFrame created from an RDD of rows, for example with " +
+ "`spark.createDataFrame(rdd, schema)`, will be converted to Arrow format.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val COMET_CONVERT_FROM_ONE_ROW_RELATION_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.oneRowRelation.enabled")
+ .category(CATEGORY_EXEC)
+ .doc(
+ "When enabled, the single row that a query without a FROM clause, such as `SELECT 1`, " +
+ "reads will be converted to Arrow format.")
+ .booleanConf
+ .createWithDefault(false)
+
val COMET_EXEC_ENABLED: ConfigEntry[Boolean] = conf(s"$COMET_EXEC_CONFIG_PREFIX.enabled")
.category(CATEGORY_EXEC)
.doc(
@@ -1053,21 +1089,28 @@ object CometConf extends ShimCometConf {
val COMET_SPARK_TO_ARROW_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.sparkToColumnar.enabled")
.category(CATEGORY_EXEC)
- .doc("Whether to enable Spark to Arrow columnar conversion. When this is turned on, " +
- "Comet will convert operators in " +
- "`spark.comet.sparkToColumnar.supportedOperatorList` into Arrow columnar format before " +
- "processing.")
+ .doc("Whether to convert the output of the leaf operators named in " +
+ "`spark.comet.sparkToColumnar.supportedOperatorList` to Arrow format. Ranges, " +
+ "in-memory cached tables, RDDs and queries without a FROM clause have " +
+ "`spark.comet.convert` configs of their own. When the list is not set, this converts " +
+ "those four instead, which is deprecated and will be removed in a future major release.")
.booleanConf
.createWithDefault(false)
val COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST: ConfigEntry[Seq[String]] =
conf("spark.comet.sparkToColumnar.supportedOperatorList")
.category(CATEGORY_EXEC)
- .doc("A comma-separated list of operators that will be converted to Arrow columnar " +
- s"format when `${COMET_SPARK_TO_ARROW_ENABLED.key}` is true.")
+ .doc(
+ "A comma-separated list of leaf operators to convert to Arrow format when " +
+ s"`${COMET_SPARK_TO_ARROW_ENABLED.key}` is true, each named by its Spark class name " +
+ "without the `Exec` suffix. It is for operators that have no `spark.comet.convert` " +
+ "config of their own, such as `BatchScan` for a Data Source V2 connector, or " +
+ "`FileSourceScan` for file formats other than Parquet, JSON and CSV. Naming `Range`, " +
+ "`InMemoryTableScan`, `RDDScan` or `OneRowRelation` is deprecated and will stop " +
+ "working in a future major release. Use their `spark.comet.convert` configs instead.")
.stringConf
.toSequence
- .createWithDefault(Seq("Range,InMemoryTableScan,RDDScan,OneRowRelation"))
+ .createWithDefault(Nil)
val COMET_CASE_CONVERSION_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.caseConversion.enabled")
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
index 081de9b26d..fbba51a448 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
@@ -19,6 +19,8 @@
package org.apache.comet.rules
+import java.util.concurrent.ConcurrentHashMap
+
import scala.collection.mutable.ListBuffer
import scala.jdk.CollectionConverters._
@@ -32,7 +34,7 @@ import org.apache.spark.sql.catalyst.util.sideBySide
import org.apache.spark.sql.comet._
import org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer
import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec}
-import org.apache.spark.sql.comet.shims.ShimCometEmptyRelation
+import org.apache.spark.sql.comet.shims.{ShimCometEmptyRelation, ShimCometOneRowRelation}
import org.apache.spark.sql.comet.util.Utils
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, BroadcastQueryStageExec, LogicalQueryStage, QueryStageExec, ShuffleQueryStageExec}
@@ -53,7 +55,7 @@ import org.apache.spark.sql.execution.window.WindowExec
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
-import org.apache.comet.{CometConf, CometExplainInfo, ExtendedExplainInfo}
+import org.apache.comet.{CometConf, CometExplainInfo, ConfigEntry, ExtendedExplainInfo}
import org.apache.comet.CometConf.{COMET_SPARK_TO_ARROW_ENABLED, COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST}
import org.apache.comet.CometSparkSessionExtensions._
import org.apache.comet.rules.CometExecRule.allExecs
@@ -167,6 +169,17 @@ object CometExecRule {
*/
val SKIP_COMET_BROADCAST_TAG: org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit] =
org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit]("comet.skipCometBroadcast")
+
+ /**
+ * The operators `spark.comet.sparkToColumnar.supportedOperatorList` named by default, before
+ * each had a `spark.comet.convert` config of its own. `spark.comet.sparkToColumnar.enabled`
+ * still converts them when the list is not set, which is deprecated.
+ */
+ private val deprecatedSparkToArrowOperators =
+ Seq("Range", "InMemoryTableScan", "RDDScan", "OneRowRelation")
+
+ /** Keys of the `spark.comet.convert` configs whose deprecated alternative has been warned. */
+ private[rules] val warnedDeprecatedConversions = ConcurrentHashMap.newKeySet[String]()
}
/**
@@ -1285,6 +1298,14 @@ case class CometExecRule(session: SparkSession)
case _: ParquetScan => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
case _ => isSparkToArrowEnabled(conf, op)
}
+ case _: RangeExec =>
+ isConversionEnabled(conf, op, CometConf.COMET_CONVERT_FROM_RANGE_ENABLED)
+ case _: InMemoryTableScanExec =>
+ isConversionEnabled(conf, op, CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED)
+ case _ if ShimCometOneRowRelation.isOneRowRelation(op) =>
+ isConversionEnabled(conf, op, CometConf.COMET_CONVERT_FROM_ONE_ROW_RELATION_ENABLED)
+ case _: RDDScanExec =>
+ isConversionEnabled(conf, op, CometConf.COMET_CONVERT_FROM_RDD_ENABLED)
// other leaf nodes
case _: LeafExecNode =>
isSparkToArrowEnabled(conf, op)
@@ -1297,11 +1318,38 @@ case class CometExecRule(session: SparkSession)
}
}
+ /**
+ * Whether to convert a leaf operator that has a `spark.comet.convert` config of its own. The
+ * `spark.comet.sparkToColumnar` settings still convert it too, which is deprecated.
+ */
+ private def isConversionEnabled(
+ conf: SQLConf,
+ op: SparkPlan,
+ entry: ConfigEntry[Boolean]): Boolean = {
+ entry.get(conf) || {
+ val deprecated = isSparkToArrowEnabled(conf, op)
+ if (deprecated && CometExecRule.warnedDeprecatedConversions.add(entry.key)) {
+ logWarning(
+ s"Using ${COMET_SPARK_TO_ARROW_ENABLED.key} to convert " +
+ s"${Utils.getSimpleName(op.getClass)} to Arrow is deprecated and will stop " +
+ s"working in a future major release. Set ${entry.key}=true instead.")
+ }
+ deprecated
+ }
+ }
+
private def isSparkToArrowEnabled(conf: SQLConf, op: SparkPlan) = {
COMET_SPARK_TO_ARROW_ENABLED.get(conf) && {
val simpleClassName = Utils.getSimpleName(op.getClass)
val nodeName = simpleClassName.replaceAll("Exec$", "")
- COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.get(conf).contains(nodeName)
+ // Without the list, the switch still converts the operators the list named by default.
+ val operators =
+ if (conf.getConfString(COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key, null) == null) {
+ CometExecRule.deprecatedSparkToArrowOperators
+ } else {
+ COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.get(conf)
+ }
+ operators.contains(nodeName)
}
}
diff --git a/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala b/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
index dcaa22c957..4ef790e917 100644
--- a/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
@@ -52,12 +52,12 @@ import org.apache.comet.shims.ShimSQLConf
// `CometExec`. However, for certain operators such as `CometCollectLimitExec` which overrides
// `executeCollect`, the redundant `ColumnarToRowExec` makes the override ineffective.
//
-// Note about the second case: When `spark.comet.sparkToColumnar.enabled` is set, Comet will add
-// `CometSparkToColumnarExec` on top of row-based operators first, but the downstream operator
-// only takes row-based input as it's a vanilla Spark operator(as Comet cannot convert it for
-// various reasons) or Spark requests row-based output such as a `collect` call. Spark will adds
-// another `ColumnarToRowExec` on top of `CometSparkToColumnarExec`. In this case, the pair could
-// be removed.
+// Note about the second case: When a Spark-to-Arrow conversion config is set, such as
+// `spark.comet.convert.parquet.enabled`, Comet will add `CometSparkToColumnarExec` on top of
+// row-based operators first, but the downstream operator only takes row-based input as it's a
+// vanilla Spark operator(as Comet cannot convert it for various reasons) or Spark requests
+// row-based output such as a `collect` call. Spark will adds another `ColumnarToRowExec` on top
+// of `CometSparkToColumnarExec`. In this case, the pair could be removed.
case class EliminateRedundantTransitions(session: SparkSession)
extends Rule[SparkPlan]
with ShimCometMapInBatch
@@ -236,7 +236,7 @@ case class EliminateRedundantTransitions(session: SparkSession)
* a row child and no Arrow producer at all;
* - over a Spark-columnar source it keeps a `ColumnarToRowExec` but drops the Arrow bridge,
* so the write would be handed Spark `ColumnarVector`s where the FFI adapter requires
- * `CometVector`s. That shape is reachable whenever `spark.comet.sparkToColumnar.enabled`
+ * `CometVector`s. That shape is reachable whenever a Spark-to-Arrow conversion config
* admits the write's source: `CometSparkToColumnarExec.createExec` wraps it in a
* `CometScanWrapper` (a `CometNativeExec`, so `requiresNativeChildren` accepts it), and
* `CometExecRule` then unwraps the placeholder, leaving the bridge directly beneath the
diff --git a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala
new file mode 100644
index 0000000000..dbc533de68
--- /dev/null
+++ b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.comet.shims
+
+import org.apache.spark.sql.execution.{RDDScanExec, SparkPlan}
+
+object ShimCometOneRowRelation {
+
+ /**
+ * Whether `plan` produces the single row of a query without a FROM clause. Before Spark 4.1,
+ * Spark plans that as an `RDDScanExec` named after the logical `OneRowRelation`.
+ */
+ def isOneRowRelation(plan: SparkPlan): Boolean = plan match {
+ case scan: RDDScanExec => scan.name == "OneRowRelation"
+ case _ => false
+ }
+}
diff --git a/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala b/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala
new file mode 100644
index 0000000000..dbc533de68
--- /dev/null
+++ b/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.comet.shims
+
+import org.apache.spark.sql.execution.{RDDScanExec, SparkPlan}
+
+object ShimCometOneRowRelation {
+
+ /**
+ * Whether `plan` produces the single row of a query without a FROM clause. Before Spark 4.1,
+ * Spark plans that as an `RDDScanExec` named after the logical `OneRowRelation`.
+ */
+ def isOneRowRelation(plan: SparkPlan): Boolean = plan match {
+ case scan: RDDScanExec => scan.name == "OneRowRelation"
+ case _ => false
+ }
+}
diff --git a/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala b/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala
new file mode 100644
index 0000000000..35e643703f
--- /dev/null
+++ b/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/shims/ShimCometOneRowRelation.scala
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.comet.shims
+
+import org.apache.spark.sql.execution.{OneRowRelationExec, SparkPlan}
+
+object ShimCometOneRowRelation {
+
+ /**
+ * Whether `plan` produces the single row of a query without a FROM clause. From Spark 4.1,
+ * Spark plans that as a `OneRowRelationExec`.
+ */
+ def isOneRowRelation(plan: SparkPlan): Boolean = plan.isInstanceOf[OneRowRelationExec]
+}
diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
index 5aabafe402..a9e6f467df 100644
--- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
@@ -232,7 +232,6 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
val table = spark.read.parquet(filename)
table.createOrReplaceTempView("t1")
@@ -511,7 +510,6 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
withTempView("t1", "t2") {
val table = spark.read.parquet(filename)
@@ -1024,7 +1022,6 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
withTempView("t1", "t2") {
val table = spark.read.parquet(filename)
@@ -1120,7 +1117,6 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
withTempView("t1", "t2") {
val table = spark.read.parquet(filename)
@@ -1171,7 +1167,6 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
withTempView("t1", "t2") {
val table = spark.read.parquet(filename)
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
index 8034ef9675..6deebef57a 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
@@ -1189,10 +1189,10 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
* hand-built `CometIcebergWriteExec -> CometSparkToColumnarExec -> source` plan and returns the
* write's final child.
*
- * Hand-built rather than driven through SQL because the shape depends on
- * `spark.comet.sparkToColumnar.enabled` admitting the write's source operator, and the set of
- * admitted operators is itself configurable. What matters is the rule's behaviour at that
- * boundary, which this pins directly.
+ * Hand-built rather than driven through SQL because the shape depends on a Spark-to-Arrow
+ * conversion config admitting the write's source operator, and the set of admitted operators is
+ * itself configurable. What matters is the rule's behaviour at that boundary, which this pins
+ * directly.
*/
private def writeChildAfterTransitionRules(source: SparkPlan): SparkPlan = {
val write = CometIcebergWriteExec(
diff --git a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala
index 7bb7ba5f13..98ef07b96f 100644
--- a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala
@@ -283,7 +283,6 @@ class CometMapExpressionSuite extends CometTestBase {
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
val df = spark.read.parquet(filename)
df.createOrReplaceTempView("t1")
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
index 1dc0f8fc84..5b78b0badf 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -2140,7 +2140,7 @@ class CometExecSuite extends CometTestBase {
SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10485760",
SQLConf.SHUFFLE_PARTITIONS.key -> "4",
CometConf.COMET_SHUFFLE_MODE.key -> "native",
- CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "Range") {
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true") {
val df = sql("""
|WITH s AS (
| SELECT id % 64 AS k, SUM(id) AS v FROM range(0, 4096, 1, 4) GROUP BY id % 64
@@ -2384,7 +2384,6 @@ class CometExecSuite extends CometTestBase {
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
CometConf.COMET_SHUFFLE_MODE.key -> "native") {
withParquetTable((0 until 10).map(i => (i, i)), "t") {
@@ -3995,8 +3994,7 @@ class CometExecSuite extends CometTestBase {
CometConf.COMET_BATCH_SIZE.key -> "2",
CometConf.COMET_SHUFFLE_MODE.key -> "native",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "RDDScan") {
+ CometConf.COMET_CONVERT_FROM_RDD_ENABLED.key -> "true") {
withTempPath { dir =>
def rdd = spark.createDataFrame(spark.sparkContext.parallelize(rows, 1), schema)
if (sourceType != "rdd") {
@@ -4032,7 +4030,7 @@ class CometExecSuite extends CometTestBase {
assert(exchanges.nonEmpty && exchanges.forall(_.shuffleType == CometNativeShuffle))
checkSparkAnswer(query.limit(1))
withSQLConf(
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_RDD_ENABLED.key -> "false",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "false") {
val (_, disabled) = checkSparkAnswer(query)
assert(collect(disabled) { case c: CometSparkToColumnarExec => c }.isEmpty)
@@ -4306,7 +4304,6 @@ class CometExecSuite extends CometTestBase {
}
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
val table = spark.read.parquet(filename)
table.createOrReplaceTempView("t1")
@@ -4324,7 +4321,6 @@ class CometExecSuite extends CometTestBase {
withSQLConf(
SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
withTempPath { dir =>
val expected = 10000L
@@ -4537,7 +4533,6 @@ class CometExecSuite extends CometTestBase {
.parquet(path)
withSQLConf(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true",
SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
val df = spark.read.parquet(path).orderBy("ts")
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheKryoSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheKryoSuite.scala
index 4518f8ab2e..05b80605db 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheKryoSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheKryoSuite.scala
@@ -109,8 +109,7 @@ class CometInMemoryCacheKryoSuite extends CometTestBase {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
try {
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCachePruningSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCachePruningSuite.scala
index db8d5f263a..0f9464b89b 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCachePruningSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCachePruningSuite.scala
@@ -164,7 +164,8 @@ class CometInMemoryCachePruningSuite extends CometTestBase {
CometConf.COMET_BATCH_SIZE.key -> "4",
CometConf.COMET_SHUFFLE_JVM_BATCH_SIZE.key -> "4",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_RDD_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED.key -> "false",
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> (writer == "native Arrow").toString) {
spark.catalog.clearCache()
val source = fixture()
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 be8f07a812..ccaa951314 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
@@ -98,6 +98,13 @@ class CometInMemoryCacheSuite extends CometTestBase {
conf
}
+ /**
+ * `withSQLConf` that also turns on the Spark-to-Arrow conversions, which this suite's conf
+ * leaves off, unlike CometTestBase's.
+ */
+ private def withConversions(pairs: (String, String)*)(f: => Unit): Unit =
+ withSQLConf(pairs ++ sparkToArrowConversionConfs(enabled = true): _*)(f)
+
private def cachedBatchTypes(table: String): Array[String] = {
val cached = spark.sharedState.cacheManager.lookupCachedData(spark.table(table)).get
cached.cachedRepresentation.cacheBuilder.cachedColumnBuffers
@@ -122,13 +129,12 @@ class CometInMemoryCacheSuite extends CometTestBase {
// The tests below are ported from Spark 4.1.2's AdaptiveQueryExecSuite; see each source link.
private def withAQECache(f: => Unit): Unit = {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.CAN_CHANGE_CACHED_PLAN_OUTPUT_PARTITIONING.key -> "true",
SQLConf.SHUFFLE_PARTITIONS.key -> "3",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
try {
f
} finally {
@@ -297,12 +303,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("CometInMemoryTableScan over CometCachedBatch") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -390,7 +395,7 @@ class CometInMemoryCacheSuite extends CometTestBase {
// reason that the reporting for a query reading the cache must leave out.
withSQLConf(
CometConf.COMET_EXEC_RANGE_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false") {
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "false") {
spark.catalog.cacheTable("explain_cached_plan")
}
val df = spark.sql("SELECT value, count(*) FROM explain_cached_plan GROUP BY value")
@@ -475,12 +480,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache disabled keeps SparkToColumnar fallback path") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -497,12 +501,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
Array("org.apache.spark.sql.comet.execution.arrow.CometCachedBatch")))
}
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "false",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "false") {
val df = spark.sql("SELECT key, count(*) FROM comet_cache_disabled GROUP BY key")
checkAnswer(df, (0L until 1000L).map(i => Row(i, 1L)))
@@ -529,12 +532,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
"named_struct('a', id, 'a', id + 1) AS payload")
test("Comet cache serializer delegates unsupported types to Spark's cache format") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
for (column <- unsupportedForArrowCache) {
spark.catalog.clearCache()
@@ -632,12 +634,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache handles multi-partition cache") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -666,12 +667,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache handles empty cache") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -692,12 +692,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache supports projection-only read") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -730,12 +729,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache supports shuffle after cache read") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -913,12 +911,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache supports stats-based batch pruning") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true",
"spark.sql.inMemoryColumnarStorage.batchSize" -> "100") {
spark.catalog.clearCache()
@@ -996,12 +993,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
// rows. With pruning off, every cached row must be decoded.
def scanRowsFor(pruning: Boolean): (Long, Long) = {
var result: (Long, Long) = (0L, 0L)
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true",
"spark.sql.inMemoryColumnarStorage.batchSize" -> "100",
SQLConf.IN_MEMORY_PARTITION_PRUNING.key -> pruning.toString) {
@@ -1044,12 +1040,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
// spill it to the DiskStore like any other cached block. Pins that: nothing is held in memory,
// the bytes really do land on disk, every partition is cached, and the cache still reads back
// through the native scan.
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
spark
@@ -1096,12 +1091,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
// So both paths must write "UTC". This is a label only -- Spark stores timestamps as micros
// since the Unix epoch regardless of session timezone -- so values must be unaffected.
Seq("America/Los_Angeles", "Asia/Kolkata").foreach { sessionTz =>
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true",
SQLConf.SESSION_LOCAL_TIMEZONE.key -> sessionTz) {
spark.catalog.clearCache()
@@ -1318,12 +1312,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache supports empty projection scan") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -1346,12 +1339,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
private def withNativeCache(f: => Unit): Unit = {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
try f
finally spark.catalog.clearCache()
@@ -1587,12 +1579,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
test("Comet in-memory cache pruning handles NaN floating-point values") {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true",
"spark.sql.inMemoryColumnarStorage.batchSize" -> "2") {
spark.catalog.clearCache()
@@ -1674,8 +1665,8 @@ class CometInMemoryCacheSuite extends CometTestBase {
withSQLConf(
Seq(
CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false",
- SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") ++ extraConfs: _*) {
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") ++
+ sparkToArrowConversionConfs(enabled = false) ++ extraConfs: _*) {
spark.read.parquet(path.toString).createOrReplaceTempView(view)
val expected = uncachedSparkAnswer(s"SELECT * FROM $view")
@@ -1823,11 +1814,10 @@ class CometInMemoryCacheSuite extends CometTestBase {
private def withCachedProjection(view: String, columns: Seq[String], chunkSize: Option[Int])(
f: (org.apache.spark.sql.execution.columnar.InMemoryRelation, Array[CachedBatch]) => Unit)
: Unit = {
- withSQLConf(
+ withConversions(
Seq(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true") ++
chunkSize.map(CometConf.COMET_EXEC_IN_MEMORY_CACHE_CHUNK_SIZE.key -> _.toString): _*) {
@@ -1903,12 +1893,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
// -- the payload records no codec, so nothing is decompressed -- and shipped broken for a
// while because the only tests that ran were on the default codec.
Seq("none", "zstd").foreach { codec =>
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_CODEC.key -> codec,
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_CODEC.key -> codec) {
spark.catalog.clearCache()
val view = s"codec_cache_$codec"
@@ -1998,12 +1987,11 @@ class CometInMemoryCacheSuite extends CometTestBase {
assert(expectedNulls.length == 1000)
Seq("none", "zstd").foreach { codec =>
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_CODEC.key -> codec,
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_CODEC.key -> codec) {
spark.catalog.clearCache()
val view = s"null_cache_$codec"
@@ -2385,10 +2373,9 @@ class CometInMemoryCacheSuite extends CometTestBase {
// the whole cache schema, or to a single placeholder column -- makes the emitted batches
// disagree with the scan's declared output, which is wrong for any consumer that reads by
// ordinal instead of by row count. See the join regression below.
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true",
SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "true") {
spark.catalog.clearCache()
@@ -2423,10 +2410,9 @@ class CometInMemoryCacheSuite extends CometTestBase {
// An empty-output cache scan can feed a join, not only a count-style aggregate. A join reads
// its inputs by ordinal, so any column the scan emits beyond its declared output shifts the
// right side's positions and silently produces wrong results rather than failing.
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
val left = spark.range(10L, 13L).cache()
@@ -2461,11 +2447,10 @@ class CometInMemoryCacheSuite extends CometTestBase {
// batches reaching serializeBatches from a shuffle or broadcast still carry independent
// dictionary providers whose IDs collide, and re-encoding one with only the first column's
// provider cannot resolve the later columns' IDs.
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_SHUFFLE_MODE.key -> "jvm") {
spark.catalog.clearCache()
val first = spark
@@ -2494,15 +2479,14 @@ class CometInMemoryCacheSuite extends CometTestBase {
// CachedRDDBuilder.cachedColumnBuffers builds its RDD by executing the cached plan, so
// touching it during planning runs jobs before the outer query is even submitted. With an
// adaptively-cached relation that also finalizes the cached plan. EXPLAIN must launch nothing.
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
// Spark caches through a session with some configs forced off, and on 3.4 that list still
// includes AQE itself, so the cached plan comes back non-adaptive and there is nothing to
// finalize. This conf is what decides that list; 3.5 defaults it on, and 4.0 stopped
// disabling AQE either way. Setting it keeps the relation adaptive on every version.
SQLConf.CAN_CHANGE_CACHED_PLAN_OUTPUT_PARTITIONING.key -> "true",
- CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
spark.catalog.clearCache()
val cached = spark.range(100).repartition(2).cache()
@@ -2741,11 +2725,10 @@ class CometInMemoryCacheSuite extends CometTestBase {
* dictionary-encoded columns, which the writer has to decode before storing them.
*/
private def withDictionaryCache(f: InMemoryRelation => Unit): Unit = {
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_SHUFFLE_MODE.key -> "jvm") {
spark.catalog.clearCache()
spark
@@ -2811,11 +2794,10 @@ class CometInMemoryCacheSuite extends CometTestBase {
// past it and leaves in place the expression IDs of whichever occurrence of the relation
// produced it. sameResult is what exchange and broadcast reuse are keyed on, so two
// equivalent scans that compare unequal make a query shuffle and aggregate one cache twice.
- withSQLConf(
+ withConversions(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true",
- CometConf.COMET_SHUFFLE_MODE.key -> "jvm",
- "spark.comet.sparkToColumnar.enabled" -> "true") {
+ CometConf.COMET_SHUFFLE_MODE.key -> "jvm") {
spark.catalog.clearCache()
spark
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometRangeExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometRangeExecSuite.scala
index ba9f5a68d7..15d13ced8e 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometRangeExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometRangeExecSuite.scala
@@ -32,10 +32,10 @@ import org.apache.comet.CometConf
class CometRangeExecSuite extends CometTestBase {
- // CometTestBase enables the Spark-to-Arrow conversion, which also applies to Range. Use its
- // production default so these tests see CometRangeExec on its own.
+ // CometTestBase enables the Spark-to-Arrow conversions, including the one for Range. Use their
+ // production defaults so these tests see CometRangeExec on its own.
override protected def sparkConf: SparkConf =
- super.sparkConf.remove(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key)
+ super.sparkConf.setAll(sparkToArrowConversionConfs(enabled = false))
/** Defines a test that runs with `CometRangeExec` enabled. */
private def rangeTest(name: String)(f: => Unit): Unit =
@@ -138,7 +138,7 @@ class CometRangeExecSuite extends CometTestBase {
spark.range(Long.MinValue, Long.MaxValue, 1, 4).selectExpr("id + 1"),
reason)
// A range CometRangeExec declines can still use the Spark-to-Arrow conversion.
- withSQLConf(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true") {
+ withSQLConf(CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true") {
val (_, plan) = checkSparkAnswer(overflowing)
assert(collect(plan) { case c: CometSparkToColumnarExec => c }.nonEmpty, plan)
}
@@ -217,7 +217,7 @@ class CometRangeExecSuite extends CometTestBase {
}
rangeTest("takes precedence over the Spark-to-Arrow conversion") {
- withSQLConf(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true") {
+ withSQLConf(CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true") {
val plan = checkCometRange(spark.range(0, 100, 1, 2).selectExpr("id + 1"))
assert(collect(plan) { case c: CometSparkToColumnarExec => c }.isEmpty, plan)
}
diff --git a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala
index 6e63f99830..83ab1fa0f8 100644
--- a/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala
@@ -46,7 +46,7 @@ import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ShuffleEx
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataTypes, DoubleType, FloatType, StructField, StructType}
-import org.apache.comet.{CometConf, CometCoverageStats, CometExplainInfo, CometSparkSessionExtensions, ExtendedExplainInfo}
+import org.apache.comet.{CometConf, CometCoverageStats, CometExplainInfo, CometSparkSessionExtensions, ConfigEntry, ExtendedExplainInfo}
import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus, withFallbackReason}
import org.apache.comet.serde.{CometAggregateExpressionSerde, Compatible, ExprOuterClass, OperatorOuterClass, QueryPlanSerde, Unsupported}
import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator}
@@ -273,7 +273,7 @@ class CometExecRuleSuite extends CometTestBase {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "Range") {
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true") {
val originalTags =
Seq(Some(SparkPlan.LOGICAL_PLAN_TAG), Some(SparkPlan.LOGICAL_PLAN_INHERITED_TAG), None)
originalTags.foreach { originalTag =>
@@ -304,7 +304,7 @@ class CometExecRuleSuite extends CometTestBase {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "Range") {
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true") {
val originalTags =
Seq(Some(SparkPlan.LOGICAL_PLAN_TAG), Some(SparkPlan.LOGICAL_PLAN_INHERITED_TAG), None)
for (originalTag <- originalTags; hasDirectLink <- Seq(false, true)) {
@@ -344,7 +344,7 @@ class CometExecRuleSuite extends CometTestBase {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "Range")(f)
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true")(f)
test("CometExecRule points an AQE-reused aggregate's stale Scan at its ShuffleScan stage") {
withAdaptiveAggregateConf {
@@ -750,8 +750,7 @@ class CometExecRuleSuite extends CometTestBase {
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
"spark.sql.optimizer.inSetConversionThreshold" -> "100",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "Range",
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true",
CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "false",
CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false",
CometConf.COMET_STRICT_FALLBACK_REASONS.key -> strict.toString) ++
@@ -2181,4 +2180,106 @@ class CometExecRuleSuite extends CometTestBase {
}
}
+ /**
+ * The leaf operators that `CometExecRule` converts to Arrow in the plan of `query`, with every
+ * Spark-to-Arrow conversion off apart from what `confs` turns on.
+ */
+ private def convertedLeaves(query: String, confs: (String, String)*): Seq[SparkPlan] = {
+ var leaves = Seq.empty[SparkPlan]
+ withSQLConf(
+ Seq(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
+ CometConf.COMET_EXEC_RANGE_ENABLED.key -> "false",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "false") ++
+ sparkToArrowConversionConfs(enabled = false) ++ confs: _*) {
+ leaves = applyCometExecRule(createSparkPlan(spark, query)).collect {
+ case c: CometSparkToColumnarExec => c.child
+ }
+ }
+ leaves
+ }
+
+ /** Runs `f` with a query over each leaf operator that has a `spark.comet.convert` config. */
+ private def withConversionQueries(f: Seq[(ConfigEntry[Boolean], String)] => Unit): Unit = {
+ withTempView("rdd_input", "cached_input") {
+ val schema = StructType(Seq(StructField("a", DataTypes.IntegerType)))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(Seq(Row(1), Row(2)), 1), schema)
+ .createOrReplaceTempView("rdd_input")
+ spark.range(10).selectExpr("CAST(id AS INT) AS a").createOrReplaceTempView("cached_input")
+ spark.catalog.cacheTable("cached_input")
+ try {
+ f(
+ Seq(
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED -> "SELECT id + 1 FROM range(10)",
+ CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED ->
+ "SELECT a + 1 FROM cached_input",
+ CometConf.COMET_CONVERT_FROM_RDD_ENABLED -> "SELECT a + 1 FROM rdd_input",
+ // Spark plans this as an RDDScanExec before 4.1, so it also checks that the RDD
+ // config leaves it alone there.
+ CometConf.COMET_CONVERT_FROM_ONE_ROW_RELATION_ENABLED -> "SELECT 1 AS a"))
+ } finally {
+ spark.catalog.uncacheTable("cached_input")
+ }
+ }
+ }
+
+ test("each spark.comet.convert config converts only its own leaf operator") {
+ withConversionQueries { queries =>
+ for ((entry, query) <- queries; (enabled, _) <- queries) {
+ withClue(s"${enabled.key}=true, $query: ") {
+ assert(convertedLeaves(query, enabled.key -> "true").nonEmpty == (entry eq enabled))
+ }
+ }
+ }
+ }
+
+ test("the deprecated sparkToColumnar settings still convert what they converted before") {
+ val switch = CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true"
+ val list = CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key
+ withConversionQueries { queries =>
+ // Without a list, the switch converts the operators the list used to name by default.
+ queries.foreach { case (_, query) =>
+ assert(convertedLeaves(query, switch).nonEmpty, query)
+ }
+ // A list replaces them.
+ val range = queries.head._2
+ assert(convertedLeaves(range, switch, list -> "RDDScan").isEmpty)
+ assert(convertedLeaves(range, switch, list -> "Range").nonEmpty)
+ }
+ // An operator without a config of its own is converted only when the list names it.
+ val values = "SELECT a FROM VALUES (1), (2) AS t(a)"
+ assert(convertedLeaves(values, switch).isEmpty)
+ assert(convertedLeaves(values, switch, list -> "LocalTableScan").nonEmpty)
+ assert(convertedLeaves(values, list -> "LocalTableScan").isEmpty)
+ }
+
+ test("the deprecated sparkToColumnar settings warn once for each config they stand in for") {
+ val switch = CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true"
+ val range = "SELECT id + 1 FROM range(10)"
+ CometExecRule.warnedDeprecatedConversions.clear()
+ val appender = new LogAppender("deprecated Spark-to-Arrow settings")
+ withLogAppender(appender, Seq(classOf[CometExecRule].getName), Some(Level.WARN)) {
+ assert(convertedLeaves(range, switch).nonEmpty)
+ assert(convertedLeaves(range, switch).nonEmpty)
+ // With its own config on, the range does not need the deprecated settings.
+ assert(
+ convertedLeaves(
+ range,
+ switch,
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true").nonEmpty)
+ // Converting an operator without a config of its own is not deprecated.
+ assert(convertedLeaves(
+ "SELECT a FROM VALUES (1), (2) AS t(a)",
+ switch,
+ CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key -> "LocalTableScan").nonEmpty)
+ }
+ val warnings = appender.loggingEvents
+ .map(_.getMessage.getFormattedMessage)
+ .filter(_.contains(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key))
+ assert(warnings.size == 1, warnings)
+ assert(warnings.head.contains(s"${CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key}=true"))
+ }
+
}
diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
index 8903f43769..28ea86efad 100644
--- a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
@@ -76,6 +76,18 @@ abstract class CometTestBase
assert(reused.nonEmpty, s"$clue:\n$plan")
}
+ /**
+ * Sets each Spark-to-Arrow conversion that [[sparkConf]] turns on, which are off by default, to
+ * `enabled`. These are the conversions `spark.comet.sparkToColumnar.enabled` stood for before
+ * each had a config of its own.
+ */
+ protected def sparkToArrowConversionConfs(enabled: Boolean): Seq[(String, String)] =
+ Seq(
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED,
+ CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED,
+ CometConf.COMET_CONVERT_FROM_RDD_ENABLED,
+ CometConf.COMET_CONVERT_FROM_ONE_ROW_RELATION_ENABLED).map(_.key -> enabled.toString)
+
protected def sparkConf: SparkConf = {
val conf = new SparkConf()
conf.set("spark.hadoop.fs.file.impl", classOf[DebugFilesystem].getName)
@@ -90,7 +102,7 @@ abstract class CometTestBase
conf.set(CometConf.COMET_ONHEAP_ENABLED.key, "true")
conf.set(CometConf.COMET_EXEC_ENABLED.key, "true")
conf.set(CometConf.COMET_SHUFFLE_ENABLED.key, "true")
- conf.set(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key, "true")
+ conf.setAll(sparkToArrowConversionConfs(enabled = true))
conf.set(CometConf.COMET_NATIVE_SCAN_ENABLED.key, "true")
conf.set(CometConf.COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK.key, "false")
conf.set(CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.key, "true")
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
index 9b0f652a19..74d4fa1025 100644
--- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
@@ -597,7 +597,8 @@ object CometInMemoryCacheBenchmark extends CometBenchmarkBase {
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> nativeCacheEnabled.toString,
- "spark.comet.sparkToColumnar.enabled" -> "true",
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED.key -> "true",
"spark.comet.exec.onHeap.enabled" -> "true",
"spark.sql.inMemoryColumnarStorage.batchSize" -> "10000")
}
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRangeBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRangeBenchmark.scala
index 7d55e90f02..ed8289af90 100644
--- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRangeBenchmark.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometRangeBenchmark.scala
@@ -65,7 +65,7 @@ object CometRangeBenchmark extends CometBenchmarkBase {
"SparkToColumnar",
cometConfs ++ Seq(
CometConf.COMET_EXEC_RANGE_ENABLED.key -> "false",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true"),
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "true"),
plan => {
assert(collect(plan) { case c: CometSparkToColumnarExec => c }.nonEmpty, plan)
assert(collect(plan) { case a: CometHashAggregateExec => a }.nonEmpty, plan)
@@ -74,7 +74,7 @@ object CometRangeBenchmark extends CometBenchmarkBase {
"CometRange",
cometConfs ++ Seq(
CometConf.COMET_EXEC_RANGE_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false"),
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "false"),
plan => {
assert(collect(plan) { case r: CometRangeExec => r }.nonEmpty, plan)
assert(collect(plan) { case a: CometHashAggregateExec => a }.nonEmpty, plan)
diff --git a/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala b/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala
index a855e4484a..76b378e418 100644
--- a/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala
+++ b/spark/src/test/spark-4.x/org/apache/comet/exec/CometEmptyRelationExecSuite.scala
@@ -31,9 +31,9 @@ import org.apache.comet.CometConf
class CometEmptyRelationExecSuite extends CometTestBase {
- // CometTestBase enables the Spark-to-Arrow bridge; use its production default here.
+ // CometTestBase enables the Spark-to-Arrow conversions; use their production defaults here.
override protected def sparkConf: SparkConf =
- super.sparkConf.remove(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key)
+ super.sparkConf.setAll(sparkToArrowConversionConfs(enabled = false))
test(
"EmptyRelationExec is discovered by AQE for joins with default Comet conversion settings") {
@@ -85,7 +85,7 @@ class CometEmptyRelationExecSuite extends CometTestBase {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.SHUFFLE_PARTITIONS.key -> "2",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_CONVERT_FROM_SPARK_PLAN_ENABLED.key -> "false") {
withParquetTable(Seq((1, 2), (2, 3)), "aqe_empty_input") {
// Retain Spark's Range input so AQE can infer emptiness from its completed shuffle.
diff --git a/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala b/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala
index 18f9ee2ec0..04585612ab 100644
--- a/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala
+++ b/spark/src/test/spark-4.x/org/apache/comet/parquet/CometEmptyRelationParquetWriterSuite.scala
@@ -35,7 +35,7 @@ class CometEmptyRelationParquetWriterSuite extends CometParquetWriterTestBase {
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.SHUFFLE_PARTITIONS.key -> "2",
"spark.sql.optimizer.plannedWrite.enabled" -> "true",
- CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_RANGE_ENABLED.key -> "false",
CometConf.COMET_SHUFFLE_CONVERT_FROM_SPARK_PLAN_ENABLED.key -> "false") {
withNativeWriter {
withTempPath { dir =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org