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 60cad718 perf: Fall back to Spark if query uses DPP with v1 data
sources (#897)
60cad718 is described below
commit 60cad718d4b118bebbbaa82cbad12ac44e8e077f
Author: Andy Grove <[email protected]>
AuthorDate: Tue Sep 3 21:11:00 2024 -0600
perf: Fall back to Spark if query uses DPP with v1 data sources (#897)
---
.../main/scala/org/apache/comet/CometConf.scala | 6 ++++
docs/source/user-guide/configs.md | 1 +
.../apache/comet/CometSparkSessionExtensions.scala | 11 +++++++
.../org/apache/comet/exec/CometExecSuite.scala | 37 +++++++++++++++++++++-
.../spark/sql/comet/CometPlanStabilitySuite.scala | 1 +
5 files changed, 55 insertions(+), 1 deletion(-)
diff --git a/common/src/main/scala/org/apache/comet/CometConf.scala
b/common/src/main/scala/org/apache/comet/CometConf.scala
index 03953881..8828b70f 100644
--- a/common/src/main/scala/org/apache/comet/CometConf.scala
+++ b/common/src/main/scala/org/apache/comet/CometConf.scala
@@ -310,6 +310,12 @@ object CometConf extends ShimCometConf {
.doubleConf
.createWithDefault(10.0)
+ val COMET_DPP_FALLBACK_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.dppFallback.enabled")
+ .doc("Whether to fall back to Spark for queries that use DPP.")
+ .booleanConf
+ .createWithDefault(true)
+
val COMET_DEBUG_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.debug.enabled")
.doc(
diff --git a/docs/source/user-guide/configs.md
b/docs/source/user-guide/configs.md
index ce887965..1b5fe736 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -34,6 +34,7 @@ Comet provides the following configuration settings.
| 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.dppFallback.enabled | Whether to fall back to Spark for queries
that use DPP. | true |
| 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 [...]
| spark.comet.exec.aggregate.enabled | Whether to enable aggregate by default.
| true |
diff --git
a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
index 268264a5..fc28f53c 100644
--- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
+++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
@@ -25,6 +25,7 @@ import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
import org.apache.spark.network.util.ByteUnit
import org.apache.spark.sql.{SparkSession, SparkSessionExtensions}
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate.{Final, Partial}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreeNode
@@ -94,7 +95,17 @@ class CometSparkSessionExtensions
}
plan
} else {
+
+ def isDynamicPruningFilter(e: Expression): Boolean =
+ e.exists(_.isInstanceOf[PlanExpression[_]])
+
plan.transform {
+ case scanExec: FileSourceScanExec
+ if COMET_DPP_FALLBACK_ENABLED.get() &&
+ scanExec.partitionFilters.exists(isDynamicPruningFilter) =>
+ withInfo(scanExec, "DPP not supported")
+ scanExec
+
// data source V2
case scanExec: BatchScanExec
if scanExec.scan.isInstanceOf[ParquetScan] &&
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 ad2242d2..c26af28c 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -49,7 +49,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.SESSION_LOCAL_TIMEZONE
import org.apache.spark.unsafe.types.UTF8String
-import org.apache.comet.CometConf
+import org.apache.comet.{CometConf, ExtendedExplainInfo}
import org.apache.comet.CometSparkSessionExtensions.{isSpark33Plus,
isSpark34Plus, isSpark35Plus, isSpark40Plus}
class CometExecSuite extends CometTestBase {
@@ -64,6 +64,41 @@ class CometExecSuite extends CometTestBase {
}
}
+ test("DPP fallback") {
+ withTempDir { path =>
+ // create test data
+ val factPath = s"${path.getAbsolutePath}/fact.parquet"
+ val dimPath = s"${path.getAbsolutePath}/dim.parquet"
+ withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "false") {
+ val one_day = 24 * 60 * 60000
+ val fact = Range(0, 100)
+ .map(i => (i, new java.sql.Date(System.currentTimeMillis() + i *
one_day), i.toString))
+ .toDF("fact_id", "fact_date", "fact_str")
+ fact.write.partitionBy("fact_date").parquet(factPath)
+ val dim = Range(0, 10)
+ .map(i => (i, new java.sql.Date(System.currentTimeMillis() + i *
one_day), i.toString))
+ .toDF("dim_id", "dim_date", "dim_str")
+ dim.write.parquet(dimPath)
+ }
+
+ // note that this test does not trigger DPP with v2 data source
+ Seq("parquet").foreach { v1List =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_DPP_FALLBACK_ENABLED.key -> "true") {
+ spark.read.parquet(factPath).createOrReplaceTempView("dpp_fact")
+ spark.read.parquet(dimPath).createOrReplaceTempView("dpp_dim")
+ val df =
+ spark.sql(
+ "select * from dpp_fact join dpp_dim on fact_date = dim_date
where dim_id > 7")
+ val (_, cometPlan) = checkSparkAnswer(df)
+ val infos = new ExtendedExplainInfo().generateExtendedInfo(cometPlan)
+ assert(infos.contains("DPP not supported"))
+ }
+ }
+ }
+ }
+
test("ShuffleQueryStageExec could be direct child node of
CometBroadcastExchangeExec") {
val table = "src"
withTable(table) {
diff --git
a/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanStabilitySuite.scala
b/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanStabilitySuite.scala
index 759833e6..83cc8982 100644
---
a/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanStabilitySuite.scala
+++
b/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanStabilitySuite.scala
@@ -260,6 +260,7 @@ trait CometPlanStabilitySuite extends
DisableAdaptiveExecutionSuite with TPCDSBa
withSQLConf(
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_DPP_FALLBACK_ENABLED.key -> "false",
CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
CometConf.COMET_CAST_ALLOW_INCOMPATIBLE.key -> "true", // needed for
v1.4/q9, v1.4/q44, v2.7.0/q6, v2.7.0/q64
"spark.sql.readSideCharPadding" -> "false",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
