This is an automated email from the ASF dual-hosted git repository.
zhztheplayer 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 91ae3b22fc [VL] Eager aggregation: Support partial-merge mode (#13284)
91ae3b22fc is described below
commit 91ae3b22fcb1dbff6937ae517fc0d816c23f4501
Author: Hongze Zhang <[email protected]>
AuthorDate: Wed Oct 14 16:34:39 2026 +0100
[VL] Eager aggregation: Support partial-merge mode (#13284)
---
.../execution/StarSchemaJoinAggregateSuite.scala | 11 +-
docs/Configuration.md | 2 +-
.../org/apache/gluten/config/ConfigBuilder.scala | 15 +
.../org/apache/gluten/config/GlutenConfig.scala | 20 +-
.../joinagg/PushAggregateThroughJoin.scala | 11 +-
.../gluten/config/SparkConfigUtilSuite.scala | 32 ++
.../execution/PushAggregateThroughJoinSuite.scala | 348 ++++++++-------------
7 files changed, 200 insertions(+), 239 deletions(-)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/StarSchemaJoinAggregateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/StarSchemaJoinAggregateSuite.scala
index 9e848847ae..373ceedef3 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/StarSchemaJoinAggregateSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/StarSchemaJoinAggregateSuite.scala
@@ -73,7 +73,14 @@ class StarSchemaJoinAggregateSingleDepthSuite extends
StarSchemaJoinAggregateSui
class StarSchemaJoinAggregatePartialMergeEnabledSuite extends
StarSchemaJoinAggregateSuite {
override protected def sparkConf: SparkConf = {
super.sparkConf
- .set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_ENABLED.key,
"true")
+ .set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_MODE.key,
"all")
+ }
+}
+
+class StarSchemaJoinAggregateTopPartialMergeSuite extends
StarSchemaJoinAggregateSuite {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_MODE.key,
"top")
}
}
@@ -102,7 +109,7 @@ class StarSchemaJoinAggregateSuite extends
VeloxTPCHTableSupport with AdaptiveSp
.set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true")
.set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_ENABLED.key, "true")
.set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_MAX_DEPTH.key,
s"${Int.MaxValue}")
- .set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_ENABLED.key,
"false")
+ .set(GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_MODE.key,
"none")
.set("spark.sql.adaptive.enabled", "false")
}
diff --git a/docs/Configuration.md b/docs/Configuration.md
index 83e3432343..117d5782b0 100644
--- a/docs/Configuration.md
+++ b/docs/Configuration.md
@@ -138,7 +138,7 @@ nav_order: 15
| spark.gluten.sql.orc.charType.scan.fallback.enabled | 🔄
Dynamic | true | Force fallback for orc char type scan.
[...]
| spark.gluten.sql.pushAggregateThroughJoin.enabled | 🔄
Dynamic | false | Enables the push-aggregate-through-join
optimization in Gluten. When enabled, aggregate operators may be pushed below
joins during logical optimization and corresponding physical plans may be
rewritten to execute the aggregation earlier.
[...]
| spark.gluten.sql.pushAggregateThroughJoin.maxDepth | 🔄
Dynamic | 2147483647 | Maximum join traversal depth when applying the
push-aggregate-through-join optimization. A value of 1 allows pushing an
aggregate through a single join; larger values allow the rule to traverse and
push through multiple consecutive joins.
[...]
-| spark.gluten.sql.pushAggregateThroughJoin.partialMerge.enabled | 🔄
Dynamic | false | Enables a PartialMerge aggregate above each
aggregate pushed through a join.
[...]
+| spark.gluten.sql.pushAggregateThroughJoin.partialMerge.mode | 🔄
Dynamic | NONE | Controls where PartialMerge aggregates are
retained during aggregate pushdown: none retains none, top retains one at the
original partial aggregate position, and all retains one at every successful
join push.
[...]
| spark.gluten.sql.removeNativeWriteFilesSortAndProject | 🔄
Dynamic | true | When true, Gluten will remove the vanilla
Spark V1Writes added sort and project for velox backend.
[...]
| spark.gluten.sql.rewrite.dateTimestampComparison | 🔄
Dynamic | true | Rewrite the comparision between date and
timestamp to timestamp comparison.For example `from_unixtime(ts) > date` will
be rewritten to `ts > to_unixtime(date)`
[...]
| spark.gluten.sql.scan.detailedMetrics.enabled | 🔄
Dynamic | true | When true (default), Velox backend scan
operators register all detailed SQL metrics. When false, only essential scan
metrics are registered to reduce driver memory usage. Also enabled
automatically when spark.gluten.sql.debug is true. Does not affect the
ClickHouse backend.
[...]
diff --git
a/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala
b/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala
index 957f1dcab6..c44a9ae5d1 100644
--- a/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala
+++ b/gluten-core/src/main/scala/org/apache/gluten/config/ConfigBuilder.scala
@@ -18,6 +18,7 @@ package org.apache.gluten.config
import org.apache.spark.network.util.{ByteUnit, JavaUtils}
+import java.util.Locale
import java.util.concurrent.TimeUnit
object BackendType extends Enumeration {
@@ -97,6 +98,10 @@ private[gluten] case class ConfigBuilder(key: String) {
new TypedConfigBuilder(this, identity)
}
+ def enumConf(e: Enumeration): TypedConfigBuilder[e.Value] = {
+ new TypedConfigBuilder(this, toEnum(_, e, key))
+ }
+
def timeConf(unit: TimeUnit): TypedConfigBuilder[Long] = {
new TypedConfigBuilder(this, timeFromString(_, unit), timeToString(_,
unit))
}
@@ -122,6 +127,16 @@ private[gluten] case class ConfigBuilder(key: String) {
}
private object ConfigHelpers {
+ def toEnum[E <: Enumeration](s: String, enumClass: E, key: String):
enumClass.Value = {
+ try {
+ enumClass.withName(s.trim.toUpperCase(Locale.ROOT))
+ } catch {
+ case _: NoSuchElementException =>
+ throw new IllegalArgumentException(
+ s"$key should be one of ${enumClass.values.mkString(", ")}, but was
$s")
+ }
+ }
+
def toNumber[T](s: String, converter: String => T, key: String, configType:
String): T = {
try {
converter(s.trim)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
index 753bc865ab..6699c6eb22 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
@@ -30,6 +30,10 @@ import java.util.Locale
import scala.collection.JavaConverters._
import scala.collection.mutable
+object PartialMergeMode extends Enumeration {
+ val NONE, TOP, ALL = Value
+}
+
trait ShuffleWriterType {
val name: String
val requiresResizingShuffleInput: Boolean
@@ -158,8 +162,8 @@ class GlutenConfig(conf: SQLConf) extends
GlutenCoreConfig(conf) {
def pushAggregateThroughJoinMaxDepth: Int =
getConf(PUSH_AGGREGATE_THROUGH_JOIN_MAX_DEPTH)
- def pushAggregateThroughJoinPartialMergeEnabled: Boolean =
- getConf(PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_ENABLED)
+ def pushAggregateThroughJoinPartialMergeMode: PartialMergeMode.Value =
+ getConf(PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_MODE)
def forceOrcCharTypeScanFallbackEnabled: Boolean =
getConf(VELOX_FORCE_ORC_CHAR_TYPE_SCAN_FALLBACK)
@@ -787,12 +791,14 @@ object GlutenConfig extends ConfigRegistry {
.checkValue(_ >= 1, "must be greater than or equal to 1.")
.createWithDefault(Int.MaxValue)
- val PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_ENABLED =
- buildConf("spark.gluten.sql.pushAggregateThroughJoin.partialMerge.enabled")
+ val PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_MODE =
+ buildConf("spark.gluten.sql.pushAggregateThroughJoin.partialMerge.mode")
.doc(
- "Enables a PartialMerge aggregate above each aggregate pushed through
a join.")
- .booleanConf
- .createWithDefault(false)
+ "Controls where PartialMerge aggregates are retained during aggregate
pushdown: " +
+ "none retains none, top retains one at the original partial
aggregate position, " +
+ "and all retains one at every successful join push.")
+ .enumConf(PartialMergeMode)
+ .createWithDefault(PartialMergeMode.NONE)
val GLUTEN_SOFT_AFFINITY_ENABLED =
buildConf("spark.gluten.soft-affinity.enabled")
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/PushAggregateThroughJoin.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/PushAggregateThroughJoin.scala
index 14d652fddc..4f318f10c8 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/PushAggregateThroughJoin.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/PushAggregateThroughJoin.scala
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.extension.joinagg
-import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.config.{GlutenConfig, PartialMergeMode}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, AttributeSet, EqualTo, Expression, NamedExpression,
PredicateHelper}
@@ -113,9 +113,6 @@ case class PushAggregateThroughJoin(spark: SparkSession)
private def maxDepth: Int = GlutenConfig.get.pushAggregateThroughJoinMaxDepth
- private def partialMergeEnabled: Boolean =
- GlutenConfig.get.pushAggregateThroughJoinPartialMergeEnabled
-
private def splitAggregate(agg: Aggregate): Option[(Aggregate, Aggregate)] =
{
// Split is intentionally child-agnostic. It only rewrites:
//
@@ -177,6 +174,7 @@ case class PushAggregateThroughJoin(spark: SparkSession)
private def pushPartialWrapperAggregate(lowerPartialAgg: Aggregate):
LogicalPlan = {
// Push one join edge per iteration. This keeps the rewrite local and lets
`maxDepth` bound
// how far a pushed aggregate is allowed to travel through a multi-join
subtree.
+ val partialMergeMode =
GlutenConfig.get.pushAggregateThroughJoinPartialMergeMode
var current: LogicalPlan = lowerPartialAgg
var changed = true
var pushCount = 0
@@ -184,7 +182,10 @@ case class PushAggregateThroughJoin(spark: SparkSession)
changed = false
current = current.transformUp {
case partialAgg: Aggregate if
isPurePartialWrapperAggregate(partialAgg) =>
- val pushed = if (partialMergeEnabled) {
+ // The first successful push leaves a merge at the original partial
aggregate position.
+ val retainPartialMerge = partialMergeMode == PartialMergeMode.ALL ||
+ (partialMergeMode == PartialMergeMode.TOP && pushCount == 0)
+ val pushed = if (retainPartialMerge) {
val (partialMergeAgg, freshLowerPartialAgg) =
splitPartialAggregateForMerge(partialAgg)
pushOnce(freshLowerPartialAgg).map {
pushedPlan => partialMergeAgg.copy(child = pushedPlan)
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/config/SparkConfigUtilSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/config/SparkConfigUtilSuite.scala
index 2bcfcb9e67..e39a98ccd6 100644
---
a/gluten-substrait/src/test/scala/org/apache/gluten/config/SparkConfigUtilSuite.scala
+++
b/gluten-substrait/src/test/scala/org/apache/gluten/config/SparkConfigUtilSuite.scala
@@ -24,6 +24,18 @@ import org.scalatest.funsuite.AnyFunSuiteLike
class SparkConfigUtilSuite extends AnyFunSuiteLike {
+ private object TestMode extends Enumeration {
+ val FIRST, LAST = Value
+ }
+
+ private object TestConfigs extends ConfigRegistry {
+ val mode = buildConf("spark.gluten.test.enum.mode")
+ .enumConf(TestMode)
+ .createWithDefault(TestMode.FIRST)
+
+ override def get: GlutenCoreConfig = GlutenCoreConfig.get
+ }
+
test("SparkConfigUtil.get and set methods") {
val conf = new SparkConf()
conf.set(GlutenConfig.SHUFFLE_WRITER_BUFFER_SIZE, Some(1024 * 1024))
@@ -43,4 +55,24 @@ class SparkConfigUtilSuite extends AnyFunSuiteLike {
assert(SparkConfigUtil.get(conf, GlutenConfig.SHUFFLE_WRITER_BUFFER_SIZE)
=== Some(1024 * 1024))
assert(SparkConfigUtil.get(conf, GlutenConfig.GLUTEN_LOAD_LIB_OS).isEmpty)
}
+
+ test("enum config through SQLConf") {
+ val conf = new SQLConf
+ val glutenConf = new GlutenCoreConfig(conf)
+ val entry = TestConfigs.mode
+
+ assert(glutenConf.getConf(entry) == TestMode.FIRST)
+ TestMode.values.foreach {
+ mode =>
+ conf.setConfString(entry.key,
mode.toString.toLowerCase(java.util.Locale.ROOT))
+ assert(glutenConf.getConf(entry) == mode)
+ }
+ conf.unsetConf(entry.key)
+ assert(glutenConf.getConf(entry) == TestMode.FIRST)
+
+ val error = intercept[IllegalArgumentException] {
+ conf.setConfString(entry.key, "invalid")
+ }
+ assert(error.getMessage.contains(entry.key))
+ }
}
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/execution/PushAggregateThroughJoinSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/execution/PushAggregateThroughJoinSuite.scala
index 44e654e790..9f204ba688 100644
---
a/gluten-substrait/src/test/scala/org/apache/gluten/execution/PushAggregateThroughJoinSuite.scala
+++
b/gluten-substrait/src/test/scala/org/apache/gluten/execution/PushAggregateThroughJoinSuite.scala
@@ -23,8 +23,7 @@ import
org.apache.gluten.extension.joinagg.PushAggregateThroughJoin
import org.apache.spark.SparkConf
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.plans.PlanTest
-import org.apache.spark.sql.catalyst.plans.logical.Aggregate
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
@@ -39,6 +38,8 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
private case class PushdownCase(inputSql: String)
+ private case class ExpectedAggCounts(none: Int, top: Int, all: Int)
+
override protected def sparkConf: SparkConf = {
// Avoid Janino projection codegen here because Spark 4's
QueryExecutionErrors
// has Arrow-typed methods, which breaks test runs as arrow-vector is
excluded.
@@ -93,89 +94,99 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
testCase: PushdownCase,
maxDepth: Int,
expectedPushCount: Int,
- expectedAggCount: Int,
- partialMergeEnabled: Boolean): Unit = {
- withSQLConf(
- GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_ENABLED.key -> "true",
- GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_MAX_DEPTH.key ->
maxDepth.toString,
- GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_ENABLED.key ->
- partialMergeEnabled.toString
- ) {
- val (withoutRuleRows, withoutRuleLogicalPlan, withoutRulePhysicalPlan) =
- withExtraPlanning(Nil, Nil) {
- val df = spark.sql(testCase.inputSql)
- (
- df.collect().toSeq.sortBy(_.toString()),
- df.queryExecution.optimizedPlan,
- finalExecutedPlan(df.queryExecution.executedPlan)
- )
- }
-
- val (withRuleRows, withRuleLogicalPlan, withRulePhysicalPlan) =
- withExtraPlanning(Seq(joinAggregateRule), Nil) {
- joinAggregateRule.resetSuccessfulPushCount()
- val df = spark.sql(testCase.inputSql)
- val withRuleRows = df.collect().toSeq.sortBy(_.toString())
- val withRulePlan = df.queryExecution.optimizedPlan
- val withRulePhysicalPlan =
finalExecutedPlan(df.queryExecution.executedPlan)
- val aggregateNodeCount = withRulePlan.collect { case _: Aggregate =>
1 }.size
- val nodesWithMissingInput = withRulePlan.collect {
- case p if p.missingInput.nonEmpty => p
+ expectedAggCounts: ExpectedAggCounts): Unit = {
+ val modes = Seq(
+ "none" -> expectedAggCounts.none,
+ "top" -> expectedAggCounts.top,
+ "all" -> expectedAggCounts.all)
+ modes.foreach {
+ case (partialMergeMode, expectedAggCount) =>
+ withClue(s"partialMergeMode=$partialMergeMode, maxDepth=$maxDepth: ") {
+ withSQLConf(
+ GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_ENABLED.key -> "true",
+ GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_MAX_DEPTH.key ->
maxDepth.toString,
+ GlutenConfig.PUSH_AGGREGATE_THROUGH_JOIN_PARTIAL_MERGE_MODE.key ->
+ partialMergeMode
+ ) {
+ val (withoutRuleRows, withoutRuleLogicalPlan,
withoutRulePhysicalPlan) =
+ withExtraPlanning(Nil, Nil) {
+ val df = spark.sql(testCase.inputSql)
+ (
+ df.collect().toSeq.sortBy(_.toString()),
+ df.queryExecution.optimizedPlan,
+ finalExecutedPlan(df.queryExecution.executedPlan)
+ )
+ }
+
+ val (withRuleRows, withRuleLogicalPlan, withRulePhysicalPlan) =
+ withExtraPlanning(Seq(joinAggregateRule), Nil) {
+ joinAggregateRule.resetSuccessfulPushCount()
+ val df = spark.sql(testCase.inputSql)
+ val withRuleRows = df.collect().toSeq.sortBy(_.toString())
+ val withRulePlan = df.queryExecution.optimizedPlan
+ val withRulePhysicalPlan =
finalExecutedPlan(df.queryExecution.executedPlan)
+ val aggregateNodeCount = withRulePlan.collect { case _:
Aggregate => 1 }.size
+ val nodesWithMissingInput = withRulePlan.collect {
+ case p if p.missingInput.nonEmpty => p
+ }
+
+ assert(
+ withRulePlan.resolved,
+ s"Optimized plan unresolved:\n${withRulePlan.treeString}\n" +
+ s"MissingInput=${withRulePlan.missingInput}")
+ assert(
+ nodesWithMissingInput.isEmpty,
+ s"Plan has missing input:\n${nodesWithMissingInput
+ .map(_.treeString)
+ .mkString("\n---\n")}")
+ assert(joinAggregateRule.getSuccessfulPushCount ==
expectedPushCount)
+ assert(aggregateNodeCount == expectedAggCount)
+ (withRuleRows, withRulePlan, withRulePhysicalPlan)
+ }
+
+ val (
+ withRuleAndStrategyRows,
+ withRuleAndStrategyLogicalPlan,
+ withRuleAndStrategyPhysicalPlan) =
+ withExtraPlanning(Seq(joinAggregateRule),
Seq(ImplementJoinAggregate(spark))) {
+ joinAggregateRule.resetSuccessfulPushCount()
+ val df = spark.sql(testCase.inputSql)
+ (
+ df.collect().toSeq.sortBy(_.toString()),
+ df.queryExecution.optimizedPlan,
+ finalExecutedPlan(df.queryExecution.executedPlan)
+ )
+ }
+
+ if (debugMode) {
+ // scalastyle:off println
+ println("=== Optimized Plan Before (without
PushJoinAggregatePreAggregation) ===")
+ println(withoutRuleLogicalPlan.treeString)
+ println("=== Optimized Plan After (with
PushJoinAggregatePreAggregation) ===")
+ println(withRuleLogicalPlan.treeString)
+ println(
+ "=== Optimized Plan After (with
PushJoinAggregatePreAggregation and strategy) ===")
+ println(withRuleAndStrategyLogicalPlan.treeString)
+ println("=== Physical Plan Before (without
PushJoinAggregatePreAggregation) ===")
+ println(withoutRulePhysicalPlan.treeString)
+ println("=== Physical Plan After (with
PushJoinAggregatePreAggregation only) ===")
+ println(withRulePhysicalPlan.treeString)
+ println(
+ "=== Physical Plan After (with PushJoinAggregatePreAggregation
and strategy) ===")
+ println(withRuleAndStrategyPhysicalPlan.treeString)
+ println("=== Result Before (without
PushJoinAggregatePreAggregation) ===")
+ println(withoutRuleRows.mkString("\n"))
+ println("=== Result After (with PushJoinAggregatePreAggregation
only) ===")
+ println(withRuleRows.mkString("\n"))
+ println("=== Result After (with PushJoinAggregatePreAggregation
and strategy) ===")
+ println(withRuleAndStrategyRows.mkString("\n"))
+ // scalastyle:on println
+ }
+
+ assertRowsEqual(withRuleRows, withoutRuleRows)
+ assertRowsEqual(withRuleAndStrategyRows, withoutRuleRows)
}
-
- assert(
- withRulePlan.resolved,
- s"Optimized plan unresolved:\n${withRulePlan.treeString}\n" +
- s"MissingInput=${withRulePlan.missingInput}")
- assert(
- nodesWithMissingInput.isEmpty,
- s"Plan has missing input:\n${nodesWithMissingInput
- .map(_.treeString)
- .mkString("\n---\n")}")
- assert(joinAggregateRule.getSuccessfulPushCount == expectedPushCount)
- assert(aggregateNodeCount == expectedAggCount)
- (withRuleRows, withRulePlan, withRulePhysicalPlan)
- }
-
- val (
- withRuleAndStrategyRows,
- withRuleAndStrategyLogicalPlan,
- withRuleAndStrategyPhysicalPlan) =
- withExtraPlanning(Seq(joinAggregateRule),
Seq(ImplementJoinAggregate(spark))) {
- joinAggregateRule.resetSuccessfulPushCount()
- val df = spark.sql(testCase.inputSql)
- (
- df.collect().toSeq.sortBy(_.toString()),
- df.queryExecution.optimizedPlan,
- finalExecutedPlan(df.queryExecution.executedPlan)
- )
}
-
- if (debugMode) {
- // scalastyle:off println
- println("=== Optimized Plan Before (without
PushJoinAggregatePreAggregation) ===")
- println(withoutRuleLogicalPlan.treeString)
- println("=== Optimized Plan After (with
PushJoinAggregatePreAggregation) ===")
- println(withRuleLogicalPlan.treeString)
- println("=== Optimized Plan After (with
PushJoinAggregatePreAggregation and strategy) ===")
- println(withRuleAndStrategyLogicalPlan.treeString)
- println("=== Physical Plan Before (without
PushJoinAggregatePreAggregation) ===")
- println(withoutRulePhysicalPlan.treeString)
- println("=== Physical Plan After (with PushJoinAggregatePreAggregation
only) ===")
- println(withRulePhysicalPlan.treeString)
- println("=== Physical Plan After (with PushJoinAggregatePreAggregation
and strategy) ===")
- println(withRuleAndStrategyPhysicalPlan.treeString)
- println("=== Result Before (without PushJoinAggregatePreAggregation)
===")
- println(withoutRuleRows.mkString("\n"))
- println("=== Result After (with PushJoinAggregatePreAggregation only)
===")
- println(withRuleRows.mkString("\n"))
- println("=== Result After (with PushJoinAggregatePreAggregation and
strategy) ===")
- println(withRuleAndStrategyRows.mkString("\n"))
- // scalastyle:on println
- }
-
- assertRowsEqual(withRuleRows, withoutRuleRows)
- assertRowsEqual(withRuleAndStrategyRows, withoutRuleRows)
}
}
@@ -224,14 +235,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
}
test("pre-aggregate store_sales for sum") {
@@ -249,14 +253,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
test("pre-aggregate store_sales for avg") {
@@ -274,14 +271,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
test("pre-aggregate store_sales for sum on fact table") {
@@ -299,14 +289,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
test("pre-aggregate store_sales for avg on fact table") {
@@ -324,14 +307,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
test("pre-aggregate store_sales for sum on three-way join") {
@@ -351,14 +327,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
}
test("pre-aggregate store_sales for sum and avg on different fact columns on
three-way join") {
@@ -379,14 +348,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
}
test("pre-aggregate store_sales for sum and avg on same fact column on
three-way join") {
@@ -407,14 +369,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
}
test("pre-aggregate store_sales by i_item_desc") {
@@ -432,14 +387,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
test("pre-aggregate store_sales by substr(i_item_desc, 3), 3 ways") {
@@ -459,14 +407,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
}
test("pre-aggregate store_sales for sum with item filter") {
@@ -483,14 +424,7 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
test("pre-aggregate three-way joins independently under union all") {
@@ -533,38 +467,17 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = 1,
expectedPushCount = 3,
- expectedAggCount = 6,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = 1,
- expectedPushCount = 3,
- expectedAggCount = 9,
- partialMergeEnabled = true)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = 2,
- expectedPushCount = 6,
- expectedAggCount = 6,
- partialMergeEnabled = false)
+ expectedAggCounts = ExpectedAggCounts(none = 6, top = 9, all = 9))
runCaseWithMaxDepth(
pushdownCase,
maxDepth = 2,
expectedPushCount = 6,
- expectedAggCount = 12,
- partialMergeEnabled = true)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 6,
- expectedAggCount = 6,
- partialMergeEnabled = false)
+ expectedAggCounts = ExpectedAggCounts(none = 6, top = 9, all = 12))
runCaseWithMaxDepth(
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 6,
- expectedAggCount = 12,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 6, top = 9, all = 12))
}
test("pre-aggregate store_sales for sum on three-way join with maxDepth=1 /
maxDepth=2") {
@@ -584,38 +497,32 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = 1,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = 1,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
runCaseWithMaxDepth(
pushdownCase,
maxDepth = 2,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = 2,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
runCaseWithMaxDepth(
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 2,
- expectedAggCount = 2,
- partialMergeEnabled = false)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 4))
+ }
+
+ test("partial merge is not retained when no side can be pushed") {
+ val pushdownCase = PushdownCase(
+ inputSql = """
+ |SELECT sum(ss_sales_price), sum(i_category_id)
+ |FROM store_sales
+ |JOIN item ON ss_item_sk = i_item_sk
+ |""".stripMargin
+ )
runCaseWithMaxDepth(
pushdownCase,
maxDepth = Int.MaxValue,
- expectedPushCount = 2,
- expectedAggCount = 4,
- partialMergeEnabled = true)
+ expectedPushCount = 0,
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 2, all = 2))
}
test("pre-aggregate with filter inside inner equi-join") {
@@ -633,13 +540,6 @@ class PushAggregateThroughJoinSuite extends PlanTest with
SharedSparkSession {
pushdownCase,
maxDepth = Int.MaxValue,
expectedPushCount = 1,
- expectedAggCount = 2,
- partialMergeEnabled = false)
- runCaseWithMaxDepth(
- pushdownCase,
- maxDepth = Int.MaxValue,
- expectedPushCount = 1,
- expectedAggCount = 3,
- partialMergeEnabled = true)
+ expectedAggCounts = ExpectedAggCounts(none = 2, top = 3, all = 3))
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
