This is an automated email from the ASF dual-hosted git repository.
viirya 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 8c15cf45 chore: Enable Comet shuffle for Spark core-1 test (#924)
8c15cf45 is described below
commit 8c15cf45804e4d45bb4ee476c303317a1f9b83f0
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Mon Sep 9 11:50:04 2024 -0700
chore: Enable Comet shuffle for Spark core-1 test (#924)
* test
* Update diffs
* more
* fix
* Update diffs
---
.github/workflows/spark_sql_test.yml | 2 +-
.github/workflows/spark_sql_test_ansi.yml | 2 +-
dev/diffs/3.4.3.diff | 56 ++++++++++++++++++-----
dev/diffs/3.5.1.diff | 75 ++++++++++++++++++++++++-------
dev/diffs/4.0.0-preview1.diff | 66 ++++++++++++++++++++++-----
5 files changed, 162 insertions(+), 39 deletions(-)
diff --git a/.github/workflows/spark_sql_test.yml
b/.github/workflows/spark_sql_test.yml
index 0d2cec01..bb81fbd5 100644
--- a/.github/workflows/spark_sql_test.yml
+++ b/.github/workflows/spark_sql_test.yml
@@ -76,7 +76,7 @@ jobs:
run: |
cd apache-spark
rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet
cache requires cleanups
- ENABLE_COMET=true ENABLE_COMET_SHUFFLE=${{ matrix.module.name ==
'sql/core-1' && 'false' || 'true' }} build/sbt ${{ matrix.module.args1 }} "${{
matrix.module.args2 }}"
+ ENABLE_COMET=true ENABLE_COMET_SHUFFLE=true build/sbt ${{
matrix.module.args1 }} "${{ matrix.module.args2 }}"
env:
LC_ALL: "C.UTF-8"
diff --git a/.github/workflows/spark_sql_test_ansi.yml
b/.github/workflows/spark_sql_test_ansi.yml
index 37212222..db2c5708 100644
--- a/.github/workflows/spark_sql_test_ansi.yml
+++ b/.github/workflows/spark_sql_test_ansi.yml
@@ -74,7 +74,7 @@ jobs:
run: |
cd apache-spark
rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet
cache requires cleanups
- RUST_BACKTRACE=1 ENABLE_COMET=true ENABLE_COMET_ANSI_MODE=true
ENABLE_COMET_SHUFFLE=${{ matrix.module.name == 'sql/core-1' && 'false' ||
'true' }} build/sbt ${{ matrix.module.args1 }} "${{ matrix.module.args2 }}"
+ RUST_BACKTRACE=1 ENABLE_COMET=true ENABLE_COMET_ANSI_MODE=true
ENABLE_COMET_SHUFFLE=true build/sbt ${{ matrix.module.args1 }} "${{
matrix.module.args2 }}"
env:
LC_ALL: "C.UTF-8"
diff --git a/dev/diffs/3.4.3.diff b/dev/diffs/3.4.3.diff
index 486f1cf9..5eecc29f 100644
--- a/dev/diffs/3.4.3.diff
+++ b/dev/diffs/3.4.3.diff
@@ -363,7 +363,7 @@ index daef11ae4d6..9f3cc9181f2 100644
assert(exchanges.size == 2)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
-index f33432ddb6f..8690c859d0c 100644
+index f33432ddb6f..9e598e7a344 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
@@ -22,6 +22,7 @@ import org.scalatest.GivenWhenThen
@@ -434,7 +434,17 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
"""
-@@ -1238,7 +1247,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1215,7 +1224,8 @@ abstract class DynamicPartitionPruningSuiteBase
+ }
+
+ test("SPARK-32509: Unused Dynamic Pruning filter shouldn't affect " +
+- "canonicalization and exchange reuse") {
++ "canonicalization and exchange reuse",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ val df = sql(
+@@ -1238,7 +1248,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -444,7 +454,7 @@ index f33432ddb6f..8690c859d0c 100644
Given("dynamic pruning filter on the build side")
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
-@@ -1279,7 +1289,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1279,7 +1290,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -454,7 +464,7 @@ index f33432ddb6f..8690c859d0c 100644
Seq(NO_CODEGEN, CODEGEN_ONLY).foreach { mode =>
Seq(true, false).foreach { pruning =>
withSQLConf(
-@@ -1311,7 +1322,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1311,7 +1323,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -464,7 +474,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
-@@ -1423,7 +1435,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1423,7 +1436,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -474,7 +484,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
""" WITH v as (
-@@ -1470,7 +1483,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1470,7 +1484,8 @@ abstract class DynamicPartitionPruningSuiteBase
checkAnswer(df, Row(3, 2) :: Row(3, 2) :: Row(3, 2) :: Row(3, 2) :: Nil)
}
@@ -484,7 +494,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1485,7 +1499,7 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1485,7 +1500,7 @@ abstract class DynamicPartitionPruningSuiteBase
}
test("SPARK-38148: Do not add dynamic partition pruning if there exists
static partition " +
@@ -493,7 +503,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
Seq(
"f.store_id = 1" -> false,
-@@ -1557,7 +1571,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1557,7 +1572,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -503,7 +513,7 @@ index f33432ddb6f..8690c859d0c 100644
withTable("duplicate_keys") {
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
Seq[(Int, String)]((1, "NL"), (1, "NL"), (3, "US"), (3, "US"), (3,
"US"))
-@@ -1588,7 +1603,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1588,7 +1604,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -513,7 +523,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1617,7 +1633,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1617,7 +1634,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -523,7 +533,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1729,6 +1746,8 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
+@@ -1729,6 +1747,8 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
case s: BatchScanExec =>
// we use f1 col for v2 tables due to schema pruning
s.output.exists(_.exists(_.argString(maxFields =
100).contains("f1")))
@@ -650,6 +660,30 @@ index 00000000000..4b31bea33de
+ }
+ }
+}
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+index fda442eeef0..1b69e4f280e 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+@@ -468,7 +468,8 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
+ }
+
+ test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
+- "on the same column") {
++ "on the same column",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on "
+
+@@ -477,7 +478,8 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
+ }
+
+ test("Runtime bloom filter join: add bloom filter if dpp filter exists on "
+
+- "a different column") {
++ "a different column",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
index 1792b4c32eb..1616e6f39bd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
diff --git a/dev/diffs/3.5.1.diff b/dev/diffs/3.5.1.diff
index 16a04b06..d463b583 100644
--- a/dev/diffs/3.5.1.diff
+++ b/dev/diffs/3.5.1.diff
@@ -342,7 +342,7 @@ index c2fe31520ac..0f54b233d14 100644
assert(exchanges.size == 2)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
-index f33432ddb6f..8690c859d0c 100644
+index f33432ddb6f..19ce507e82b 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
@@ -22,6 +22,7 @@ import org.scalatest.GivenWhenThen
@@ -413,7 +413,17 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
"""
-@@ -1238,7 +1247,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1215,7 +1224,8 @@ abstract class DynamicPartitionPruningSuiteBase
+ }
+
+ test("SPARK-32509: Unused Dynamic Pruning filter shouldn't affect " +
+- "canonicalization and exchange reuse") {
++ "canonicalization and exchange reuse",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ val df = sql(
+@@ -1238,7 +1248,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -423,7 +433,7 @@ index f33432ddb6f..8690c859d0c 100644
Given("dynamic pruning filter on the build side")
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
-@@ -1279,7 +1289,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1279,7 +1290,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -433,7 +443,7 @@ index f33432ddb6f..8690c859d0c 100644
Seq(NO_CODEGEN, CODEGEN_ONLY).foreach { mode =>
Seq(true, false).foreach { pruning =>
withSQLConf(
-@@ -1311,7 +1322,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1311,7 +1323,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -443,7 +453,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
-@@ -1423,7 +1435,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1423,7 +1436,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -453,7 +463,17 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
""" WITH v as (
-@@ -1470,7 +1483,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1454,7 +1468,8 @@ abstract class DynamicPartitionPruningSuiteBase
+ }
+ }
+
+- test("SPARK-35568: Fix UnsupportedOperationException when enabling both AQE
and DPP") {
++ test("SPARK-35568: Fix UnsupportedOperationException when enabling both AQE
and DPP",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+ val df = sql(
+ """
+ |SELECT s.store_id, f.product_id
+@@ -1470,7 +1485,8 @@ abstract class DynamicPartitionPruningSuiteBase
checkAnswer(df, Row(3, 2) :: Row(3, 2) :: Row(3, 2) :: Row(3, 2) :: Nil)
}
@@ -463,7 +483,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1485,7 +1499,7 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1485,7 +1501,7 @@ abstract class DynamicPartitionPruningSuiteBase
}
test("SPARK-38148: Do not add dynamic partition pruning if there exists
static partition " +
@@ -472,7 +492,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
Seq(
"f.store_id = 1" -> false,
-@@ -1557,7 +1571,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1557,7 +1573,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -482,7 +502,7 @@ index f33432ddb6f..8690c859d0c 100644
withTable("duplicate_keys") {
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
Seq[(Int, String)]((1, "NL"), (1, "NL"), (3, "US"), (3, "US"), (3,
"US"))
-@@ -1588,7 +1603,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1588,7 +1605,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -492,7 +512,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1617,7 +1633,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1617,7 +1635,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -502,7 +522,7 @@ index f33432ddb6f..8690c859d0c 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1729,6 +1746,8 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
+@@ -1729,6 +1748,8 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
case s: BatchScanExec =>
// we use f1 col for v2 tables due to schema pruning
s.output.exists(_.exists(_.argString(maxFields =
100).contains("f1")))
@@ -629,6 +649,30 @@ index 00000000000..4b31bea33de
+ }
+ }
+}
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+index fedfd9ff587..c5bfc8f16e4 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+@@ -505,7 +505,8 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
+ }
+
+ test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
+- "on the same column") {
++ "on the same column",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on "
+
+@@ -514,7 +515,8 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
+ }
+
+ test("Runtime bloom filter join: add bloom filter if dpp filter exists on "
+
+- "a different column") {
++ "a different column",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
index 7af826583bd..3c3def1eb67 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
@@ -1266,14 +1310,14 @@ index 47679ed7865..9ffbaecb98e 100644
assert(collectWithSubqueries(plan) { case s: SortAggregateExec => s
}.length == sortAggCount)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
-index 5a413c77754..37eee6e2c07 100644
+index 5a413c77754..c52f4b3818c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution
import org.apache.spark.sql.{Dataset, QueryTest, Row, SaveMode}
import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
import org.apache.spark.sql.catalyst.expressions.codegen.{ByteCodeStats,
CodeAndComment, CodeGenerator}
-+import org.apache.spark.sql.comet.CometSortMergeJoinExec
++import org.apache.spark.sql.comet.{CometSortExec, CometSortMergeJoinExec}
import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
import org.apache.spark.sql.execution.aggregate.{HashAggregateExec,
SortAggregateExec}
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
@@ -1313,18 +1357,19 @@ index 5a413c77754..37eee6e2c07 100644
}.size === 2)
checkAnswer(twoJoinsDF, Seq(Row(6), Row(7), Row(8), Row(9)))
}
-@@ -536,7 +537,9 @@ class WholeStageCodegenSuite extends QueryTest with
SharedSparkSession
+@@ -536,7 +537,10 @@ class WholeStageCodegenSuite extends QueryTest with
SharedSparkSession
val plan = df.queryExecution.executedPlan
assert(plan.exists(p =>
p.isInstanceOf[WholeStageCodegenExec] &&
- p.asInstanceOf[WholeStageCodegenExec].child.isInstanceOf[SortExec]))
+ p.asInstanceOf[WholeStageCodegenExec].collect {
+ case _: SortExec => true
++ case _: CometSortExec => true
+ }.nonEmpty))
assert(df.collect() === Array(Row(1), Row(2), Row(3)))
}
-@@ -716,7 +719,9 @@ class WholeStageCodegenSuite extends QueryTest with
SharedSparkSession
+@@ -716,7 +720,9 @@ class WholeStageCodegenSuite extends QueryTest with
SharedSparkSession
.write.mode(SaveMode.Overwrite).parquet(path)
withSQLConf(SQLConf.WHOLESTAGE_MAX_NUM_FIELDS.key -> "255",
diff --git a/dev/diffs/4.0.0-preview1.diff b/dev/diffs/4.0.0-preview1.diff
index 4c900c2b..b473bbf0 100644
--- a/dev/diffs/4.0.0-preview1.diff
+++ b/dev/diffs/4.0.0-preview1.diff
@@ -440,7 +440,7 @@ index 16a493b5290..3f0b70e2d59 100644
assert(exchanges.size == 2)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
-index 2c24cc7d570..de265cfaeae 100644
+index 2c24cc7d570..65163e35666 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
@@ -22,6 +22,7 @@ import org.scalatest.GivenWhenThen
@@ -511,7 +511,17 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
"""
-@@ -1238,7 +1247,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1215,7 +1224,8 @@ abstract class DynamicPartitionPruningSuiteBase
+ }
+
+ test("SPARK-32509: Unused Dynamic Pruning filter shouldn't affect " +
+- "canonicalization and exchange reuse") {
++ "canonicalization and exchange reuse",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ val df = sql(
+@@ -1238,7 +1248,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -521,7 +531,7 @@ index 2c24cc7d570..de265cfaeae 100644
Given("dynamic pruning filter on the build side")
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
-@@ -1279,7 +1289,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1279,7 +1290,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -531,7 +541,7 @@ index 2c24cc7d570..de265cfaeae 100644
Seq(NO_CODEGEN, CODEGEN_ONLY).foreach { mode =>
Seq(true, false).foreach { pruning =>
withSQLConf(
-@@ -1311,7 +1322,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1311,7 +1323,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -541,7 +551,7 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
-@@ -1424,7 +1436,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1424,7 +1437,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -551,7 +561,17 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true") {
val df = sql(
""" WITH v as (
-@@ -1471,7 +1484,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1455,7 +1469,8 @@ abstract class DynamicPartitionPruningSuiteBase
+ }
+ }
+
+- test("SPARK-35568: Fix UnsupportedOperationException when enabling both AQE
and DPP") {
++ test("SPARK-35568: Fix UnsupportedOperationException when enabling both AQE
and DPP",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+ val df = sql(
+ """
+ |SELECT s.store_id, f.product_id
+@@ -1471,7 +1486,8 @@ abstract class DynamicPartitionPruningSuiteBase
checkAnswer(df, Row(3, 2) :: Row(3, 2) :: Row(3, 2) :: Row(3, 2) :: Nil)
}
@@ -561,7 +581,7 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1486,7 +1500,7 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1486,7 +1502,7 @@ abstract class DynamicPartitionPruningSuiteBase
}
test("SPARK-38148: Do not add dynamic partition pruning if there exists
static partition " +
@@ -570,7 +590,7 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
Seq(
"f.store_id = 1" -> false,
-@@ -1558,7 +1572,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1558,7 +1574,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -580,7 +600,7 @@ index 2c24cc7d570..de265cfaeae 100644
withTable("duplicate_keys") {
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
Seq[(Int, String)]((1, "NL"), (1, "NL"), (3, "US"), (3, "US"), (3,
"US"))
-@@ -1589,7 +1604,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1589,7 +1606,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -590,7 +610,7 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1618,7 +1634,8 @@ abstract class DynamicPartitionPruningSuiteBase
+@@ -1618,7 +1636,8 @@ abstract class DynamicPartitionPruningSuiteBase
}
}
@@ -600,7 +620,7 @@ index 2c24cc7d570..de265cfaeae 100644
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") {
val df = sql(
"""
-@@ -1730,6 +1747,8 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
+@@ -1730,6 +1749,8 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
case s: BatchScanExec =>
// we use f1 col for v2 tables due to schema pruning
s.output.exists(_.exists(_.argString(maxFields =
100).contains("f1")))
@@ -727,6 +747,30 @@ index 00000000000..4b31bea33de
+ }
+ }
+}
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+index 027477a8291..f2568916e88 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+@@ -442,7 +442,8 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
+ }
+
+ test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
+- "on the same column") {
++ "on the same column",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on "
+
+@@ -451,7 +452,8 @@ class InjectRuntimeFilterSuite extends QueryTest with
SQLTestUtils with SharedSp
+ }
+
+ test("Runtime bloom filter join: add bloom filter if dpp filter exists on "
+
+- "a different column") {
++ "a different column",
++ IgnoreComet("TODO: Support SubqueryBroadcastExec in Comet: #242")) {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
index 53e47f428c3..a55d8f0c161 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
