You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ag...@apache.org on 2024/08/16 12:46:21 UTC
(datafusion-comet) branch main updated: feat: Add specific configs for converting Spark Parquet and JSON data to Arrow (#832)
This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 6051232a feat: Add specific configs for converting Spark Parquet and JSON data to Arrow (#832)
6051232a is described below
commit 6051232ae76f7dd0b7797f8bb23a4133332768e4
Author: Andy Grove <ag...@apache.org>
AuthorDate: Fri Aug 16 06:46:17 2024 -0600
feat: Add specific configs for converting Spark Parquet and JSON data to Arrow (#832)
* Add option to enable JSON scan
* support v1 and v2 data sources
* formatting
* format
* Add schema check
* fix regression
* improve configs and docs
* fix
* improve test and fix typo in doc
* renaming some variables but no change to public config names
* improve test
* improve test
* Update spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
Co-authored-by: Liang-Chi Hsieh <vi...@gmail.com>
* Update spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
Co-authored-by: Liang-Chi Hsieh <vi...@gmail.com>
* improve config description
* fix path
---------
Co-authored-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../main/scala/org/apache/comet/CometConf.scala | 38 +++++++++++++-----
docs/source/index.rst | 5 ++-
docs/source/user-guide/configs.md | 6 ++-
docs/source/user-guide/datasources.md | 32 +++++++++++++++
docs/source/user-guide/operators.md | 27 +++++++------
.../apache/comet/CometSparkSessionExtensions.scala | 39 ++++++++++++++++---
.../test/resources/test-data/json-test-1.ndjson | 4 ++
.../org/apache/comet/CometExpressionSuite.scala | 45 ++++++++++++----------
.../comet/exec/CometColumnarShuffleSuite.scala | 2 +-
.../org/apache/comet/exec/CometExecSuite.scala | 14 +++++++
.../scala/org/apache/spark/sql/CometTestBase.scala | 2 +-
11 files changed, 158 insertions(+), 56 deletions(-)
diff --git a/common/src/main/scala/org/apache/comet/CometConf.scala b/common/src/main/scala/org/apache/comet/CometConf.scala
index 1db80298..324c1d02 100644
--- a/common/src/main/scala/org/apache/comet/CometConf.scala
+++ b/common/src/main/scala/org/apache/comet/CometConf.scala
@@ -84,15 +84,33 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(sys.env.getOrElse("ENABLE_COMET", "true").toBoolean)
- val COMET_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled")
+ val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled")
.doc(
- "Whether to enable Comet scan. When this is turned on, Spark will use Comet to read " +
- "Parquet data source. Note that to enable native vectorized execution, both this " +
- "config and 'spark.comet.exec.enabled' need to be enabled. By default, this config " +
- "is true.")
+ "Whether to enable native scans. When this is turned on, Spark will use Comet to " +
+ "read supported data sources (currently only Parquet is supported natively). Note " +
+ "that to enable native vectorized execution, both this config and " +
+ "'spark.comet.exec.enabled' need to be enabled. By default, this config is true.")
.booleanConf
.createWithDefault(true)
+ val COMET_CONVERT_FROM_PARQUET_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.parquet.enabled")
+ .doc(
+ "When enabled, data from Spark (non-native) Parquet v1 and v2 scans will be converted to " +
+ "Arrow format. Note that to enable native vectorized execution, both this config and " +
+ "'spark.comet.exec.enabled' need to be enabled.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val COMET_CONVERT_FROM_JSON_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.json.enabled")
+ .doc(
+ "When enabled, data from Spark (non-native) JSON v1 and v2 scans will be converted to " +
+ "Arrow format. Note that to enable native vectorized execution, both this config and " +
+ "'spark.comet.exec.enabled' need to be enabled.")
+ .booleanConf
+ .createWithDefault(false)
+
val COMET_EXEC_ENABLED: ConfigEntry[Boolean] = conf(s"$COMET_EXEC_CONFIG_PREFIX.enabled")
.doc(
"Whether to enable Comet native vectorized execution for Spark. This controls whether " +
@@ -457,20 +475,20 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(COMET_SCHEMA_EVOLUTION_ENABLED_DEFAULT)
- val COMET_SPARK_TO_COLUMNAR_ENABLED: ConfigEntry[Boolean] =
+ val COMET_SPARK_TO_ARROW_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.sparkToColumnar.enabled")
.internal()
- .doc("Whether to enable Spark to Comet columnar conversion. When this is turned on, " +
+ .doc("Whether to enable Spark to Arrow columnar conversion. When this is turned on, " +
"Comet will convert operators in " +
- "`spark.comet.sparkToColumnar.supportedOperatorList` into Comet columnar based before " +
+ "`spark.comet.sparkToColumnar.supportedOperatorList` into Arrow columnar format before " +
"processing.")
.booleanConf
.createWithDefault(false)
- val COMET_SPARK_TO_COLUMNAR_SUPPORTED_OPERATOR_LIST: ConfigEntry[Seq[String]] =
+ val COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST: ConfigEntry[Seq[String]] =
conf("spark.comet.sparkToColumnar.supportedOperatorList")
.doc(
- "A comma-separated list of operators that will be converted to Comet columnar " +
+ "A comma-separated list of operators that will be converted to Arrow columnar " +
"format when 'spark.comet.sparkToColumnar.enabled' is true")
.stringConf
.toSequence
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 101c5785..4bf5d9fd 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -42,9 +42,10 @@ as a native runtime to achieve improvement in terms of query efficiency and quer
Comet Overview <user-guide/overview>
Installing Comet <user-guide/installation>
- Supported Expressions <user-guide/expressions>
- Supported Operators <user-guide/operators>
+ Supported Data Sources <user-guide/datasources>
Supported Data Types <user-guide/datatypes>
+ Supported Operators <user-guide/operators>
+ Supported Expressions <user-guide/expressions>
Configuration Settings <user-guide/configs>
Compatibility Guide <user-guide/compatibility>
Tuning Guide <user-guide/tuning>
diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md
index 836f6444..8465baa7 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -30,6 +30,8 @@ Comet provides the following configuration settings.
| spark.comet.columnar.shuffle.async.max.thread.num | Maximum number of threads on an executor used for Comet async columnar shuffle. By default, this config is 100. This is the upper bound of total number of shuffle threads per executor. In other words, if the number of cores * the number of shuffle threads per task `spark.comet.columnar.shuffle.async.thread.num` is larger than this config. Comet will use this config as the number of shuffle threads per executor instead. | 100 |
| spark.comet.columnar.shuffle.async.thread.num | Number of threads used for Comet async columnar shuffle per shuffle task. By default, this config is 3. Note that more threads means more memory requirement to buffer shuffle data before flushing to disk. Also, more threads may not always improve performance, and should be set based on the number of cores available. | 3 |
| spark.comet.columnar.shuffle.memory.factor | Fraction of Comet memory to be allocated per executor process for Comet shuffle. Comet memory size is specified by `spark.comet.memoryOverhead` or calculated by `spark.comet.memory.overhead.factor` * `spark.executor.memory`. By default, this config is 1.0. | 1.0 |
+| spark.comet.convert.json.enabled | When enabled, data from Spark (non-native) JSON v1 and v2 scans will be converted to Arrow format. Note that to enable native vectorized execution, both this config and 'spark.comet.exec.enabled' need to be enabled. | false |
+| spark.comet.convert.parquet.enabled | When enabled, data from Spark (non-native) Parquet v1 and v2 scans will be converted to Arrow format. Note that to enable native vectorized execution, both this config and 'spark.comet.exec.enabled' need to be enabled. | false |
| spark.comet.debug.enabled | Whether to enable debug mode for Comet. By default, this config is false. When enabled, Comet will do additional checks for debugging purpose. For example, validating array when importing arrays from JVM at native side. Note that these checks may be expensive in performance and should only be enabled for debugging purpose. | false |
| spark.comet.enabled | Whether to enable Comet extension for Spark. When this is turned on, Spark will use Comet to read Parquet data source. Note that to enable native vectorized execution, both this config and 'spark.comet.exec.enabled' need to be enabled. By default, this config is the value of the env var `ENABLE_COMET` if set, or true otherwise. | true |
| spark.comet.exceptionOnDatetimeRebase | Whether to throw exception when seeing dates/timestamps from the legacy hybrid (Julian + Gregorian) calendar. Since Spark 3, dates/timestamps were written according to the Proleptic Gregorian calendar. When this is true, Comet will throw exceptions when seeing these dates/timestamps that were written by Spark version before 3.0. If this is false, these dates/timestamps will be read as if they were written to the Proleptic Gregorian calendar and w [...]
@@ -64,8 +66,8 @@ Comet provides the following configuration settings.
| spark.comet.nativeLoadRequired | Whether to require Comet native library to load successfully when Comet is enabled. If not, Comet will silently fallback to Spark when it fails to load the native lib. Otherwise, an error will be thrown and the Spark job will be aborted. | false |
| spark.comet.parquet.enable.directBuffer | Whether to use Java direct byte buffer when reading Parquet. By default, this is false | false |
| spark.comet.regexp.allowIncompatible | Comet is not currently fully compatible with Spark for all regular expressions. Set this config to true to allow them anyway using Rust's regular expression engine. See compatibility guide for more information. | false |
-| spark.comet.scan.enabled | Whether to enable Comet scan. When this is turned on, Spark will use Comet to read Parquet data source. Note that to enable native vectorized execution, both this config and 'spark.comet.exec.enabled' need to be enabled. By default, this config is true. | true |
+| spark.comet.scan.enabled | Whether to enable native scans. When this is turned on, Spark will use Comet to read supported data sources (currently only Parquet is supported natively). Note that to enable native vectorized execution, both this config and 'spark.comet.exec.enabled' need to be enabled. By default, this config is true. | true |
| spark.comet.scan.preFetch.enabled | Whether to enable pre-fetching feature of CometScan. By default is disabled. | false |
| spark.comet.scan.preFetch.threadNum | The number of threads running pre-fetching for CometScan. Effective if spark.comet.scan.preFetch.enabled is enabled. By default it is 2. Note that more pre-fetching threads means more memory requirement to store pre-fetched row groups. | 2 |
| spark.comet.shuffle.preferDictionary.ratio | The ratio of total values to distinct values in a string column to decide whether to prefer dictionary encoding when shuffling the column. If the ratio is higher than this config, dictionary encoding will be used on shuffling string column. This config is effective if it is higher than 1.0. By default, this config is 10.0. Note that this config is only used when `spark.comet.exec.shuffle.mode` is `jvm`. | 10.0 |
-| spark.comet.sparkToColumnar.supportedOperatorList | A comma-separated list of operators that will be converted to Comet columnar format when 'spark.comet.sparkToColumnar.enabled' is true | Range,InMemoryTableScan |
+| spark.comet.sparkToColumnar.supportedOperatorList | A comma-separated list of operators that will be converted to Arrow columnar format when 'spark.comet.sparkToColumnar.enabled' is true | Range,InMemoryTableScan |
diff --git a/docs/source/user-guide/datasources.md b/docs/source/user-guide/datasources.md
new file mode 100644
index 00000000..92c13f67
--- /dev/null
+++ b/docs/source/user-guide/datasources.md
@@ -0,0 +1,32 @@
+<!---
+ 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.
+-->
+
+# Supported Spark Data Sources
+
+## Parquet
+
+When `spark.comet.scan.enabled` is enabled, Parquet scans will be performed natively by Comet if all data types
+in the schema are supported. When this option is not enabled, the scan will fall back to Spark. In this case,
+enabling `spark.comet.convert.parquet.enabled` will immediately convert the data into Arrow format, allowing native
+execution to happen after that, but the process may not be efficient.
+
+## JSON
+
+Comet does not provide native JSON scan, but when `spark.comet.convert.json.enabled` is enabled, data is immediately
+converted into Arrow format, allowing native execution to happen after that.
diff --git a/docs/source/user-guide/operators.md b/docs/source/user-guide/operators.md
index e3a3ac52..7edbb1e7 100644
--- a/docs/source/user-guide/operators.md
+++ b/docs/source/user-guide/operators.md
@@ -22,17 +22,16 @@
The following Spark operators are currently replaced with native versions. Query stages that contain any operators
not supported by Comet will fall back to regular Spark execution.
-| Operator | Notes |
-| -------------------------------------------- | ----- |
-| FileSourceScanExec/BatchScanExec for Parquet | |
-| Projection | |
-| Filter | |
-| Sort | |
-| Hash Aggregate | |
-| Limit | |
-| Sort-merge Join | |
-| Hash Join | |
-| BroadcastHashJoinExec | |
-| Shuffle | |
-| Expand | |
-| Union | |
+| Operator | Notes |
+| --------------------- | ----- |
+| Projection | |
+| Filter | |
+| Sort | |
+| Hash Aggregate | |
+| Limit | |
+| Sort-merge Join | |
+| Hash Join | |
+| BroadcastHashJoinExec | |
+| Shuffle | |
+| Expand | |
+| Union | |
diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
index f24624dc..e978e7d3 100644
--- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
+++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
@@ -35,8 +35,10 @@ import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.{AQEShuffleReadExec, BroadcastQueryStageExec, ShuffleQueryStageExec}
import org.apache.spark.sql.execution.aggregate.HashAggregateExec
import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.json.JsonFileFormat
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.json.JsonScan
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ReusedExchangeExec, ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
@@ -1095,7 +1097,7 @@ object CometSparkSessionExtensions extends Logging {
}
private[comet] def isCometScanEnabled(conf: SQLConf): Boolean = {
- COMET_SCAN_ENABLED.get(conf)
+ COMET_NATIVE_SCAN_ENABLED.get(conf)
}
private[comet] def isCometExecEnabled(conf: SQLConf): Boolean = {
@@ -1131,12 +1133,39 @@ object CometSparkSessionExtensions extends Logging {
// operators can have a chance to be converted to columnar. Leaf operators that output
// columnar batches, such as Spark's vectorized readers, will also be converted to native
// comet batches.
- // TODO: consider converting other intermediate operators to columnar.
- op.isInstanceOf[LeafExecNode] && CometSparkToColumnarExec.isSchemaSupported(op.schema) &&
- COMET_SPARK_TO_COLUMNAR_ENABLED.get(conf) && {
+ if (CometSparkToColumnarExec.isSchemaSupported(op.schema)) {
+ op match {
+ // Convert Spark DS v1 scan to Arrow format
+ case scan: FileSourceScanExec =>
+ scan.relation.fileFormat match {
+ case _: JsonFileFormat => CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf)
+ case _: ParquetFileFormat => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
+ case _ => isSparkToArrowEnabled(conf, op)
+ }
+ // Convert Spark DS v2 scan to Arrow format
+ case scan: BatchScanExec =>
+ scan.scan match {
+ case _: JsonScan => CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf)
+ case _: ParquetScan => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
+ case _ => isSparkToArrowEnabled(conf, op)
+ }
+ // other leaf nodes
+ case _: LeafExecNode =>
+ isSparkToArrowEnabled(conf, op)
+ case _ =>
+ // TODO: consider converting other intermediate operators to columnar.
+ false
+ }
+ } else {
+ false
+ }
+ }
+
+ 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_COLUMNAR_SUPPORTED_OPERATOR_LIST.get(conf).contains(nodeName)
+ COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.get(conf).contains(nodeName)
}
}
diff --git a/spark/src/test/resources/test-data/json-test-1.ndjson b/spark/src/test/resources/test-data/json-test-1.ndjson
new file mode 100644
index 00000000..324b3060
--- /dev/null
+++ b/spark/src/test/resources/test-data/json-test-1.ndjson
@@ -0,0 +1,4 @@
+{ "a": [1], "b": { "c": "foo" }}
+{ "a": 2, "b": { "d": "bar" }}
+{ "b": { "d": "bar" }}
+{ "a": 3 }
diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
index b9c5c3ca..d27c1791 100644
--- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -1974,27 +1974,30 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
test("get_struct_field") {
- withSQLConf(
- CometConf.COMET_SPARK_TO_COLUMNAR_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_COLUMNAR_SUPPORTED_OPERATOR_LIST.key -> "FileSourceScan") {
- withTempPath { dir =>
- var df = spark
- .range(5)
- // Add both a null struct and null inner value
- .select(
- when(
- col("id") > 1,
- struct(
- when(col("id") > 2, col("id")).alias("id"),
- when(col("id") > 2, struct(when(col("id") > 3, col("id")).alias("id")))
- .as("nested2")))
- .alias("nested1"))
-
- df.write.parquet(dir.toString())
-
- df = spark.read.parquet(dir.toString())
- checkSparkAnswerAndOperator(df.select("nested1.id"))
- checkSparkAnswerAndOperator(df.select("nested1.nested2.id"))
+ Seq("", "parquet").foreach { v1List =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ var df = spark
+ .range(5)
+ // Add both a null struct and null inner value
+ .select(
+ when(
+ col("id") > 1,
+ struct(
+ when(col("id") > 2, col("id")).alias("id"),
+ when(col("id") > 2, struct(when(col("id") > 3, col("id")).alias("id")))
+ .as("nested2")))
+ .alias("nested1"))
+
+ df.write.parquet(dir.toString())
+
+ df = spark.read.parquet(dir.toString())
+ checkSparkAnswerAndOperator(df.select("nested1.id"))
+ checkSparkAnswerAndOperator(df.select("nested1.nested2.id"))
+ }
}
}
}
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
index 78b4bbb9..a2f07155 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
@@ -831,7 +831,7 @@ abstract class CometColumnarShuffleSuite extends CometTestBase with AdaptiveSpar
withSQLConf(
SQLConf.USE_V1_SOURCE_LIST.key -> "", // Use DataSourceV2
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", // Disable AQE
- CometConf.COMET_SCAN_ENABLED.key -> "false") { // Disable CometScan to use Spark BatchScan
+ CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") { // Disable CometScan to use Spark BatchScan
withParquetTable((0 until 5).map(i => (i, (i + 1).toLong)), "tbl") {
val df = sql("SELECT * FROM tbl")
val shuffled = df
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 5ca9401c..15864a33 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -1606,6 +1606,20 @@ class CometExecSuite extends CometTestBase {
}
})
}
+
+ test("read JSON file") {
+ Seq("", "json").foreach { v1List =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_JSON_ENABLED.key -> "true") {
+ spark.read
+ .json("src/test/resources/test-data/json-test-1.ndjson")
+ .createOrReplaceTempView("tbl")
+ checkSparkAnswerAndOperator("SELECT a, b.c, b.d FROM tbl")
+ }
+ }
+ }
}
case class BucketedTableTestSpec(
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 fd0328c8..a88f9f45 100644
--- a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
@@ -79,7 +79,7 @@ abstract class CometTestBase
conf.set(CometConf.COMET_ENABLED.key, "true")
conf.set(CometConf.COMET_EXEC_ENABLED.key, "true")
conf.set(CometConf.COMET_EXEC_ALL_OPERATOR_ENABLED.key, "true")
- conf.set(CometConf.COMET_SPARK_TO_COLUMNAR_ENABLED.key, "true")
+ conf.set(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key, "true")
conf.set(CometConf.COMET_MEMORY_OVERHEAD.key, "2g")
conf
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org