This is an automated email from the ASF dual-hosted git repository.
jackylee-ch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 2a68f30029 [GLUTEN-12807][CORE] Remove the leftover Spark 3.3 version
checks (#12981)
2a68f30029 is described below
commit 2a68f3002967b7ee159cc6fe1541b491c22815c5
Author: YangJie <[email protected]>
AuthorDate: Thu Sep 10 04:54:22 2026 -0400
[GLUTEN-12807][CORE] Remove the leftover Spark 3.3 version checks (#12981)
---
.github/workflows/util/install-spark-deps.sh | 1 -
...ClickHouseIcebergMOREqualityDeletionSuite.scala | 2 +-
.../execution/iceberg/ClickHouseIcebergSuite.scala | 6 +-
.../kafka/ClickhouseGlutenKafkaScanSuite.scala | 5 +-
.../backendsapi/clickhouse/CHListenerApi.scala | 3 -
.../execution/GlutenClickHouseDecimalSuite.scala | 2 +-
.../GlutenClickHouseTPCHBucketSuite.scala | 52 +----
...lutenClickHouseWholeStageTransformerSuite.scala | 16 +-
.../execution/GlutenFunctionValidateSuite.scala | 13 +-
.../GlutenClickhouseStringFunctionsSuite.scala | 2 +-
.../hive/GlutenClickHouseHiveTableSuite.scala | 20 +-
.../GlutenClickHouseNativeWriteTableSuite.scala | 19 +-
.../GlutenClickHouseMergeTreeOptimizeSuite.scala | 2 +-
...tenClickHouseMergeTreePathBasedWriteSuite.scala | 2 +-
...GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala | 4 +-
...eMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala | 4 +-
.../GlutenClickHouseMergeTreeWriteOnS3Suite.scala | 4 +-
.../GlutenClickHouseMergeTreeWriteSuite.scala | 4 +-
.../parquet/GlutenParquetFilterSuite.scala | 2 +-
.../GlutenClickHouseTPCHParquetBucketSuite.scala | 52 +----
.../tpch/GlutenClickHouseTPCHParquetRFSuite.scala | 16 +-
.../GlutenClickHouseTPCHSaltNullParquetSuite.scala | 3 +-
.../gluten/backendsapi/velox/VeloxBackend.scala | 9 +-
.../backendsapi/velox/VeloxListenerApi.scala | 7 -
.../gluten/backendsapi/velox/VeloxRuleApi.scala | 11 +-
.../functions/ArithmeticAnsiValidateSuite.scala | 15 +-
.../execution/VeloxParquetWriteForHiveSuite.scala | 210 ++++++++-------------
.../sql/execution/VeloxParquetWriteSuite.scala | 10 +-
.../apache/spark/sql/execution/WriteUtils.scala | 8 +-
.../org/apache/spark/util/SparkPlanUtil.scala | 9 +-
.../org/apache/spark/util/SparkTaskUtil.scala | 29 +--
.../org/apache/spark/util/SparkVersionUtil.scala | 5 +-
.../org/apache/gluten/execution/IcebergSuite.scala | 2 +-
.../gluten/backendsapi/BackendSettingsApi.scala | 7 +-
.../gluten/expression/ExpressionConverter.scala | 7 +-
.../columnar/rewrite/AlignExpandOutputTypes.scala | 82 --------
.../datasources/GlutenWriterColumnarRules.scala | 72 -------
.../rewrite/AlignExpandOutputTypesSuite.scala | 77 --------
.../GlutenCheckOverflowTransformerSuite.scala | 8 +-
.../org/apache/spark/sql/GlutenImplicitsTest.scala | 9 +-
.../datasource/GlutenFormatWriterInjects.scala | 6 -
41 files changed, 174 insertions(+), 643 deletions(-)
diff --git a/.github/workflows/util/install-spark-deps.sh
b/.github/workflows/util/install-spark-deps.sh
index 71ddc69e27..fda2ec8148 100755
--- a/.github/workflows/util/install-spark-deps.sh
+++ b/.github/workflows/util/install-spark-deps.sh
@@ -112,7 +112,6 @@ function setup_minio {
local spark_version_short=$(echo "${spark_version}" | cut -d '.' -f 1,2 | tr
-d '.')
case "$spark_version" in
- 3.3) hadoop_aws_version="3.3.2"; aws_sdk_artifact="aws-java-sdk-bundle";
aws_sdk_version="1.12.262" ;;
3.4|3.5*) hadoop_aws_version="3.3.4";
aws_sdk_artifact="aws-java-sdk-bundle"; aws_sdk_version="1.12.262" ;;
4.0) hadoop_aws_version="3.4.0"; aws_sdk_artifact="bundle";
aws_sdk_version="2.25.11" ;;
4.1) hadoop_aws_version="3.4.1"; aws_sdk_artifact="bundle";
aws_sdk_version="2.25.11" ;;
diff --git
a/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergMOREqualityDeletionSuite.scala
b/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergMOREqualityDeletionSuite.scala
index 09dcc50d66..6bdc104b50 100644
---
a/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergMOREqualityDeletionSuite.scala
+++
b/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergMOREqualityDeletionSuite.scala
@@ -60,7 +60,7 @@ class ClickHouseIcebergMOREqualityDeletionSuite extends
GlutenClickHouseWholeSta
FileUtils.copyDirectory(equalityDeletedData, icebergPathDir)
}
- testWithSpecifiedSparkVersion("iceberg read mor table with equality
deletion", "3.3", "3.5") {
+ testWithSpecifiedSparkVersion("iceberg read mor table with equality
deletion", "3.5") {
// The table 'test_upsert_query' was generated by Flink + Iceberg from the
iceberg ut,
// the root path must be the '/tmp/junit6640909127060857423/default'
val testTableName = "local.db.test_upsert_query"
diff --git
a/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergSuite.scala
b/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergSuite.scala
index a5d23e9f7b..12446c8759 100644
---
a/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergSuite.scala
+++
b/backends-clickhouse/src-iceberg/test/scala/org/apache/gluten/execution/iceberg/ClickHouseIcebergSuite.scala
@@ -55,7 +55,7 @@ class ClickHouseIcebergSuite extends
GlutenClickHouseWholeStageTransformerSuite
}
}
- testWithSpecifiedSparkVersion("iceberg bucketed join", "3.3", "3.5") {
+ testWithSpecifiedSparkVersion("iceberg bucketed join", "3.5") {
val leftTable = "p_str_tb"
val rightTable = "p_int_tb"
withTable(leftTable, rightTable) {
@@ -135,7 +135,7 @@ class ClickHouseIcebergSuite extends
GlutenClickHouseWholeStageTransformerSuite
}
}
- testWithSpecifiedSparkVersion("iceberg bucketed join with partition", "3.3",
"3.5") {
+ testWithSpecifiedSparkVersion("iceberg bucketed join with partition", "3.5")
{
val leftTable = "p_str_tb"
val rightTable = "p_int_tb"
withTable(leftTable, rightTable) {
@@ -371,7 +371,7 @@ class ClickHouseIcebergSuite extends
GlutenClickHouseWholeStageTransformerSuite
}
}
- testWithSpecifiedSparkVersion("iceberg bucketed join with partition filter",
"3.3", "3.5") {
+ testWithSpecifiedSparkVersion("iceberg bucketed join with partition filter",
"3.5") {
val leftTable = "p_str_tb"
val rightTable = "p_int_tb"
withTable(leftTable, rightTable) {
diff --git
a/backends-clickhouse/src-kafka/test/scala/org/apache/gluten/execution/kafka/ClickhouseGlutenKafkaScanSuite.scala
b/backends-clickhouse/src-kafka/test/scala/org/apache/gluten/execution/kafka/ClickhouseGlutenKafkaScanSuite.scala
index c6bfb9d367..c3f10350a0 100644
---
a/backends-clickhouse/src-kafka/test/scala/org/apache/gluten/execution/kafka/ClickhouseGlutenKafkaScanSuite.scala
+++
b/backends-clickhouse/src-kafka/test/scala/org/apache/gluten/execution/kafka/ClickhouseGlutenKafkaScanSuite.scala
@@ -103,8 +103,9 @@ class ClickhouseGlutenKafkaScanSuite
}
}
- // TODO: after rebase-25.12, failed with spark35 (0 did not equal 100000),
fix later
- testWithSpecifiedSparkVersion("GLUTEN-9681: test kafka data consistency",
"3.3") {
+ // Ignored: fails on Spark 3.5 after the ClickHouse 25.12 rebase (0 did not
equal 100000).
+ // TODO: fix and re-enable.
+ ignore("GLUTEN-9681: test kafka data consistency") {
withTempDir(
dir => {
val table_name = "data_consistency"
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHListenerApi.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHListenerApi.scala
index 41772f265c..d949c572c5 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHListenerApi.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHListenerApi.scala
@@ -31,7 +31,6 @@ import org.apache.spark.api.plugin.PluginContext
import org.apache.spark.internal.Logging
import org.apache.spark.listener.CHGlutenSQLAppStatusListener
import org.apache.spark.rpc.{GlutenDriverEndpoint, GlutenExecutorEndpoint}
-import org.apache.spark.sql.execution.datasources.GlutenWriterColumnarRules
import org.apache.spark.sql.execution.datasources.v1._
import org.apache.spark.sql.internal.SparkConfigUtil._
import org.apache.spark.sql.utils.ExpressionUtil
@@ -122,8 +121,6 @@ class CHListenerApi extends ListenerApi with Logging {
new CHParquetWriterInjects,
new CHOrcWriterInjects,
new CHMergeTreeWriterInjects)
- GlutenFormatFactory.injectPostRuleFactory(
- session => GlutenWriterColumnarRules.NativeWritePostRule(session))
GlutenFormatFactory.register(new CHRowSplitter())
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDecimalSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDecimalSuite.scala
index 9fc08ac7df..1e12b7a8fa 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDecimalSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDecimalSuite.scala
@@ -57,7 +57,7 @@ class GlutenClickHouseDecimalSuite extends ParquetSuite {
(DecimalType.apply(18, 8), Seq()),
// 3/10: all value is null and compare with limit
// 1 Spark 3.5
- (DecimalType.apply(38, 19), if (isSparkVersionLE("3.3")) Seq(3, 10) else
Seq(3, 10))
+ (DecimalType.apply(38, 19), Seq(3, 10))
)
private def createDecimalTables(dataType: DecimalType): Unit = {
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHBucketSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHBucketSuite.scala
index 92e27edf23..111e2987fc 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHBucketSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHBucketSuite.scala
@@ -139,18 +139,7 @@ class GlutenClickHouseTPCHBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(3)(
- df => {
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 8)
- }
- })
+ check(3)
}
}
@@ -185,18 +174,7 @@ class GlutenClickHouseTPCHBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(4)(
- df => {
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 4)
- }
- })
+ check(4)
}
}
@@ -244,18 +222,7 @@ class GlutenClickHouseTPCHBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(12)(
- df => {
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 4)
- }
- })
+ check(12)
}
}
@@ -333,18 +300,7 @@ class GlutenClickHouseTPCHBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(20)(
- df => {
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 6)
- }
- })
+ check(20)
}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseWholeStageTransformerSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseWholeStageTransformerSuite.scala
index 4febbaefca..ae8e881c98 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseWholeStageTransformerSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseWholeStageTransformerSuite.scala
@@ -65,7 +65,6 @@ class GlutenClickHouseWholeStageTransformerSuite
val CH_DEFAULT_STORAGE_DIR = "/data"
- protected def spark33: Boolean = sparkVersion.equals("3.3")
protected def spark35: Boolean = sparkVersion.equals("3.5")
def AlmostEqualsIsRel(expected: Double, actual: Double, EPSILON: Double =
DBL_EPSILON): Unit = {
@@ -155,15 +154,16 @@ class GlutenClickHouseWholeStageTransformerSuite
final override protected val resourcePath: String = "" // ch not need this
override protected val fileFormat: String = "parquet"
- protected def testSparkVersionLE33(testName: String, testTag: Tag*)(testFun:
=> Any): Unit = {
- if (isSparkVersionLE("3.3")) {
- test(testName, testTag: _*)(testFun)
- } else {
- ignore(s"[$SPARK_VERSION_SHORT]-$testName", testTag: _*)(testFun)
- }
+ /**
+ * These cases only ever passed on Spark 3.3, which is no longer supported.
They are registered as
+ * ignored rather than deleted so that they stay visible in the test report
until someone either
+ * makes them pass on a supported version or removes them.
+ */
+ protected def ignoreSpark33OnlyCase(testName: String, testTag:
Tag*)(testFun: => Any): Unit = {
+ ignore(s"[$SPARK_VERSION_SHORT]-$testName", testTag: _*)(testFun)
}
- lazy val pruningTimeValueSpark: Int = if (isSparkVersionLE("3.3")) -1 else 0
+ lazy val pruningTimeValueSpark: Int = 0
override protected def prepareTestTables(): Unit = {}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
index 602d57d4a8..6ee8dbd598 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
@@ -699,17 +699,10 @@ class GlutenFunctionValidateSuite extends
GlutenClickHouseWholeStageTransformerS
}
test("test common subexpression eliminate") {
+ // TODO: the expected operator counts at the call sites below only held on
Spark 3.3. Re-derive
+ // them for the supported versions and turn this back into a real
assertion.
def checkOperatorCount[T <: TransformSupport](count: Int)(df:
DataFrame)(implicit
- tag: ClassTag[T]): Unit = {
- if (spark33) {
- assert(
- getExecutedPlan(df).count(
- plan => {
- plan.getClass == tag.runtimeClass
- }) == count,
- s"executed plan: ${getExecutedPlan(df)}")
- }
- }
+ tag: ClassTag[T]): Unit = {}
withSQLConf((GlutenConfig.ENABLE_COMMON_SUBEXPRESSION_ELIMINATE.key,
"true")) {
// CSE in project
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/compatibility/GlutenClickhouseStringFunctionsSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/compatibility/GlutenClickhouseStringFunctionsSuite.scala
index c771260667..32f5ed4489 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/compatibility/GlutenClickhouseStringFunctionsSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/compatibility/GlutenClickhouseStringFunctionsSuite.scala
@@ -161,7 +161,7 @@ class GlutenClickhouseStringFunctionsSuite extends
GlutenClickHouseWholeStageTra
}
}
- testSparkVersionLE33("base64") {
+ ignoreSpark33OnlyCase("base64") {
// fallback on Spark-352, see https://github.com/apache/spark/pull/47303
val tableName = "base64_table"
withTable(tableName) {
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseHiveTableSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseHiveTableSuite.scala
index 4f138ef9f0..48d48fc0e7 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseHiveTableSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseHiveTableSuite.scala
@@ -1038,25 +1038,13 @@ class GlutenClickHouseHiveTableSuite
spark.sql(
s"CREATE FUNCTION my_add as " +
s"'org.apache.hadoop.hive.contrib.udf.example.UDFExampleAdd2' USING
JAR '$jarUrl'")
- if (isSparkVersionLE("3.3")) {
- runQueryAndCompare("select MY_ADD(id, id+1) from range(10)")(
- checkGlutenPlan[ProjectExecTransformer])
- } else {
- runQueryAndCompare("select MY_ADD(id, id+1) from range(10)", noFallBack
= false)(_ => {})
- }
+ runQueryAndCompare("select MY_ADD(id, id+1) from range(10)", noFallBack =
false)(_ => {})
}
+ // TODO: the expected operator counts at the call sites below only held on
Spark 3.3. Re-derive
+ // them for the supported versions and turn this back into a real assertion.
def checkOperatorCount[T <: TransformSupport](count: Int)(df:
DataFrame)(implicit
- tag: ClassTag[T]): Unit = {
- if (spark33) {
- assert(
- getExecutedPlan(df).count(
- plan => {
- plan.getClass == tag.runtimeClass
- }) == count,
- s"executed plan: ${getExecutedPlan(df)}")
- }
- }
+ tag: ClassTag[T]): Unit = {}
test("GLUTEN-4333: fix CSE in aggregate operator") {
val createTableSql =
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseNativeWriteTableSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseNativeWriteTableSuite.scala
index 6f0885951f..89ca4d7f59 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseNativeWriteTableSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/hive/GlutenClickHouseNativeWriteTableSuite.scala
@@ -275,7 +275,7 @@ class GlutenClickHouseNativeWriteTableSuite
assertResult(1)(files.length)
assert(files.head.getAbsolutePath.contains("another_date_field=2020-01-01"))
},
- isSparkVersionLE("3.3")
+ checkNative = false
)
}
}
@@ -326,8 +326,9 @@ class GlutenClickHouseNativeWriteTableSuite
}
}
- // TODO: after rebase-25.12, failed with spark35 + Orc (Memory limit
exceeded), fix later
- testWithSpecifiedSparkVersion("test 2-col partitioned table", "3.3") {
+ // Ignored: fails on Spark 3.5 with Orc after the ClickHouse 25.12 rebase
(Memory limit
+ // exceeded). TODO: fix and re-enable.
+ ignore("test 2-col partitioned table") {
val fields: ListMap[String, String] = ListMap(
("string_field", "string"),
("int_field", "int"),
@@ -550,8 +551,9 @@ class GlutenClickHouseNativeWriteTableSuite
}
}
- // TODO: after rebase-25.12, failed with spark35 + Orc (Memory limit
exceeded), fix later
- testWithMaxSparkVersion("test 1-col partitioned + 2-col bucketed table",
"3.3") {
+ // Ignored: fails on Spark 3.5 with Orc after the ClickHouse 25.12 rebase
(Memory limit
+ // exceeded). TODO: fix and re-enable.
+ ignore("test 1-col partitioned + 2-col bucketed table") {
val fields: ListMap[String, String] = ListMap(
("string_field", "string"),
("int_field", "int"),
@@ -625,8 +627,9 @@ class GlutenClickHouseNativeWriteTableSuite
}
}
- // TODO: after rebase-25.12, failed with spark35 + Orc (Memory limit
exceeded), fix later
- testWithMaxSparkVersion("test decimal with rand()", "3.3") {
+ // Ignored: fails on Spark 3.5 with Orc after the ClickHouse 25.12 rebase
(Memory limit
+ // exceeded). TODO: fix and re-enable.
+ ignore("test decimal with rand()") {
nativeWrite {
format =>
val table_name = table_name_template.format(format)
@@ -828,7 +831,7 @@ class GlutenClickHouseNativeWriteTableSuite
s"select * from $table_name",
compareResult = true,
_ => {},
- isSparkVersionLE("3.3")
+ noFallBack = false
)
)
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeOptimizeSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeOptimizeSuite.scala
index f12c58103a..eb1977af78 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeOptimizeSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeOptimizeSuite.scala
@@ -304,7 +304,7 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
assertResult(600572)(ret.apply(0).get(0))
}
- testSparkVersionLE33("test mergetree optimize table with partition and
bucket") {
+ ignoreSpark33OnlyCase("test mergetree optimize table with partition and
bucket") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_optimize_p6;
|""".stripMargin)
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreePathBasedWriteSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreePathBasedWriteSuite.scala
index 2db25acbb8..2dfcca994c 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreePathBasedWriteSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreePathBasedWriteSuite.scala
@@ -652,7 +652,7 @@ class GlutenClickHouseMergeTreePathBasedWriteSuite extends
CreateMergeTreeSuite
}
}
- testSparkVersionLE33("test mergetree path based write with bucket table") {
+ ignoreSpark33OnlyCase("test mergetree path based write with bucket table") {
val dataPath = s"$dataHome/lineitem_mergetree_bucket"
clearDataPath(dataPath)
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
index b143d29428..eca1529ea5 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala
@@ -388,7 +388,7 @@ class GlutenClickHouseMergeTreeWriteOnHDFSSuite extends
CreateMergeTreeSuite {
spark.sql("select * from partition_escape").show()
}
- testSparkVersionLE33("test mergetree write with bucket table") {
+ ignoreSpark33OnlyCase("test mergetree write with bucket table") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_bucket_hdfs;
|""".stripMargin)
@@ -458,7 +458,7 @@ class GlutenClickHouseMergeTreeWriteOnHDFSSuite extends
CreateMergeTreeSuite {
spark.sql("drop table lineitem_mergetree_bucket_hdfs")
}
- testSparkVersionLE33("test mergetree write with the path based bucket
table") {
+ ignoreSpark33OnlyCase("test mergetree write with the path based bucket
table") {
val dataPath = s"$remotePath/lineitem_mergetree_bucket_hdfs"
val sourceDF = spark.sql(s"""
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala
index 76e55443f8..9230e93db2 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala
@@ -337,7 +337,7 @@ class
GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite extends CreateMer
spark.sql("drop table lineitem_mergetree_partition_hdfs")
}
- testSparkVersionLE33("test mergetree write with bucket table") {
+ ignoreSpark33OnlyCase("test mergetree write with bucket table") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_bucket_hdfs;
|""".stripMargin)
@@ -407,7 +407,7 @@ class
GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite extends CreateMer
spark.sql("drop table lineitem_mergetree_bucket_hdfs purge")
}
- testSparkVersionLE33("test mergetree write with the path based bucket
table") {
+ ignoreSpark33OnlyCase("test mergetree write with the path based bucket
table") {
val dataPath = s"$remotePath/lineitem_mergetree_bucket_hdfs"
val sourceDF = spark.sql(s"""
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnS3Suite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnS3Suite.scala
index 0659b9e6d7..b6e0da5918 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnS3Suite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteOnS3Suite.scala
@@ -378,7 +378,7 @@ class GlutenClickHouseMergeTreeWriteOnS3Suite extends
CreateMergeTreeSuite {
}
- testSparkVersionLE33("test mergetree write with bucket table") {
+ ignoreSpark33OnlyCase("test mergetree write with bucket table") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_bucket_s3;
|""".stripMargin)
@@ -448,7 +448,7 @@ class GlutenClickHouseMergeTreeWriteOnS3Suite extends
CreateMergeTreeSuite {
spark.sql("drop table lineitem_mergetree_bucket_s3")
}
- testSparkVersionLE33("test mergetree write with the path based bucket
table") {
+ ignoreSpark33OnlyCase("test mergetree write with the path based bucket
table") {
val dataPath = s"s3a://$BUCKET_NAME/lineitem_mergetree_bucket_s3"
val sourceDF = spark.sql(s"""
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
index 53b725cfbe..64931c3f4b 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
@@ -720,7 +720,7 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
}
}
- testSparkVersionLE33("test mergetree write with bucket table") {
+ ignoreSpark33OnlyCase("test mergetree write with bucket table") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_bucket;
|""".stripMargin)
@@ -1596,7 +1596,7 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
}
}
- testSparkVersionLE33("test mergetree with primary keys filter pruning by
driver with bucket") {
+ ignoreSpark33OnlyCase("test mergetree with primary keys filter pruning by
driver with bucket") {
spark.sql(s"""
|DROP TABLE IF EXISTS
lineitem_mergetree_pk_pruning_by_driver_bucket;
|""".stripMargin)
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/parquet/GlutenParquetFilterSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/parquet/GlutenParquetFilterSuite.scala
index b663638cae..cc137c2642 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/parquet/GlutenParquetFilterSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/parquet/GlutenParquetFilterSuite.scala
@@ -474,7 +474,7 @@ class GlutenParquetFilterSuite
tpchQueries.zipWithIndex.foreach {
case (q, i) =>
if (q == "q2" || q == "q9") {
- testSparkVersionLE33(q) {
+ ignoreSpark33OnlyCase(q) {
runTest(i)
}
} else {
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetBucketSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetBucketSuite.scala
index 31b31cfc2c..512614f102 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetBucketSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetBucketSuite.scala
@@ -174,18 +174,7 @@ class GlutenClickHouseTPCHParquetBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(3) {
- df =>
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 4)
- }
- }
+ check(3)
}
}
@@ -220,18 +209,7 @@ class GlutenClickHouseTPCHParquetBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(4) {
- df =>
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 2)
- }
- }
+ check(4)
}
}
@@ -279,18 +257,7 @@ class GlutenClickHouseTPCHParquetBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(12) {
- df =>
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 2)
- }
- }
+ check(12)
}
}
@@ -368,18 +335,7 @@ class GlutenClickHouseTPCHParquetBucketSuite
withSQLConf(
("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold",
"1KB"),
("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")) {
- customCheck(20) {
- df =>
- if (spark33) {
- val plans = collectWithSubqueries(df.queryExecution.executedPlan) {
- case aggExec: HashAggregateExecBaseTransformer
- if aggExec.aggregateExpressions.exists(
-
_.aggregateFunction.getClass.getSimpleName.equals("BloomFilterAggregate")) =>
- aggExec
- }
- assert(plans.size == 3)
- }
- }
+ check(20)
}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetRFSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetRFSuite.scala
index b78a1cd364..9258a7707d 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetRFSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHParquetRFSuite.scala
@@ -16,8 +16,6 @@
*/
package org.apache.gluten.execution.tpch
-import org.apache.gluten.execution._
-
import org.apache.spark.SparkConf
class GlutenClickHouseTPCHParquetRFSuite extends
GlutenClickHouseTPCHSaltNullParquetSuite {
@@ -55,19 +53,7 @@ class GlutenClickHouseTPCHParquetRFSuite extends
GlutenClickHouseTPCHSaltNullPar
|
|""".stripMargin,
compareResult = true,
- df => {
- if (spark33) {
- val filterExecs = df.queryExecution.executedPlan.collect {
- case filter: FilterExecTransformerBase => filter
- }
- assert(filterExecs.size == 4)
- assert(
- filterExecs.head
- .asInstanceOf[FilterExecTransformer]
- .toString
- .contains("might_contain"))
- }
- }
+ NOOP
)
}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
index f6277c214f..8c4e55311b 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpch/GlutenClickHouseTPCHSaltNullParquetSuite.scala
@@ -1216,7 +1216,8 @@ class GlutenClickHouseTPCHSaltNullParquetSuite
}
}
- testSparkVersionLE33("test posexplode issue:
https://github.com/oap-project/gluten/issues/1767";) {
+ ignoreSpark33OnlyCase(
+ "test posexplode issue:
https://github.com/oap-project/gluten/issues/1767";) {
spark.sql("create table test_1767 (id bigint, data map<string, string>)
using parquet")
spark.sql("INSERT INTO test_1767 values(1, map('k', 'v'))")
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
index 197f708112..6458a84a6e 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
@@ -39,8 +39,7 @@ import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.execution.{ColumnarCachedBatchSerializer,
SparkPlan}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
-import
org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand
-import org.apache.spark.sql.execution.datasources.{FileFormat,
InsertIntoHadoopFsRelationCommand}
+import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat,
ParquetOptions}
import org.apache.spark.sql.hive.execution.HiveFileFormat
import org.apache.spark.sql.internal.SQLConf
@@ -536,12 +535,6 @@ object VeloxBackendSettings extends BackendSettingsApi {
override def insertPostProjectForGenerate(): Boolean = true
- override def skipNativeCtas(ctas: CreateDataSourceTableAsSelectCommand):
Boolean = true
-
- override def skipNativeInsertInto(insertInto:
InsertIntoHadoopFsRelationCommand): Boolean = {
- insertInto.bucketSpec.nonEmpty
- }
-
override def alwaysFailOnMapExpression(): Boolean = true
override def requiredChildOrderingForWindowGroupLimit(): Boolean = false
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
index 02392bb886..597882479d 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala
@@ -30,7 +30,6 @@ import org.apache.gluten.memory.{MemoryUsageRecorder,
SimpleMemoryUsageRecorder}
import org.apache.gluten.memory.listener.ReservationListener
import org.apache.gluten.memory.memtarget.MemoryTarget
import org.apache.gluten.monitor.VeloxMemoryProfiler
-import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.udf.UdfJniWrapper
import org.apache.gluten.utils._
@@ -44,7 +43,6 @@ import org.apache.spark.rpc.{GlutenDriverEndpoint,
GlutenExecutorEndpoint}
import org.apache.spark.shuffle.{ColumnarShuffleDependency, LookupKey,
ShuffleManagerRegistry}
import org.apache.spark.shuffle.sort.ColumnarShuffleManager
import org.apache.spark.sql.execution.ColumnarCachedBatchSerializer
-import org.apache.spark.sql.execution.datasources.GlutenWriterColumnarRules
import
org.apache.spark.sql.execution.datasources.velox.{VeloxParquetWriterInjects,
VeloxRowSplitter}
import org.apache.spark.sql.expression.UDFResolver
import org.apache.spark.sql.internal.{GlutenConfigUtil, StaticSQLConf}
@@ -239,11 +237,6 @@ class VeloxListenerApi extends ListenerApi with Logging {
// Inject backend-specific implementations to override spark classes.
GlutenFormatFactory.register(new VeloxParquetWriterInjects)
- // Only register NativeWritePostRule for Spark 3.3
- if (SparkShimLoader.getSparkVersion.startsWith("3.3")) {
- GlutenFormatFactory.injectPostRuleFactory(
- session => GlutenWriterColumnarRules.NativeWritePostRule(session))
- }
GlutenFormatFactory.register(new VeloxRowSplitter())
}
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
index 8dfa38dc3d..11d1944590 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
@@ -34,7 +34,6 @@ import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.noop.GlutenNoopWriterRule
-import org.apache.spark.util.SparkVersionUtil
class VeloxRuleApi extends RuleApi {
import VeloxRuleApi._
@@ -104,12 +103,10 @@ object VeloxRuleApi {
Seq(
RewriteIn,
RewriteMultiChildrenCount,
- RewriteJoin) ++
- (if (SparkVersionUtil.eqSpark33) Seq(AlignExpandOutputTypes) else
Seq.empty) ++
- Seq(
- PullOutPreProject,
- PullOutPostProject,
- ProjectColumnPruning)
+ RewriteJoin,
+ PullOutPreProject,
+ PullOutPostProject,
+ ProjectColumnPruning)
injector.injectTransform(
c =>
HeuristicTransform.WithRewrites(
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala
index 25eefb9ff0..9b7a14a51f 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala
@@ -78,13 +78,8 @@ class ArithmeticAnsiValidateSuite extends
FunctionsValidateSuite {
runQueryAndCompare("SELECT int_field1 / 2 FROM datatab WHERE int_field1 IS
NOT NULL") {
checkGlutenPlan[ProjectExecTransformer]
}
- if (isSparkVersionGE("3.4")) {
- // Spark 3.4+ throws exception for division by zero in ANSI mode
- intercept[SparkException] {
- sql("SELECT 1 / 0").collect()
- }
- } else {
- // Spark 3.3 doesn't throw exception for division by zero in ANSI mode
+ // Spark 3.4+ throws exception for division by zero in ANSI mode
+ intercept[SparkException] {
sql("SELECT 1 / 0").collect()
}
}
@@ -93,10 +88,8 @@ class ArithmeticAnsiValidateSuite extends
FunctionsValidateSuite {
runQueryAndCompare("SELECT int_field1 div 2 FROM datatab WHERE int_field1
IS NOT NULL") {
checkGlutenPlan[ProjectExecTransformer]
}
- if (isSparkVersionGE("3.4")) {
- intercept[SparkException] {
- sql("SELECT 1 div 0 ").collect()
- }
+ intercept[SparkException] {
+ sql("SELECT 1 div 0 ").collect()
}
}
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteForHiveSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteForHiveSuite.scala
index cdd36201ec..4dad62e1d1 100644
---
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteForHiveSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteForHiveSuite.scala
@@ -238,21 +238,14 @@ class VeloxParquetWriteForHiveSuite extends
GlutenQueryTest with SQLTestUtils wi
test("test hive write dir") {
withTempPath {
f =>
- // compatible with Spark3.3 and later
+ // compatible with Spark3.4 and later
withSQLConf("spark.sql.hive.convertMetastoreInsertDir" -> "false") {
- if (isSparkVersionGE("3.4")) {
- checkNativeWrite(
- s"""
- |INSERT OVERWRITE DIRECTORY '${f.getCanonicalPath}' STORED AS
PARQUET SELECT 1 as c
- |""".stripMargin,
- expectNative = false
- )
- } else {
- checkNativeWrite(
- s"""
- |INSERT OVERWRITE DIRECTORY '${f.getCanonicalPath}' STORED AS
PARQUET SELECT 1 as c
- |""".stripMargin)
- }
+ checkNativeWrite(
+ s"""
+ |INSERT OVERWRITE DIRECTORY '${f.getCanonicalPath}' STORED AS
PARQUET SELECT 1 as c
+ |""".stripMargin,
+ expectNative = false
+ )
checkAnswer(spark.read.parquet(f.getCanonicalPath), Row(1))
}
}
@@ -274,7 +267,11 @@ class VeloxParquetWriteForHiveSuite extends
GlutenQueryTest with SQLTestUtils wi
}
}
- test("native writer should respect table properties") {
+ // Ignored: the native writer uses spark.sql.parquet.compression.codec
instead of the
+ // parquet.compression table property, so this writes GZIP where zstd is
expected. The
+ // comparison below used to be a discarded Boolean, which is why the case
passed before.
+ // TODO: fix and re-enable, see #12987.
+ ignore("native writer should respect table properties") {
Seq(true, false).foreach {
enableNativeWrite =>
withSQLConf(GlutenConfig.NATIVE_WRITER_ENABLED.key ->
enableNativeWrite.toString) {
@@ -298,7 +295,11 @@ class VeloxParquetWriteForHiveSuite extends
GlutenQueryTest with SQLTestUtils wi
reader =>
val column =
reader.getFooter.getBlocks.get(0).getColumns.get(0)
// native writer and vanilla spark hive writer should be
consistent
- "zstd".equalsIgnoreCase(column.getCodec.toString)
+ assert(
+ "zstd".equalsIgnoreCase(column.getCodec.toString),
+ s"expected zstd but got ${column.getCodec} in
${files.head.getPath}, " +
+ s"nativeWriteEnabled=$enableNativeWrite"
+ )
}
}
}
@@ -307,138 +308,93 @@ class VeloxParquetWriteForHiveSuite extends
GlutenQueryTest with SQLTestUtils wi
}
test("Native writer support compatible hive bucket write with dynamic
partition") {
- if (isSparkVersionGE("3.4")) {
- Seq("true", "false").foreach {
- enableConvertMetastore =>
- withSQLConf("spark.sql.hive.convertMetastoreParquet" ->
enableConvertMetastore) {
- val source = "hive_source_table"
- val target = "hive_bucketed_table"
- withTable(source, target) {
- sql(s"""
- |CREATE TABLE IF NOT EXISTS $target (i int, j string)
- |PARTITIONED BY(k string)
- |CLUSTERED BY (i, j) SORTED BY (i) INTO 8 BUCKETS
- |STORED AS PARQUET
+ Seq("true", "false").foreach {
+ enableConvertMetastore =>
+ withSQLConf("spark.sql.hive.convertMetastoreParquet" ->
enableConvertMetastore) {
+ val source = "hive_source_table"
+ val target = "hive_bucketed_table"
+ withTable(source, target) {
+ sql(s"""
+ |CREATE TABLE IF NOT EXISTS $target (i int, j string)
+ |PARTITIONED BY(k string)
+ |CLUSTERED BY (i, j) SORTED BY (i) INTO 8 BUCKETS
+ |STORED AS PARQUET
""".stripMargin)
- val df =
- (0 until 50).map(i => (i % 13, i.toString, i % 5)).toDF("i",
"j", "k")
- df.write.mode(SaveMode.Overwrite).saveAsTable(source)
+ val df =
+ (0 until 50).map(i => (i % 13, i.toString, i % 5)).toDF("i",
"j", "k")
+ df.write.mode(SaveMode.Overwrite).saveAsTable(source)
- withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
- checkNativeWrite(s"INSERT INTO $target SELECT * FROM $source")
- }
+ withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
+ checkNativeWrite(s"INSERT INTO $target SELECT * FROM $source")
+ }
- for (k <- 0 until 5) {
- testBucketing(
- new File(tableDir(target), s"k=$k"),
- "parquet",
- 8,
- Seq("i", "j"),
- Seq("i"),
- df,
- bucketIdExpression,
- getBucketIdFromFileName)
- }
+ for (k <- 0 until 5) {
+ testBucketing(
+ new File(tableDir(target), s"k=$k"),
+ "parquet",
+ 8,
+ Seq("i", "j"),
+ Seq("i"),
+ df,
+ bucketIdExpression,
+ getBucketIdFromFileName)
}
}
- }
+ }
}
}
test("bucket writer with non-dynamic partition") {
- if (isSparkVersionGE("3.4")) {
- Seq("true", "false").foreach {
- enableConvertMetastore =>
- withSQLConf("spark.sql.hive.convertMetastoreParquet" ->
enableConvertMetastore) {
- val source = "hive_source_table"
- val target = "hive_bucketed_table"
- withTable(source, target) {
- sql(s"""
- |CREATE TABLE IF NOT EXISTS $target (i int, j string)
- |PARTITIONED BY(k string)
- |CLUSTERED BY (i, j) SORTED BY (i) INTO 8 BUCKETS
- |STORED AS PARQUET
+ Seq("true", "false").foreach {
+ enableConvertMetastore =>
+ withSQLConf("spark.sql.hive.convertMetastoreParquet" ->
enableConvertMetastore) {
+ val source = "hive_source_table"
+ val target = "hive_bucketed_table"
+ withTable(source, target) {
+ sql(s"""
+ |CREATE TABLE IF NOT EXISTS $target (i int, j string)
+ |PARTITIONED BY(k string)
+ |CLUSTERED BY (i, j) SORTED BY (i) INTO 8 BUCKETS
+ |STORED AS PARQUET
""".stripMargin)
- val df =
- (0 until 50).map(i => (i % 13, i.toString, i % 5)).toDF("i",
"j", "k")
- df.write.mode(SaveMode.Overwrite).saveAsTable(source)
-
- // hive relation convert always use dynamic, so it will offload
to native.
- checkNativeWrite(s"INSERT INTO $target PARTITION(k='0') SELECT
i, j FROM $source")
- val files = tableDir(target)
- .listFiles()
- .filterNot(f => f.getName.startsWith(".") ||
f.getName.startsWith("_"))
- assert(files.length == 1 && files.head.getName.contains("k=0"))
- checkAnswer(spark.table(target).select("i", "j"), df.select("i",
"j"))
- }
+ val df =
+ (0 until 50).map(i => (i % 13, i.toString, i % 5)).toDF("i",
"j", "k")
+ df.write.mode(SaveMode.Overwrite).saveAsTable(source)
+
+ // hive relation convert always use dynamic, so it will offload to
native.
+ checkNativeWrite(s"INSERT INTO $target PARTITION(k='0') SELECT i,
j FROM $source")
+ val files = tableDir(target)
+ .listFiles()
+ .filterNot(f => f.getName.startsWith(".") ||
f.getName.startsWith("_"))
+ assert(files.length == 1 && files.head.getName.contains("k=0"))
+ checkAnswer(spark.table(target).select("i", "j"), df.select("i",
"j"))
}
- }
+ }
}
}
test("bucket writer with non-partition table") {
- if (isSparkVersionGE("3.4")) {
- Seq("true", "false").foreach {
- enableConvertMetastore =>
- withSQLConf("spark.sql.hive.convertMetastoreParquet" ->
enableConvertMetastore) {
- val source = "hive_source_table"
- val target = "hive_bucketed_table"
- withTable(source, target) {
- sql(s"""
- |CREATE TABLE IF NOT EXISTS $target (i int, j string)
- |CLUSTERED BY (i, j) SORTED BY (i) INTO 8 BUCKETS
- |STORED AS PARQUET
+ Seq("true", "false").foreach {
+ enableConvertMetastore =>
+ withSQLConf("spark.sql.hive.convertMetastoreParquet" ->
enableConvertMetastore) {
+ val source = "hive_source_table"
+ val target = "hive_bucketed_table"
+ withTable(source, target) {
+ sql(s"""
+ |CREATE TABLE IF NOT EXISTS $target (i int, j string)
+ |CLUSTERED BY (i, j) SORTED BY (i) INTO 8 BUCKETS
+ |STORED AS PARQUET
""".stripMargin)
- val df =
- (0 until 50).map(i => (i % 13, i.toString)).toDF("i", "j")
- df.write.mode(SaveMode.Overwrite).saveAsTable(source)
+ val df =
+ (0 until 50).map(i => (i % 13, i.toString)).toDF("i", "j")
+ df.write.mode(SaveMode.Overwrite).saveAsTable(source)
- checkNativeWrite(s"INSERT INTO $target SELECT i, j FROM $source")
+ checkNativeWrite(s"INSERT INTO $target SELECT i, j FROM $source")
- checkAnswer(spark.table(target), df)
- }
- }
- }
- }
- }
-
- testWithMaxSparkVersion(
- "Native writer should keep the same compression codec if
`hive.exec.compress.output` is true",
- "3.3") {
- Seq(false, true).foreach {
- enableNativeWrite =>
- withSQLConf(GlutenConfig.NATIVE_WRITER_ENABLED.key ->
enableNativeWrite.toString) {
- withTable("t") {
- withSQLConf(
- "spark.sql.hive.convertMetastoreParquet" -> "false",
- "spark.sql.parquet.compression.codec" -> "gzip") {
- spark.sql("SET hive.exec.compress.output=true")
- spark.sql("SET parquet.compression=gzip")
- spark.sql(
- "SET
mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec")
- checkNativeWrite(
- "CREATE TABLE t STORED AS PARQUET TBLPROPERTIES
('parquet.compression'='zstd') " +
- "AS SELECT 1 as c",
- expectNative = enableNativeWrite)
- val tableDir = new
Path(s"${conf.getConf(StaticSQLConf.WAREHOUSE_PATH)}/t")
- val configuration = spark.sessionState.newHadoopConf()
- val files = tableDir
- .getFileSystem(configuration)
- .listStatus(tableDir)
- .filterNot(_.getPath.getName.startsWith("\\."))
- assert(files.nonEmpty)
- val in = HadoopInputFile.fromStatus(files.head,
spark.sessionState.newHadoopConf())
- Utils.tryWithResource(ParquetFileReader.open(in)) {
- reader =>
- val compression =
-
reader.getFooter.getBlocks.get(0).getColumns.get(0).getCodec.toString
- // native writer and vanilla spark hive writer should be
consistent
- assert("zstd".equalsIgnoreCase(compression))
- }
- }
+ checkAnswer(spark.table(target), df)
}
}
}
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala
index 9a6d423ed2..ef7ff5d23c 100644
---
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala
@@ -72,7 +72,7 @@ class VeloxParquetWriteSuite extends
VeloxWholeStageTransformerSuite with WriteU
// TODO: maybe remove constant complex type restriction (Spark 3.4+)
checkNativeWrite(
s"INSERT OVERWRITE DIRECTORY '$path' USING PARQUET SELECT
array(struct(1), null) as var1",
- expectNative = !isSparkVersionGE("3.4"))
+ expectNative = false)
}
}
@@ -178,7 +178,7 @@ class VeloxParquetWriteSuite extends
VeloxWholeStageTransformerSuite with WriteU
// TODO: maybe remove constant complex type restriction (Spark 3.4+)
checkNativeWrite(
"INSERT INTO src SELECT array(1, 2, 3)",
- expectNative = !isSparkVersionGE("3.4"))
+ expectNative = false)
spark.sql("CREATE TABLE t (ids ARRAY<INT>) USING PARQUET")
checkNativeWrite("INSERT INTO t SELECT ids FROM src")
checkAnswer(spark.table("t"), Row(Seq(1, 2, 3)))
@@ -191,7 +191,7 @@ class VeloxParquetWriteSuite extends
VeloxWholeStageTransformerSuite with WriteU
// TODO: maybe remove constant complex type restriction (Spark 3.4+)
checkNativeWrite(
"INSERT INTO src SELECT map('a', 1, 'b', 2)",
- expectNative = !isSparkVersionGE("3.4"))
+ expectNative = false)
spark.sql("CREATE TABLE t (kv MAP<STRING, INT>) USING PARQUET")
checkNativeWrite("INSERT INTO t SELECT kv FROM src")
checkAnswer(spark.table("t"), Row(Map("a" -> 1, "b" -> 2)))
@@ -219,9 +219,7 @@ class VeloxParquetWriteSuite extends
VeloxWholeStageTransformerSuite with WriteU
.range(100)
.toDF("id")
.createOrReplaceTempView("ctas_temp")
- checkNativeWrite(
- "CREATE TABLE velox_ctas USING PARQUET AS SELECT * FROM ctas_temp",
- expectNative = isSparkVersionGE("3.4"))
+ checkNativeWrite("CREATE TABLE velox_ctas USING PARQUET AS SELECT * FROM
ctas_temp")
}
}
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/WriteUtils.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/WriteUtils.scala
index 416ce41f57..bbd738f36c 100644
---
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/WriteUtils.scala
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/WriteUtils.scala
@@ -16,8 +16,6 @@
*/
package org.apache.spark.sql.execution
-import org.apache.gluten.execution.VeloxColumnarToCarrierRowExec
-
import org.apache.spark.sql.{DataFrame, GlutenQueryTest}
import org.apache.spark.sql.catalyst.expressions.{BitwiseAnd, Expression,
HiveHash, Literal, Pmod, UnsafeProjection}
import org.apache.spark.sql.functions._
@@ -39,11 +37,7 @@ trait WriteUtils extends GlutenQueryTest with SQLTestUtils {
override def onFailure(f: String, qe: QueryExecution, e: Exception):
Unit = {}
override def onSuccess(funcName: String, qe: QueryExecution, duration:
Long): Unit = {
if (!nativeUsed) {
- nativeUsed = if (isSparkVersionGE("3.4")) {
- qe.executedPlan.exists(_.isInstanceOf[ColumnarWriteFilesExec])
- } else {
-
qe.executedPlan.exists(_.isInstanceOf[VeloxColumnarToCarrierRowExec])
- }
+ nativeUsed =
qe.executedPlan.exists(_.isInstanceOf[ColumnarWriteFilesExec])
}
}
}
diff --git
a/gluten-core/src/main/scala/org/apache/spark/util/SparkPlanUtil.scala
b/gluten-core/src/main/scala/org/apache/spark/util/SparkPlanUtil.scala
index 5ce14f0904..c23327c99c 100644
--- a/gluten-core/src/main/scala/org/apache/spark/util/SparkPlanUtil.scala
+++ b/gluten-core/src/main/scala/org/apache/spark/util/SparkPlanUtil.scala
@@ -18,6 +18,7 @@ package org.apache.spark.util
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command.DataWritingCommandExec
+import org.apache.spark.sql.execution.datasources.V1WriteCommand
import org.apache.spark.sql.internal.SQLConf
object SparkPlanUtil {
@@ -27,14 +28,8 @@ object SparkPlanUtil {
}
def isPlannedV1Write(plan: DataWritingCommandExec): Boolean = {
- if (SparkVersionUtil.eqSpark33) {
- return false
- }
-
- val v1WriteCommandClass =
-
Utils.classForName("org.apache.spark.sql.execution.datasources.V1WriteCommand")
val plannedWriteEnabled =
SQLConf.get.getConfString("spark.sql.optimizer.plannedWrite.enabled",
"true").toBoolean
- v1WriteCommandClass.isAssignableFrom(plan.cmd.getClass) &&
plannedWriteEnabled
+ plan.cmd.isInstanceOf[V1WriteCommand] && plannedWriteEnabled
}
}
diff --git
a/gluten-core/src/main/scala/org/apache/spark/util/SparkTaskUtil.scala
b/gluten-core/src/main/scala/org/apache/spark/util/SparkTaskUtil.scala
index a19579e433..46fb999ff0 100644
--- a/gluten-core/src/main/scala/org/apache/spark/util/SparkTaskUtil.scala
+++ b/gluten-core/src/main/scala/org/apache/spark/util/SparkTaskUtil.scala
@@ -49,40 +49,25 @@ object SparkTaskUtil {
val partitionId = -1.asInstanceOf[Object]
val taskAttemptId = -1L.asInstanceOf[Object]
val attemptNumber = -1.asInstanceOf[Object]
- val numPartitions = -1.asInstanceOf[Object] // Added in Spark 3.4.
+ val numPartitions = -1.asInstanceOf[Object]
val taskMemoryManager = new TaskMemoryManager(memoryManager,
-1L).asInstanceOf[Object]
val localProperties = properties.asInstanceOf[Object]
val metricsSystem =
MetricsSystem.createMetricsSystem("GLUTEN_UNSAFE",
conf).asInstanceOf[Object]
val taskMetrics = TaskMetrics.empty.asInstanceOf[Object]
- val cpus = 1.asInstanceOf[Object] // Added in Spark 3.3.
+ val cpus = 1.asInstanceOf[Object]
val resources = Map.empty.asInstanceOf[Object]
val ctor = {
val ctors = classOf[TaskContextImpl].getDeclaredConstructors
- assert(ctors.size == 1)
+ require(
+ ctors.size == 1,
+ s"Expected TaskContextImpl to declare exactly one constructor, found
${ctors.size}")
ctors.head
}
- if (SparkVersionUtil.eqSpark33) {
- return ctor
- .newInstance(
- stageId,
- stageAttemptNumber,
- partitionId,
- taskAttemptId,
- attemptNumber,
- taskMemoryManager,
- localProperties,
- metricsSystem,
- taskMetrics,
- cpus,
- resources
- )
- .asInstanceOf[TaskContext]
- }
-
- // Since Spark 3.4.
+ // The arguments below are positional: their order must match
TaskContextImpl's primary
+ // constructor.
ctor
.newInstance(
stageId,
diff --git
a/gluten-core/src/main/scala/org/apache/spark/util/SparkVersionUtil.scala
b/gluten-core/src/main/scala/org/apache/spark/util/SparkVersionUtil.scala
index 97d428515c..f7326e98d6 100644
--- a/gluten-core/src/main/scala/org/apache/spark/util/SparkVersionUtil.scala
+++ b/gluten-core/src/main/scala/org/apache/spark/util/SparkVersionUtil.scala
@@ -17,10 +17,7 @@
package org.apache.spark.util
object SparkVersionUtil {
- private val comparedWithSpark33 = compareMajorMinorVersion((3, 3))
- private val comparedWithSpark35 = compareMajorMinorVersion((3, 5))
- val eqSpark33: Boolean = comparedWithSpark33 == 0
- val gteSpark35: Boolean = comparedWithSpark35 >= 0
+ val gteSpark35: Boolean = compareMajorMinorVersion((3, 5)) >= 0
val gteSpark40: Boolean = compareMajorMinorVersion((4, 0)) >= 0
val gteSpark41: Boolean = compareMajorMinorVersion((4, 1)) >= 0
diff --git
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
index 56f3fbdace..0ee193ce0f 100644
---
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
+++
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
@@ -619,7 +619,7 @@ abstract class IcebergSuite extends
WholeStageTransformerSuite {
// Spark configuration spark.sql.iceberg.handle-timestamp-without-timezone
is not supported
// in Spark 3.4
- testWithSpecifiedSparkVersion("iceberg partition type - timestamp", "3.3",
"3.5") {
+ testWithSpecifiedSparkVersion("iceberg partition type - timestamp", "3.5") {
Seq("true", "false").foreach {
flag =>
withSQLConf(
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
index cf2026af63..d596c2dfad 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
@@ -27,8 +27,7 @@ import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.connector.read.Scan
-import
org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand
-import org.apache.spark.sql.execution.datasources.{FileFormat,
InsertIntoHadoopFsRelationCommand}
+import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.types.{StructField, StructType}
import org.apache.hadoop.conf.Configuration
@@ -123,10 +122,6 @@ trait BackendSettingsApi {
def insertPostProjectForGenerate(): Boolean = false
- def skipNativeCtas(ctas: CreateDataSourceTableAsSelectCommand): Boolean =
false
-
- def skipNativeInsertInto(insertInto: InsertIntoHadoopFsRelationCommand):
Boolean = false
-
def alwaysFailOnMapExpression(): Boolean = false
def requiredChildOrderingForWindowGroupLimit(): Boolean = true
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
index 6c6a463564..3cffa2cf98 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
@@ -570,16 +570,15 @@ object ExpressionConverter extends SQLConfHelper with
Logging {
),
r
)
- case instr: TernaryExpression if
instr.getClass.getSimpleName.equals("RegExpInStr") =>
+ case instr: RegExpInStr =>
// Spark's RegExpInStr carries a third `idx` child but ignores it
during
// evaluation (it always returns the start position of the whole
match).
// Velox's regexp_instr only takes (subject, regexp), so drop the idx
child.
- // Matched by class name because RegExpInStr does not exist in Spark
3.3.
GenericExpressionTransformer(
substraitExprName,
Seq(
- replaceWithExpressionTransformer0(instr.first, attributeSeq,
expressionsMap),
- replaceWithExpressionTransformer0(instr.second, attributeSeq,
expressionsMap)
+ replaceWithExpressionTransformer0(instr.subject, attributeSeq,
expressionsMap),
+ replaceWithExpressionTransformer0(instr.regexp, attributeSeq,
expressionsMap)
),
instr
)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypes.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypes.scala
deleted file mode 100644
index 1fe963ab45..0000000000
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypes.scala
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * 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.gluten.extension.columnar.rewrite
-
-import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.execution.{ExpandExec, SparkPlan}
-import org.apache.spark.sql.types.DataType
-
-/**
- * Spark 3.3 may produce Expand projections whose expression output types do
not exactly match the
- * corresponding Expand output attributes. Spark's row path tolerates this,
but native Expand
- * conversion requires each projection column to have a consistent type.
- *
- * This rule rewrites each projection column: null literals are replaced with
a typed null matching
- * the output type; non-matching expressions are wrapped in a Cast to the
output type.
- */
-object AlignExpandOutputTypes extends RewriteSingleNode {
- override def isRewritable(plan: SparkPlan): Boolean = {
- plan match {
- case _: ExpandExec => true
- case _ => false
- }
- }
-
- override def rewrite(plan: SparkPlan): SparkPlan = plan match {
- case expand: ExpandExec =>
- val alignedProjections = ExpandOutputTypeAlignment.alignProjections(
- expand.projections,
- expand.output,
- expand.child.output)
- if (alignedProjections == expand.projections) {
- expand
- } else {
- val newExpand = expand.copy(projections = alignedProjections)
- newExpand.copyTagsFrom(expand)
- newExpand
- }
- case _ => plan
- }
-}
-
-private[gluten] object ExpandOutputTypeAlignment {
- def alignProjections(
- projections: Seq[Seq[Expression]],
- output: Seq[Attribute],
- inputAttributes: Seq[Attribute]): Seq[Seq[Expression]] = {
- projections.map {
- projection =>
- projection.zipWithIndex.map {
- case (expression, colIdx) if colIdx < output.length =>
- alignExpression(expression, output(colIdx).dataType)
- case (expression, _) =>
- expression
- }
- }
- }
-
- private def alignExpression(expression: Expression, outputType: DataType):
Expression = {
- expression match {
- case Literal(null, _) =>
- Literal.create(null, outputType)
- case _ if expression.dataType != outputType =>
- Cast(expression, outputType)
- case _ =>
- expression
- }
- }
-}
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/datasources/GlutenWriterColumnarRules.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/datasources/GlutenWriterColumnarRules.scala
index 346d6b7f4d..494217d4f7 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/datasources/GlutenWriterColumnarRules.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/datasources/GlutenWriterColumnarRules.scala
@@ -18,55 +18,12 @@ package org.apache.spark.sql.execution.datasources
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.execution.ColumnarToRowExecBase
-import org.apache.gluten.execution.datasource.GlutenFormatFactory
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
-import
org.apache.spark.sql.execution.command.{CreateDataSourceTableAsSelectCommand,
DataWritingCommand, DataWritingCommandExec}
-import org.apache.spark.sql.hive.execution.{CreateHiveTableAsSelectCommand,
InsertIntoHiveDirCommand, InsertIntoHiveTable}
-import org.apache.spark.sql.sources.DataSourceRegister
object GlutenWriterColumnarRules {
- // TODO: support ctas in Spark3.4, see
https://github.com/apache/spark/pull/39220
- // TODO: support dynamic partition and bucket write
- // 1. pull out `Empty2Null` and required ordering to `WriteFilesExec`, see
Spark3.4 `V1Writes`
- // 2. support detect partition value, partition path, bucket value, bucket
path at native side,
- // see `BaseDynamicPartitionDataWriter`
- private val formatMapping = Map(
- "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat" -> "orc",
- "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat" ->
"parquet"
- )
- private def getNativeFormat(cmd: DataWritingCommand): Option[String] = {
- cmd match {
- case command: CreateDataSourceTableAsSelectCommand
- if !BackendsApiManager.getSettings.skipNativeCtas(command) =>
- command.table.provider.filter(GlutenFormatFactory.isRegistered)
- case command: InsertIntoHadoopFsRelationCommand
- if !BackendsApiManager.getSettings.skipNativeInsertInto(command) =>
- command.fileFormat match {
- case register: DataSourceRegister
- if GlutenFormatFactory.isRegistered(register.shortName()) =>
- Some(register.shortName())
- case _ => None
- }
- case command: InsertIntoHiveDirCommand =>
- command.storage.outputFormat
- .flatMap(formatMapping.get)
- .filter(GlutenFormatFactory.isRegistered)
- case command: InsertIntoHiveTable =>
- command.table.storage.outputFormat
- .flatMap(formatMapping.get)
- .filter(GlutenFormatFactory.isRegistered)
- case command: CreateHiveTableAsSelectCommand =>
- command.tableDesc.storage.outputFormat
- .flatMap(formatMapping.get)
- .filter(GlutenFormatFactory.isRegistered)
- case _ =>
- None
- }
- }
private[datasources] def injectFakeRowAdaptor(command: SparkPlan, child:
SparkPlan): SparkPlan = {
child match {
@@ -95,35 +52,6 @@ object GlutenWriterColumnarRules {
}
}
- case class NativeWritePostRule(session: SparkSession) extends
Rule[SparkPlan] {
-
- override def apply(p: SparkPlan): SparkPlan = p match {
- case rc @ DataWritingCommandExec(cmd, child) =>
- // The same thread can set these properties in the last query
submission.
- val format =
- if (BackendsApiManager.getSettings.enableNativeWriteFiles()) {
- getNativeFormat(cmd)
- } else {
- None
- }
- val numStaticPartitions: Option[Int] = cmd match {
- case cmd: InsertIntoHadoopFsRelationCommand =>
- Some(cmd.staticPartitions.size)
- case _ =>
- None
- }
- injectSparkLocalProperty(session, format, numStaticPartitions)
- format match {
- case Some(_) =>
- injectFakeRowAdaptor(rc, child)
- case None =>
- rc.withNewChildren(rc.children.map(apply))
- }
-
- case plan: SparkPlan => plan.withNewChildren(plan.children.map(apply))
- }
- }
-
// TODO: This makes FileFormatWriter#write caller-sensitive.
// Remove this workaround once we have a better solution.
def injectSparkLocalProperty(
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypesSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypesSuite.scala
deleted file mode 100644
index 56bab3a745..0000000000
---
a/gluten-substrait/src/test/scala/org/apache/gluten/extension/columnar/rewrite/AlignExpandOutputTypesSuite.scala
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.gluten.extension.columnar.rewrite
-
-import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.types._
-
-import org.scalatest.funsuite.AnyFunSuite
-
-class AlignExpandOutputTypesSuite extends AnyFunSuite {
- test("align null literal type to expand output type") {
- val outputType = DecimalType(20, 6)
- val expression = Literal.create(null, NullType)
- val output = AttributeReference("value", outputType, nullable = true)()
-
- val aligned = ExpandOutputTypeAlignment.alignProjections(
- Seq(Seq(expression)),
- Seq(output),
- Seq.empty)
-
- assert(aligned.head.head == Literal.create(null, outputType))
- }
-
- test("cast decimal expression whose result type differs from output type") {
- val inputType = DecimalType(20, 6)
- val outputType = DecimalType(30, 6)
- val amount = AttributeReference("amount", inputType, nullable = true)()
- val expression = Add(amount, Literal(Decimal(1), inputType))
- val output = AttributeReference("value", outputType, nullable = true)()
-
- val aligned = ExpandOutputTypeAlignment.alignProjections(
- Seq(Seq(expression)),
- Seq(output),
- Seq(amount))
-
- aligned.head.head match {
- case Cast(child, castType, _, _) =>
- assert(child == expression)
- assert(castType == outputType)
- case other =>
- fail(s"Expected decimal expression to be cast to $outputType, got
$other")
- }
- }
-
- test("cast non-decimal expression whose result type differs from output
type") {
- val id = AttributeReference("id", IntegerType, nullable = false)()
- val expression = Add(id, Literal(1))
- val output = AttributeReference("value", LongType, nullable = false)()
-
- val aligned = ExpandOutputTypeAlignment.alignProjections(
- Seq(Seq(expression)),
- Seq(output),
- Seq(id))
-
- aligned.head.head match {
- case Cast(child, castType, _, _) =>
- assert(child == expression)
- assert(castType == LongType)
- case other =>
- fail(s"Expected integer expression to be cast to LongType, got $other")
- }
- }
-}
diff --git
a/gluten-ut/test/src/test/scala/org/apache/gluten/expressions/GlutenCheckOverflowTransformerSuite.scala
b/gluten-ut/test/src/test/scala/org/apache/gluten/expressions/GlutenCheckOverflowTransformerSuite.scala
index d97dea2f38..8218773662 100644
---
a/gluten-ut/test/src/test/scala/org/apache/gluten/expressions/GlutenCheckOverflowTransformerSuite.scala
+++
b/gluten-ut/test/src/test/scala/org/apache/gluten/expressions/GlutenCheckOverflowTransformerSuite.scala
@@ -37,13 +37,7 @@ class GlutenCheckOverflowTransformerSuite extends
GlutenQueryTest with SharedSpa
.set("spark.ui.enabled", "false")
}
- testWithSpecifiedSparkVersion(
- "CheckOverflow transformer casts transformed child type",
- "3.3",
- "3.4",
- "3.5",
- "4.0",
- "4.1") {
+ test("CheckOverflow transformer casts transformed child type") {
assume(BackendTestUtils.isVeloxBackendLoaded())
val targetType = DecimalType(38, 17)
diff --git
a/gluten-ut/test/src/test/scala/org/apache/spark/sql/GlutenImplicitsTest.scala
b/gluten-ut/test/src/test/scala/org/apache/spark/sql/GlutenImplicitsTest.scala
index 07f6e48b3e..29ba396f76 100644
---
a/gluten-ut/test/src/test/scala/org/apache/spark/sql/GlutenImplicitsTest.scala
+++
b/gluten-ut/test/src/test/scala/org/apache/spark/sql/GlutenImplicitsTest.scala
@@ -137,11 +137,10 @@ class GlutenImplicitsTest extends GlutenQueryTest with
SharedSparkSession {
withAQEEnabledAndDisabled {
withTable("tmp") {
val df = spark.sql("create table tmp using parquet as select * from
t1")
- // Spark 3.3 counts one Gluten node here. Since 3.4 the CTAS is
executed as an
- // ExecutedCommandExec, which collectFallbackNodes walks past without
counting anything,
- // so the summary reports neither a Gluten node nor a fallback node.
- val expectedGlutenNodes = if (isSparkVersionGE("3.4")) 0 else 1
- assert(df.fallbackSummary().numGlutenNodes == expectedGlutenNodes,
df.fallbackSummary())
+ // The CTAS is executed as an ExecutedCommandExec, which
collectFallbackNodes walks past
+ // without counting anything, so the summary reports neither a Gluten
node nor a fallback
+ // node.
+ assert(df.fallbackSummary().numGlutenNodes == 0, df.fallbackSummary())
assert(df.fallbackSummary().numFallbackNodes == 0,
df.fallbackSummary())
}
}
diff --git
a/shims/common/src/main/scala/org/apache/gluten/execution/datasource/GlutenFormatWriterInjects.scala
b/shims/common/src/main/scala/org/apache/gluten/execution/datasource/GlutenFormatWriterInjects.scala
index 0e402dd334..cf976ecd33 100644
---
a/shims/common/src/main/scala/org/apache/gluten/execution/datasource/GlutenFormatWriterInjects.scala
+++
b/shims/common/src/main/scala/org/apache/gluten/execution/datasource/GlutenFormatWriterInjects.scala
@@ -17,7 +17,6 @@
package org.apache.gluten.execution.datasource
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.datasources.{BlockStripes, OutputWriter}
import org.apache.spark.sql.types.StructType
@@ -57,7 +56,6 @@ trait GlutenRowSplitter {
object GlutenFormatFactory {
private var instances: Map[String, GlutenFormatWriterInjects] = _
- private var postRuleFactory: SparkSession => Rule[SparkPlan] = _
private var rowSplitterInstance: GlutenRowSplitter = _
def register(items: GlutenFormatWriterInjects*): Unit = {
@@ -72,10 +70,6 @@ object GlutenFormatFactory {
throw new IllegalStateException(s"GlutenFormatWriterInjects for $name is
not initialized"))
}
- def injectPostRuleFactory(factory: SparkSession => Rule[SparkPlan]): Unit = {
- postRuleFactory = factory
- }
-
def register(rowSplitter: GlutenRowSplitter): Unit = {
rowSplitterInstance = rowSplitter
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
