This is an automated email from the ASF dual-hosted git repository.
zzcclp 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 8226101934 [MINOR][CH] Remove residual Spark 3.2 branches from
clickhouse tests (#12532)
8226101934 is described below
commit 82261019346f2a860778d12d2b320bb0dacc63f3
Author: YangJie <[email protected]>
AuthorDate: Thu Jul 16 19:39:42 2026 +0800
[MINOR][CH] Remove residual Spark 3.2 branches from clickhouse tests
(#12532)
[MINOR][CH] Remove residual Spark 3.2 branches from clickhouse tests
(#12532)
Spark 3.2 was dropped by #11351 / #11687 / #11731 / #11887; the
`spark32` protected def in `GlutenClickHouseWholeStageTransformerSuite`
was defined as `sparkVersion.equals("3.2")` and has been dead since.
This PR removes the definition and prunes every reachable
`if (spark32) ...` / `if (!spark32) ...` / `${if (spark32) ... else ...}`
branch to keep only the Spark 3.3+ path.
Test-code changes (all in `backends-clickhouse/src/test/`):
- `GlutenClickHouseWholeStageTransformerSuite.scala`: drop the
`protected def spark32` definition (always false).
- `GlutenClickHouseTPCHBucketSuite.scala`: `hasSortByCol = !spark32`
collapses to `true` on all supported Sparks; version-gated
`if (spark32) ...` branches removed.
- `GlutenClickHouseTPCHParquetBucketSuite.scala`,
`GlutenClickHouseDeltaParquetWriteSuite.scala`,
`GlutenClickHouseMergeTreeWriteSuite.scala`,
`GlutenClickHouseMergeTreeOptimizeSuite.scala`,
`GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala`,
`GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala`,
`GlutenClickHouseMergeTreeWriteOnS3Suite.scala`,
`GlutenClickHouseMergeTreePathBasedWriteSuite.scala`: every reachable
`if (spark32) ... else ...` block, `if (!spark32) ...` guard, and
inline `${if (spark32) "" else "SORTED BY (...)"}` interpolation is
reduced to the Spark 3.3+ path (always emit `SORTED BY`).
- `GlutenClickHouseTPCDSParquetAQESuite.scala`,
`GlutenClickHouseTPCDSParquetColumnarShuffleAQESuite.scala`:
comments narrowed from "On Spark 3.2, ... on Spark 3.3, ..." to
describe only the surviving Spark 3.3+ shape.
- `hive/GlutenClickHouseNativeWriteTableSuite.scala`: drop stale
`// spark 3.2 without orc or parquet suffix` comment.
Main-code change (one file):
- `RowToCHNativeColumnarExec.scala`: drop the `// For spark 3.2.`
comment above `withNewChildInternal`. The override is required by
`TreeNode`'s API on every Spark version currently supported by
Gluten, not a Spark 3.2-only quirk. Mirrors the same cleanup for
`RowToVeloxColumnarExec` included in #12525.
Explicitly kept for a separate follow-up PR:
- `backends-clickhouse/.../ExtendedColumnPruning.scala:66-72` — the
local `getAttributeToExtractValues` re-implementation exists because
Spark 3.2's upstream signature was 2-arg. On 3.3+ it is 3-arg; the
local copy could be replaced with a delegate. That is a real
refactor, not a comment fix.
- `backends-clickhouse/.../CHColumnarWrite.scala:157` — the
`bucketSpec` reflection was needed for Spark 3.2, may be replaceable
with direct access on 3.3+. Also a real refactor.
- `CustomSum.scala:28` — historical provenance of a copied file, not
a version gate; keep as-is.
Verified with `mvn -pl backends-clickhouse -am install -Pspark-3.3,
backends-clickhouse,delta` (SUCCESS) and `scalastyle:check spotless:check`
(SUCCESS).
---
.../sql/execution/RowToCHNativeColumnarExec.scala | 1 -
.../GlutenClickHouseDeltaParquetWriteSuite.scala | 42 ++---
.../GlutenClickHouseTPCHBucketSuite.scala | 81 +++-------
...lutenClickHouseWholeStageTransformerSuite.scala | 1 -
.../GlutenClickHouseNativeWriteTableSuite.scala | 1 -
.../GlutenClickHouseMergeTreeOptimizeSuite.scala | 102 +++----------
...tenClickHouseMergeTreePathBasedWriteSuite.scala | 5 +-
...GlutenClickHouseMergeTreeWriteOnHDFSSuite.scala | 17 +--
...eMergeTreeWriteOnHDFSWithRocksDBMetaSuite.scala | 17 +--
.../GlutenClickHouseMergeTreeWriteOnS3Suite.scala | 19 +--
.../GlutenClickHouseMergeTreeWriteSuite.scala | 169 ++++++++++-----------
.../GlutenClickHouseTPCDSParquetAQESuite.scala | 3 +-
...kHouseTPCDSParquetColumnarShuffleAQESuite.scala | 3 +-
.../GlutenClickHouseTPCHParquetBucketSuite.scala | 66 +++-----
14 files changed, 167 insertions(+), 360 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/RowToCHNativeColumnarExec.scala
b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/RowToCHNativeColumnarExec.scala
index aea368c143..991398f704 100644
---
a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/RowToCHNativeColumnarExec.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/RowToCHNativeColumnarExec.scala
@@ -93,7 +93,6 @@ case class RowToCHNativeColumnarExec(child: SparkPlan)
child.executeBroadcast[T]()
}
- // For spark 3.2.
protected def withNewChildInternal(newChild: SparkPlan):
RowToCHNativeColumnarExec =
copy(child = newChild)
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
index e85a4d52ab..58ec1bde1f 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
@@ -959,11 +959,7 @@ class GlutenClickHouseDeltaParquetWriteSuite extends
ParquetTPCHSuite {
spark.sparkContext.setJobGroup("test3", "test3")
spark.sql("optimize lineitem_delta_parquet_optimize_p2")
val job_ids = spark.sparkContext.statusTracker.getJobIdsForGroup("test3")
- if (spark32) {
- assert(job_ids.length === 7) // WILL trigger actual merge job
- } else {
- assert(job_ids.length === 8) // WILL trigger actual merge job
- }
+ assert(job_ids.length === 8) // WILL trigger actual merge job
spark.sparkContext.clearJobGroup()
@@ -972,11 +968,7 @@ class GlutenClickHouseDeltaParquetWriteSuite extends
ParquetTPCHSuite {
assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p2")) === 23)
spark.sql("VACUUM lineitem_delta_parquet_optimize_p2 RETAIN 0 HOURS")
- if (spark32) {
- assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p2")) === 5)
- } else {
- assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p2")) === 7)
- }
+ assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p2")) === 7)
val ret2 = spark.sql("select count(*) from
lineitem_delta_parquet_optimize_p2").collect()
assert(ret2.apply(0).get(0) === 600572)
@@ -1002,11 +994,7 @@ class GlutenClickHouseDeltaParquetWriteSuite extends
ParquetTPCHSuite {
assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p4")) === 149)
spark.sql("VACUUM lineitem_delta_parquet_optimize_p4 RETAIN 0 HOURS")
- if (spark32) {
- assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p4")) === 23)
- } else {
- assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p4")) === 25)
- }
+ assert(countFiles(new
File(s"$dataHome/lineitem_delta_parquet_optimize_p4")) === 25)
val ret2 = spark.sql("select count(*) from
lineitem_delta_parquet_optimize_p4").collect()
assert(ret2.apply(0).get(0) === 600572)
@@ -1038,12 +1026,8 @@ class GlutenClickHouseDeltaParquetWriteSuite extends
ParquetTPCHSuite {
assert(countFiles(new File(dataPath)) === 77)
clickhouseTable.vacuum(0.0)
- if (spark32) {
- assert(countFiles(new File(dataPath)) === 27)
- } else {
- // There are 25 parquet files + 4 json files after vacuum
- assert(countFiles(new File(dataPath)) === 29)
- }
+ // There are 25 parquet files + 4 json files after vacuum
+ assert(countFiles(new File(dataPath)) === 29)
val ret = spark.sql(s"select count(*) from
clickhouse.`$dataPath`").collect()
assert(ret.apply(0).get(0) === 600572)
@@ -1057,12 +1041,8 @@ class GlutenClickHouseDeltaParquetWriteSuite extends
ParquetTPCHSuite {
clickhouseTable.optimize().executeCompaction()
clickhouseTable.vacuum(0.0)
- if (spark32) {
- assert(countFiles(new File(dataPath)) === 6)
- } else {
- // There are 3 parquet files + 7 json files + 2 check point files
after vacuum
- assert(countFiles(new File(dataPath)) === 12)
- }
+ // There are 3 parquet files + 7 json files + 2 check point files after
vacuum
+ assert(countFiles(new File(dataPath)) === 12)
val ret = spark.sql(s"select count(*) from
clickhouse.`$dataPath`").collect()
assert(ret.apply(0).get(0) === 600572)
@@ -1073,12 +1053,8 @@ class GlutenClickHouseDeltaParquetWriteSuite extends
ParquetTPCHSuite {
clickhouseTable.optimize().executeCompaction()
clickhouseTable.vacuum(0.0)
- if (spark32) {
- assert(countFiles(new File(dataPath)) === 5)
- } else {
- // There are 1 parquet file + 10 json files + 2 check point files after
vacuum
- assert(countFiles(new File(dataPath)) === 13)
- }
+ // There are 1 parquet file + 10 json files + 2 check point files after
vacuum
+ assert(countFiles(new File(dataPath)) === 13)
val ret = spark.sql(s"select count(*) from
clickhouse.`$dataPath`").collect()
assert(ret.apply(0).get(0) === 600572)
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 59c4277520..92e27edf23 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
@@ -31,9 +31,9 @@ class GlutenClickHouseTPCHBucketSuite
with TPCHBucketTableSource
with TPCHMergeTreeResult {
- // On Spark 3.2, the bucket table does not support creating a bucket column
- // with sort columns for DS V2
- lazy val hasSortByCol: Boolean = !spark32
+ // Bucket tables with sort columns are supported for DS V2 on all currently
+ // supported Spark versions.
+ lazy val hasSortByCol: Boolean = true
lazy val tableFormat: String = "clickhouse"
override protected def sparkConf: SparkConf = {
@@ -97,25 +97,13 @@ class GlutenClickHouseTPCHBucketSuite
.asInstanceOf[HashJoinLikeExecTransformer]
.left
.isInstanceOf[InputIteratorTransformer])
- if (spark32) {
- assert(
- plans(9)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .right
- .isInstanceOf[InputIteratorTransformer])
- } else {
- assert(
- plans(9)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .right
- .isInstanceOf[FilterExecTransformerBase])
- }
+ assert(
+ plans(9)
+ .asInstanceOf[HashJoinLikeExecTransformer]
+ .right
+ .isInstanceOf[FilterExecTransformerBase])
- if (spark32) {
-
assert(!plans(11).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- } else {
-
assert(plans(11).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- }
+
assert(plans(11).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
assert(plans(11).metrics("numFiles").value === 1)
assert(plans(11).metrics("numOutputRows").value === 1000)
})
@@ -128,30 +116,18 @@ class GlutenClickHouseTPCHBucketSuite
case scanExec: BasicScanExecTransformer => scanExec
case joinExec: HashJoinLikeExecTransformer => joinExec
}
- if (spark32) {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[InputIteratorTransformer])
- } else {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[ProjectExecTransformer])
- }
+ assert(
+ plans(1)
+ .asInstanceOf[HashJoinLikeExecTransformer]
+ .left
+ .isInstanceOf[ProjectExecTransformer])
assert(
plans(1)
.asInstanceOf[HashJoinLikeExecTransformer]
.right
.isInstanceOf[InputIteratorTransformer])
- if (spark32) {
-
assert(!plans(2).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- } else {
-
assert(plans(2).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- }
+
assert(plans(2).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
assert(plans(2).metrics("numFiles").value === 2)
assert(plans(2).metrics("numOutputRows").value === 3111)
@@ -320,19 +296,11 @@ class GlutenClickHouseTPCHBucketSuite
val plans = collect(df.queryExecution.executedPlan) {
case joinExec: HashJoinLikeExecTransformer => joinExec
}
- if (spark32) {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[InputIteratorTransformer])
- } else {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[FilterExecTransformerBase])
- }
+ assert(
+ plans(1)
+ .asInstanceOf[HashJoinLikeExecTransformer]
+ .left
+ .isInstanceOf[FilterExecTransformerBase])
assert(
plans(1)
.asInstanceOf[HashJoinLikeExecTransformer]
@@ -560,13 +528,8 @@ class GlutenClickHouseTPCHBucketSuite
runSql(SQL6)(
df => {
checkResult(df, Array(Row(600572)))
- if (spark32) {
- // there is a shuffle between two phase hash aggregate.
- checkHashAggregateCount(df, 2)
- } else {
- // the delta will use the delta log meta to response this sql
- checkHashAggregateCount(df, 0)
- }
+ // the delta will use the delta log meta to response this sql
+ checkHashAggregateCount(df, 0)
})
// test sort aggregates
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 86a7f7dabb..4febbaefca 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 spark32: Boolean = sparkVersion.equals("3.2")
protected def spark33: Boolean = sparkVersion.equals("3.3")
protected def spark35: Boolean = sparkVersion.equals("3.5")
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 289a58106c..6f0885951f 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
@@ -122,7 +122,6 @@ class GlutenClickHouseNativeWriteTableSuite
(s"test_insert_into_${format}_supplier", null, sql)
},
(table_name, format) => {
- // spark 3.2 without orc or parquet suffix
val files = recursiveListFiles(new File(s"$dataHome/$table_name"))
.map(_.getName)
.filterNot(s => s.endsWith(s".crc") || s.equals("_SUCCESS"))
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 b684ca01c7..f12c58103a 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
@@ -147,12 +147,8 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
assertResult(22728)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p")))
spark.sql("VACUUM lineitem_mergetree_optimize_p RETAIN 0 HOURS")
- if (spark32) {
- assertResult(22728)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p")))
- } else {
- // For Spark 3.3 + Delta 2.3, vacuum command will create two commit
files in deltalog dir.
- assertResult(22730)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p")))
- }
+ // For Spark 3.3 + Delta 2.3, vacuum command will create two commit files
in deltalog dir.
+ assertResult(22730)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p")))
val ret2 = spark.sql("select count(*) from
lineitem_mergetree_optimize_p").collect()
assertResult(600572)(ret2.apply(0).get(0))
@@ -174,11 +170,7 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
spark.sparkContext.setJobGroup("test2", "test2")
with_ut_conf(spark.sql("optimize lineitem_mergetree_optimize_p2"))
val job_ids = spark.sparkContext.statusTracker.getJobIdsForGroup("test2")
- if (spark32) {
- assertResult(7)(job_ids.length) // WILL trigger actual merge job
- } else {
- assertResult(8)(job_ids.length) // WILL trigger actual merge job
- }
+ assertResult(8)(job_ids.length) // WILL trigger actual merge job
spark.sparkContext.clearJobGroup()
@@ -187,18 +179,10 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
assertResult(343)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
spark.sql("VACUUM lineitem_mergetree_optimize_p2 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(239)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
- } else {
- assertResult(240)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
- }
+ assertResult(240)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
spark.sql("VACUUM lineitem_mergetree_optimize_p2 RETAIN 0 HOURS")
// the second VACUUM will remove some empty folders
- if (spark32) {
- assertResult(220)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
- } else {
- assertResult(229)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
- }
+ assertResult(229)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p2")))
val ret2 = spark.sql("select count(*) from
lineitem_mergetree_optimize_p2").collect()
assertResult(600572)(ret2.apply(0).get(0))
@@ -224,17 +208,9 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
assertResult(458)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
spark.sql("VACUUM lineitem_mergetree_optimize_p3 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(302)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
- } else {
- assertResult(306)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
- }
+ assertResult(306)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
spark.sql("VACUUM lineitem_mergetree_optimize_p3 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(275)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
- } else {
- assertResult(288)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
- }
+ assertResult(288)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p3")))
val ret2 = spark.sql("select count(*) from
lineitem_mergetree_optimize_p3").collect()
assertResult(600572)(ret2.apply(0).get(0))
@@ -261,17 +237,9 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
assertResult(458)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
spark.sql("VACUUM lineitem_mergetree_optimize_p4 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(302)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
- } else {
- assertResult(306)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
- }
+ assertResult(306)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
spark.sql("VACUUM lineitem_mergetree_optimize_p4 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(275)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
- } else {
- assertResult(288)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
- }
+ assertResult(288)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p4")))
val ret2 = spark.sql("select count(*) from
lineitem_mergetree_optimize_p4").collect()
assertResult(600572)(ret2.apply(0).get(0))
@@ -297,13 +265,9 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
spark.sql("VACUUM lineitem_mergetree_optimize_p5 RETAIN 0 HOURS")
spark.sql("VACUUM lineitem_mergetree_optimize_p5 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(99)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
- } else {
- // For Spark 3.3 + Delta 2.3, vacuum command will create two commit
files in deltalog dir.
- // this case will create a checkpoint
- assertResult(106)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
- }
+ // For Spark 3.3 + Delta 2.3, vacuum command will create two commit
files in deltalog dir.
+ // this case will create a checkpoint
+ assertResult(106)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
val ret = spark.sql("select count(*) from
lineitem_mergetree_optimize_p5").collect()
assertResult(600572)(ret.apply(0).get(0))
@@ -321,12 +285,8 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
spark.sql("VACUUM lineitem_mergetree_optimize_p5 RETAIN 0 HOURS")
spark.sql("VACUUM lineitem_mergetree_optimize_p5 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(93)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
- } else {
- // For Spark 3.3 + Delta 2.3, vacuum command will create two commit
files in deltalog dir.
- assertResult(106)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
- }
+ // For Spark 3.3 + Delta 2.3, vacuum command will create two commit
files in deltalog dir.
+ assertResult(106)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
val ret = spark.sql("select count(*) from
lineitem_mergetree_optimize_p5").collect()
assertResult(600572)(ret.apply(0).get(0))
@@ -337,12 +297,8 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
spark.sql("VACUUM lineitem_mergetree_optimize_p5 RETAIN 0 HOURS")
spark.sql("VACUUM lineitem_mergetree_optimize_p5 RETAIN 0 HOURS")
- if (spark32) {
- assertResult(77)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
- } else {
- // For Spark 3.3 + Delta 2.3, vacuum command will create two commit
files in deltalog dir.
- assertResult(94)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
- }
+ // For Spark 3.3 + Delta 2.3, vacuum command will create two commit files
in deltalog dir.
+ assertResult(94)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p5")))
val ret = spark.sql("select count(*) from
lineitem_mergetree_optimize_p5").collect()
assertResult(600572)(ret.apply(0).get(0))
@@ -359,7 +315,7 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
|PARTITIONED BY (l_returnflag)
|LOCATION '$dataHome/lineitem_mergetree_optimize_p6'
| CLUSTERED BY (l_partkey)
- | ${if (spark32) "" else "SORTED BY (l_partkey)"} INTO 2
BUCKETS
+ | SORTED BY (l_partkey) INTO 2 BUCKETS
| as select * from lineitem
|""".stripMargin)
@@ -368,12 +324,10 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
val ret = spark.sql("select count(*) from
lineitem_mergetree_optimize_p6").collect()
assertResult(600572)(ret.apply(0).get(0))
- assertResult(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p6")))(
- if (spark32) 491 else 489)
+ assertResult(489)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p6")))
spark.sql("VACUUM lineitem_mergetree_optimize_p6 RETAIN 0 HOURS")
spark.sql("VACUUM lineitem_mergetree_optimize_p6 RETAIN 0 HOURS")
- assertResult(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p6")))(
- if (spark32) 315 else 333)
+ assertResult(333)(countFiles(new
File(s"$dataHome/lineitem_mergetree_optimize_p6")))
val ret2 = spark.sql("select count(*) from
lineitem_mergetree_optimize_p6").collect()
assertResult(600572)(ret2.apply(0).get(0))
@@ -442,11 +396,7 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
clickhouseTable.vacuum(0.0)
clickhouseTable.vacuum(0.0)
- if (spark32) {
- assertResult(99)(countFiles(new File(dataPath)))
- } else {
- assertResult(106)(countFiles(new File(dataPath)))
- }
+ assertResult(106)(countFiles(new File(dataPath)))
val ret = spark.sql(s"select count(*) from
clickhouse.`$dataPath`").collect()
assertResult(600572)(ret.apply(0).get(0))
@@ -465,11 +415,7 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
clickhouseTable.vacuum(0.0)
clickhouseTable.vacuum(0.0)
- if (spark32) {
- assertResult(93)(countFiles(new File(dataPath)))
- } else {
- assertResult(106)(countFiles(new File(dataPath)))
- }
+ assertResult(106)(countFiles(new File(dataPath)))
val ret = spark.sql(s"select count(*) from
clickhouse.`$dataPath`").collect()
assertResult(600572)(ret.apply(0).get(0))
@@ -481,11 +427,7 @@ class GlutenClickHouseMergeTreeOptimizeSuite extends
CreateMergeTreeSuite {
clickhouseTable.vacuum(0.0)
clickhouseTable.vacuum(0.0)
- if (spark32) {
- assertResult(77)(countFiles(new File(dataPath)))
- } else {
- assertResult(94)(countFiles(new File(dataPath)))
- }
+ assertResult(94)(countFiles(new File(dataPath)))
val ret = spark.sql(s"select count(*) from
clickhouse.`$dataPath`").collect()
assertResult(600572)(ret.apply(0).get(0))
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 b8907d1384..2db25acbb8 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
@@ -1089,10 +1089,7 @@ class GlutenClickHouseMergeTreePathBasedWriteSuite
extends CreateMergeTreeSuite
.count()
val result = df.collect()
assertResult(600572)(result(0).getLong(0))
- // Spark 3.2 + Delta 2.0 does not support this feature
- if (!spark32) {
- assert(df.queryExecution.executedPlan.isInstanceOf[LocalTableScanExec])
- }
+ assert(df.queryExecution.executedPlan.isInstanceOf[LocalTableScanExec])
}
test(
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 0e1bfbe72c..b143d29428 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
@@ -416,7 +416,7 @@ class GlutenClickHouseMergeTreeWriteOnHDFSSuite extends
CreateMergeTreeSuite {
|USING clickhouse
|PARTITIONED BY (l_returnflag)
|CLUSTERED BY (l_orderkey)
- |${if (spark32) "" else "SORTED BY (l_partkey)"} INTO 4
BUCKETS
+ |SORTED BY (l_partkey) INTO 4 BUCKETS
|LOCATION '$remotePath/lineitem_mergetree_bucket_hdfs'
|TBLPROPERTIES (storage_policy='__hdfs_main')
|""".stripMargin)
@@ -439,17 +439,10 @@ class GlutenClickHouseMergeTreeWriteOnHDFSSuite extends
CreateMergeTreeSuite {
val fileIndex =
mergetreeScan.relation.location.asInstanceOf[TahoeFileIndex]
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).clickhouseTableConfigs.nonEmpty)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).bucketOption.isDefined)
- if (spark32) {
- assert(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey === StorageMeta.DEFAULT_ORDER_BY_KEY)
- } else {
- assertResult("l_partkey")(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey)
- }
+ assertResult("l_partkey")(
+ ClickHouseTableV2
+ .getTable(fileIndex.deltaLog)
+ .orderByKey)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).primaryKey.isEmpty)
assertResult(1)(ClickHouseTableV2.getTable(fileIndex.deltaLog).partitionColumns.size)
assertResult("l_returnflag")(
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 a818470147..76e55443f8 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
@@ -365,7 +365,7 @@ class
GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite extends CreateMer
|USING clickhouse
|PARTITIONED BY (l_returnflag)
|CLUSTERED BY (l_orderkey)
- |${if (spark32) "" else "SORTED BY (l_partkey)"} INTO 4
BUCKETS
+ |SORTED BY (l_partkey) INTO 4 BUCKETS
|LOCATION '$remotePath/lineitem_mergetree_bucket_hdfs'
|TBLPROPERTIES (storage_policy='__hdfs_main_rocksdb')
|""".stripMargin)
@@ -388,17 +388,10 @@ class
GlutenClickHouseMergeTreeWriteOnHDFSWithRocksDBMetaSuite extends CreateMer
val fileIndex =
mergetreeScan.relation.location.asInstanceOf[TahoeFileIndex]
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).clickhouseTableConfigs.nonEmpty)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).bucketOption.isDefined)
- if (spark32) {
- assert(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey === StorageMeta.DEFAULT_ORDER_BY_KEY)
- } else {
- assertResult("l_partkey")(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey)
- }
+ assertResult("l_partkey")(
+ ClickHouseTableV2
+ .getTable(fileIndex.deltaLog)
+ .orderByKey)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).primaryKey.isEmpty)
assertResult(1)(ClickHouseTableV2.getTable(fileIndex.deltaLog).partitionColumns.size)
assertResult("l_returnflag")(
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 425c640aae..0659b9e6d7 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
@@ -406,7 +406,7 @@ class GlutenClickHouseMergeTreeWriteOnS3Suite extends
CreateMergeTreeSuite {
|USING clickhouse
|PARTITIONED BY (l_returnflag)
|CLUSTERED BY (l_orderkey)
- |${if (spark32) "" else "SORTED BY (l_partkey)"} INTO 4
BUCKETS
+ |SORTED BY (l_partkey) INTO 4 BUCKETS
|LOCATION 's3a://$BUCKET_NAME/lineitem_mergetree_bucket_s3'
|TBLPROPERTIES (storage_policy='__s3_main')
|""".stripMargin)
@@ -429,17 +429,10 @@ class GlutenClickHouseMergeTreeWriteOnS3Suite extends
CreateMergeTreeSuite {
val fileIndex =
mergetreeScan.relation.location.asInstanceOf[TahoeFileIndex]
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).clickhouseTableConfigs.nonEmpty)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).bucketOption.isDefined)
- if (spark32) {
- assert(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey === StorageMeta.DEFAULT_ORDER_BY_KEY)
- } else {
- assertResult("l_partkey")(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey)
- }
+ assertResult("l_partkey")(
+ ClickHouseTableV2
+ .getTable(fileIndex.deltaLog)
+ .orderByKey)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).primaryKey.isEmpty)
assertResult(1)(ClickHouseTableV2.getTable(fileIndex.deltaLog).partitionColumns.size)
assertResult("l_returnflag")(
@@ -624,7 +617,7 @@ class GlutenClickHouseMergeTreeWriteOnS3Suite extends
CreateMergeTreeSuite {
|USING clickhouse
|PARTITIONED BY (l_returnflag)
|CLUSTERED BY (l_orderkey)
- |${if (spark32) "" else "SORTED BY (l_partkey)"} INTO 4
BUCKETS
+ |SORTED BY (l_partkey) INTO 4 BUCKETS
|LOCATION 's3a://$BUCKET_NAME/lineitem_mergetree_bucket_s3'
|TBLPROPERTIES (storage_policy='__s3_main')
|""".stripMargin)
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 b94db0ca9e..53b725cfbe 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
@@ -748,7 +748,7 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
|USING clickhouse
|PARTITIONED BY (l_returnflag)
|CLUSTERED BY (l_partkey)
- |${if (spark32) "" else "SORTED BY (l_orderkey)"} INTO 4
BUCKETS
+ |SORTED BY (l_orderkey) INTO 4 BUCKETS
|LOCATION '$dataHome/lineitem_mergetree_bucket'
|""".stripMargin)
@@ -770,17 +770,10 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
val fileIndex =
mergetreeScan.relation.location.asInstanceOf[TahoeFileIndex]
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).clickhouseTableConfigs.nonEmpty)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).bucketOption.isDefined)
- if (spark32) {
- assert(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey === StorageMeta.DEFAULT_ORDER_BY_KEY)
- } else {
- assertResult("l_orderkey")(
- ClickHouseTableV2
- .getTable(fileIndex.deltaLog)
- .orderByKey)
- }
+ assertResult("l_orderkey")(
+ ClickHouseTableV2
+ .getTable(fileIndex.deltaLog)
+ .orderByKey)
assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).primaryKey.isEmpty)
assertResult(1)(ClickHouseTableV2.getTable(fileIndex.deltaLog).partitionColumns.size)
assertResult("l_returnflag")(
@@ -1633,7 +1626,7 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
|)
|USING clickhouse
|CLUSTERED by (l_orderkey)
- |${if (spark32) "" else "SORTED BY (l_receiptdate)"} INTO 2
BUCKETS
+ |SORTED BY (l_receiptdate) INTO 2 BUCKETS
|LOCATION
'$dataHome/lineitem_mergetree_pk_pruning_by_driver_bucket'
|""".stripMargin)
@@ -1650,7 +1643,7 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
| o_comment string)
|USING clickhouse
|CLUSTERED by (o_orderkey)
- |${if (spark32) "" else "SORTED BY (o_orderdate)"} INTO 2
BUCKETS
+ |SORTED BY (o_orderdate) INTO 2 BUCKETS
|LOCATION
'$dataHome/orders_mergetree_pk_pruning_by_driver_bucket'
|""".stripMargin)
@@ -1766,10 +1759,7 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
assertResult(1)(result.length)
assertResult("600572")(result(0).getLong(0).toString)
- // Spark 3.2 + Delta 2.0 does not support this feature
- if (!spark32) {
-
assert(df.queryExecution.executedPlan.isInstanceOf[LocalTableScanExec])
- }
+
assert(df.queryExecution.executedPlan.isInstanceOf[LocalTableScanExec])
})
}
}
@@ -1897,82 +1887,79 @@ class GlutenClickHouseMergeTreeWriteSuite extends
CreateMergeTreeSuite {
test(
"GLUTEN-7812: Fix the query failed for the mergetree format " +
"when the 'spark.databricks.delta.stats.skipping' is off") {
- // Spark 3.2 + Delta 2.0 doesn't not support this feature
- if (!spark32) {
- withSQLConf(("spark.databricks.delta.stats.skipping", "false")) {
- spark.sql(s"""
- |DROP TABLE IF EXISTS lineitem_mergetree_stats_skipping;
- |""".stripMargin)
+ withSQLConf(("spark.databricks.delta.stats.skipping", "false")) {
+ spark.sql(s"""
+ |DROP TABLE IF EXISTS lineitem_mergetree_stats_skipping;
+ |""".stripMargin)
- spark.sql(s"""
- |CREATE TABLE IF NOT EXISTS
lineitem_mergetree_stats_skipping
- |(
- | l_orderkey bigint,
- | l_partkey bigint,
- | l_suppkey bigint,
- | l_linenumber bigint,
- | l_quantity double,
- | l_extendedprice double,
- | l_discount double,
- | l_tax double,
- | l_returnflag string,
- | l_linestatus string,
- | l_shipdate date,
- | l_commitdate date,
- | l_receiptdate date,
- | l_shipinstruct string,
- | l_shipmode string,
- | l_comment string
- |)
- |USING clickhouse
- |PARTITIONED BY (l_returnflag)
- |TBLPROPERTIES (orderByKey='l_orderkey',
- | primaryKey='l_orderkey')
- |LOCATION '$dataHome/lineitem_mergetree_stats_skipping'
- |""".stripMargin)
+ spark.sql(s"""
+ |CREATE TABLE IF NOT EXISTS
lineitem_mergetree_stats_skipping
+ |(
+ | l_orderkey bigint,
+ | l_partkey bigint,
+ | l_suppkey bigint,
+ | l_linenumber bigint,
+ | l_quantity double,
+ | l_extendedprice double,
+ | l_discount double,
+ | l_tax double,
+ | l_returnflag string,
+ | l_linestatus string,
+ | l_shipdate date,
+ | l_commitdate date,
+ | l_receiptdate date,
+ | l_shipinstruct string,
+ | l_shipmode string,
+ | l_comment string
+ |)
+ |USING clickhouse
+ |PARTITIONED BY (l_returnflag)
+ |TBLPROPERTIES (orderByKey='l_orderkey',
+ | primaryKey='l_orderkey')
+ |LOCATION '$dataHome/lineitem_mergetree_stats_skipping'
+ |""".stripMargin)
- // dynamic partitions
- spark.sql(s"""
- | insert into table lineitem_mergetree_stats_skipping
- | select * from lineitem
- |""".stripMargin)
+ // dynamic partitions
+ spark.sql(s"""
+ | insert into table lineitem_mergetree_stats_skipping
+ | select * from lineitem
+ |""".stripMargin)
- val sqlStr =
- s"""
- |SELECT
- | o_orderpriority,
- | count(*) AS order_count
- |FROM
- | orders
- |WHERE
- | o_orderdate >= date'1993-07-01'
- | AND o_orderdate < date'1993-07-01' + interval 3 month
- | AND EXISTS (
- | SELECT
- | *
- | FROM
- | lineitem
- | WHERE
- | l_orderkey = o_orderkey
- | AND l_commitdate < l_receiptdate)
- |GROUP BY
- | o_orderpriority
- |ORDER BY
- | o_orderpriority;
- |
- |""".stripMargin
- runSql(sqlStr)(
- df => {
- val result = df.collect()
- assertResult(5)(result.length)
- assertResult("1-URGENT")(result(0).getString(0))
- assertResult(999)(result(0).getLong(1))
- assertResult("2-HIGH")(result(1).getString(0))
- assertResult(997)(result(1).getLong(1))
- assertResult("5-LOW")(result(4).getString(0))
- assertResult(1077)(result(4).getLong(1))
- })
- }
+ val sqlStr =
+ s"""
+ |SELECT
+ | o_orderpriority,
+ | count(*) AS order_count
+ |FROM
+ | orders
+ |WHERE
+ | o_orderdate >= date'1993-07-01'
+ | AND o_orderdate < date'1993-07-01' + interval 3 month
+ | AND EXISTS (
+ | SELECT
+ | *
+ | FROM
+ | lineitem
+ | WHERE
+ | l_orderkey = o_orderkey
+ | AND l_commitdate < l_receiptdate)
+ |GROUP BY
+ | o_orderpriority
+ |ORDER BY
+ | o_orderpriority;
+ |
+ |""".stripMargin
+ runSql(sqlStr)(
+ df => {
+ val result = df.collect()
+ assertResult(5)(result.length)
+ assertResult("1-URGENT")(result(0).getString(0))
+ assertResult(999)(result(0).getLong(1))
+ assertResult("2-HIGH")(result(1).getString(0))
+ assertResult(997)(result(1).getLong(1))
+ assertResult("5-LOW")(result(4).getString(0))
+ assertResult(1077)(result(4).getLong(1))
+ })
}
}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetAQESuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetAQESuite.scala
index 52a72c4494..fc5824ed11 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetAQESuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetAQESuite.scala
@@ -123,8 +123,7 @@ class GlutenClickHouseTPCDSParquetAQESuite
case r: ReusedSubqueryExec => true
case _ => false
}
- // On Spark 3.2, there are 15 AdaptiveSparkPlanExec,
- // and on Spark 3.3, there are 5 AdaptiveSparkPlanExec and 10
ReusedSubqueryExec
+ // 5 AdaptiveSparkPlanExec and 10 ReusedSubqueryExec on Spark 3.3+.
assertResult(15)(subqueryAdaptiveSparkPlan.count(_ == true))
}
}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetColumnarShuffleAQESuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetColumnarShuffleAQESuite.scala
index ac4640640f..92338a369f 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetColumnarShuffleAQESuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/tpcds/GlutenClickHouseTPCDSParquetColumnarShuffleAQESuite.scala
@@ -119,8 +119,7 @@ class GlutenClickHouseTPCDSParquetColumnarShuffleAQESuite
case r: ReusedSubqueryExec => true
case _ => false
}
- // On Spark 3.2, there are 15 AdaptiveSparkPlanExec,
- // and on Spark 3.3, there are 5 AdaptiveSparkPlanExec and 10
ReusedSubqueryExec
+ // 5 AdaptiveSparkPlanExec and 10 ReusedSubqueryExec on Spark 3.3+.
assertResult(15)(subqueryAdaptiveSparkPlan.count(_ == true))
}
}
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 bd168fa595..31b31cfc2c 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
@@ -131,25 +131,13 @@ class GlutenClickHouseTPCHParquetBucketSuite
.asInstanceOf[HashJoinLikeExecTransformer]
.left
.isInstanceOf[InputIteratorTransformer])
- if (spark32) {
- assert(
- plans(9)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .right
- .isInstanceOf[InputIteratorTransformer])
- } else {
- assert(
- plans(9)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .right
- .isInstanceOf[FilterExecTransformerBase])
- }
+ assert(
+ plans(9)
+ .asInstanceOf[HashJoinLikeExecTransformer]
+ .right
+ .isInstanceOf[FilterExecTransformerBase])
- if (spark32) {
-
assert(!plans(11).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- } else {
-
assert(plans(11).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- }
+
assert(plans(11).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
assert(plans(11).metrics("numFiles").value === 1)
assert(plans(11).metrics("numOutputRows").value === 1000)
}
@@ -162,19 +150,11 @@ class GlutenClickHouseTPCHParquetBucketSuite
case scanExec: BasicScanExecTransformer => scanExec
case joinExec: HashJoinLikeExecTransformer => joinExec
}
- if (spark32) {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[InputIteratorTransformer])
- } else {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[ProjectExecTransformer])
- }
+ assert(
+ plans(1)
+ .asInstanceOf[HashJoinLikeExecTransformer]
+ .left
+ .isInstanceOf[ProjectExecTransformer])
assert(
plans(1)
@@ -182,11 +162,7 @@ class GlutenClickHouseTPCHParquetBucketSuite
.right
.isInstanceOf[InputIteratorTransformer])
- if (spark32) {
-
assert(!plans(2).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- } else {
-
assert(plans(2).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
- }
+
assert(plans(2).asInstanceOf[FileSourceScanExecTransformer].bucketedScan)
assert(plans(2).metrics("numFiles").value === 4)
assert(plans(2).metrics("numOutputRows").value === 15000)
@@ -355,19 +331,11 @@ class GlutenClickHouseTPCHParquetBucketSuite
val plans = collect(df.queryExecution.executedPlan) {
case joinExec: HashJoinLikeExecTransformer => joinExec
}
- if (spark32) {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[InputIteratorTransformer])
- } else {
- assert(
- plans(1)
- .asInstanceOf[HashJoinLikeExecTransformer]
- .left
- .isInstanceOf[FilterExecTransformerBase])
- }
+ assert(
+ plans(1)
+ .asInstanceOf[HashJoinLikeExecTransformer]
+ .left
+ .isInstanceOf[FilterExecTransformerBase])
assert(
plans(1)
.asInstanceOf[HashJoinLikeExecTransformer]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
