sunchao commented on code in PR #57576:
URL: https://github.com/apache/spark/pull/57576#discussion_r3705822024
##########
sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala:
##########
@@ -35,7 +42,403 @@ import org.apache.spark.tags.SlowSQLTest
class ApproximatePercentileQuerySuite extends SharedSparkSession {
import testImplicits._
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set(SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED.key,
"true")
Review Comment:
Fixed in `03ca69ee`. I removed the suite-wide `sparkConf` override and added
a small `fusionTest` helper that enables the flag only around
optimization-specific tests. The 20 pre-existing percentile tests therefore run
with the shipped `false` default again; the explicit disabled-path regression
also verifies that the original two physical digests remain separate.
##########
sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala:
##########
@@ -35,7 +42,403 @@ import org.apache.spark.tags.SlowSQLTest
class ApproximatePercentileQuerySuite extends SharedSparkSession {
import testImplicits._
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set(SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED.key,
"true")
+
private val table = "percentile_approx"
+ private val constantFoldingRule =
+ "org.apache.spark.sql.catalyst.optimizer.ConstantFolding"
+
+ private def excludedRules: Seq[String] = {
+ spark.sessionState.conf.optimizerExcludedRules.toSeq
+ .flatMap(_.split(","))
+ .map(_.trim)
+ .filter(_.nonEmpty)
+ }
+
+ private def assertPercentileDigestCount(query: DataFrame, expected: Int):
Unit = {
+ val counts = query.queryExecution.sparkPlan.collect {
+ case aggregate: ObjectHashAggregateExec =>
+ aggregate.aggregateExpressions.count(
+ _.aggregateFunction.isInstanceOf[ApproximatePercentile])
+ }
+ assert(counts.nonEmpty)
+ assert(counts.forall(_ == expected), counts)
+ }
+
+ private def checkMatchesUnfusedBaseline(sql: String, expectedDigests: Int):
Unit = {
+ val withoutConstantFolding = (excludedRules :+
constantFoldingRule).distinct
+ val baseline = withSQLConf(
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
withoutConstantFolding.mkString(","),
+ SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED.key -> "false") {
+ spark.sql(sql).collect().toSeq
+ }
+
+ withSQLConf(
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
withoutConstantFolding.mkString(",")) {
+ val query = spark.sql(sql)
+ checkAnswer(query, baseline)
+ assertPercentileDigestCount(query, expectedDigests)
+ }
+ }
+
+ test("approximate percentile fusion can be disabled") {
+ withSQLConf(SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED.key ->
"false") {
+ val query = spark.sql(
+ "SELECT percentile_approx(id, 0.5D), percentile_approx(id, 0.9D) FROM
range(10)")
+ checkAnswer(query, Row(4L, 8L))
+ assertPercentileDigestCount(query, 2)
+ }
+ }
+
+ test("compatible scalar percentiles share one physical percentile digest") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ val query = spark.sql(
+ s"""SELECT
+ | approx_percentile(col, 0.5, 10000),
+ | approx_percentile(col, 0.9, 10000),
+ | approx_percentile(col, 0.95, 10000)
+ |FROM $table
+ |""".stripMargin)
+
+ checkAnswer(query, Row(500, 900, 950))
+ assertPercentileDigestCount(query, 1)
+ val optimizedPercentiles =
query.queryExecution.optimizedPlan.expressions.flatMap {
+ _.collect { case percentile: ApproximatePercentile => percentile }
+ }
+ assert(optimizedPercentiles.nonEmpty)
+ assert(optimizedPercentiles.forall(_.prettyName == "approx_percentile"))
+ val modes = query.queryExecution.sparkPlan.collect {
+ case aggregate: ObjectHashAggregateExec =>
+ aggregate.aggregateExpressions.collect {
+ case expression @ AggregateExpression(_: ApproximatePercentile, _,
_, _, _) =>
+ expression.mode
+ }
+ }.flatten.toSet
+ assert(modes == Set(Partial, Final))
+ }
+ }
+
+ test("do not fuse duplicate percentages already shared by physical
planning") {
+ withTempView(table) {
+ (1 to 1000).toDF("col").createOrReplaceTempView(table)
+ val query = spark.sql(
+ s"""SELECT
+ | percentile_approx(col, 0.5D),
+ | percentile_approx(col, 0.25D + 0.25D)
+ |FROM $table
+ |""".stripMargin)
+
+ checkAnswer(query, Row(500, 500))
+ val percentiles = query.queryExecution.sparkPlan.collect {
+ case aggregate: ObjectHashAggregateExec =>
+ aggregate.aggregateExpressions.collect {
+ case AggregateExpression(
+ percentile: ApproximatePercentile, _, _, _, _) => percentile
+ }
+ }.flatten
+ assert(percentiles.nonEmpty)
+ assert(percentiles.forall(_.percentageExpression.dataType == DoubleType))
+ }
+ }
+
+ test("preserve structural input and filter evaluation") {
+ checkAnswer(
+ spark.sql(
+ """SELECT
+ | percentile_approx((a + b) + c, 0.5D),
+ | percentile_approx(a + (b + c), 0.9D)
+ |FROM VALUES (
+ | CAST(10000000000000000 AS DOUBLE),
+ | CAST(-10000000000000000 AS DOUBLE),
+ | CAST(1 AS DOUBLE)
+ |) AS t(a, b, c)
+ |""".stripMargin),
+ Row(1.0d, 0.0d))
+
+ checkAnswer(
+ spark.sql(
+ """SELECT
+ | percentile_approx(v, 0.5D)
+ | FILTER (WHERE (a + b) + c = 1D),
+ | percentile_approx(v, 0.9D)
+ | FILTER (WHERE a + (b + c) = 1D)
+ |FROM VALUES (
+ | 7,
+ | CAST(10000000000000000 AS DOUBLE),
+ | CAST(-10000000000000000 AS DOUBLE),
+ | CAST(1 AS DOUBLE)
+ |) AS t(v, a, b, c)
+ |""".stripMargin),
+ Row(7, null))
+
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ val exception = intercept[SparkArithmeticException] {
+ spark.sql(
+ """SELECT
+ | percentile_approx(a + (b + c), 0.5D),
+ | percentile_approx((a + b) + c, 0.9D)
+ |FROM VALUES (
+ | CAST(2147483647 AS INT),
+ | CAST(1 AS INT),
+ | CAST(-1 AS INT)
+ |) AS t(a, b, c)
+ |""".stripMargin).collect()
+ }
+ assert(exception.getCondition == "ARITHMETIC_OVERFLOW")
+ }
+ }
+
+ test("do not fuse canonical input or filter collisions") {
+ checkAnswer(
+ spark.sql(
+ """SELECT
+ | percentile_approx(a + (b + c), 0.5D),
+ | percentile_approx((a + b) + c, 0.5D),
+ | percentile_approx((a + b) + c, 0.9D),
+ | percentile_approx(a + (b + c), 0.9D)
+ |FROM VALUES (
+ | CAST(10000000000000000 AS DOUBLE),
+ | CAST(-10000000000000000 AS DOUBLE),
+ | CAST(1 AS DOUBLE)
+ |) AS t(a, b, c)
+ |""".stripMargin),
+ Row(0.0d, 0.0d, 1.0d, 1.0d))
+
+ val crossDistinctCollision =
+ """SELECT
+ | percentile_approx(DISTINCT a + (b + c), 0.5D),
+ | percentile_approx(DISTINCT a + (b + c), 0.9D),
+ | percentile_approx((a + b) + c, 0.5D),
+ | percentile_approx((a + b) + c, 0.9D)
+ |FROM VALUES (
+ | CAST(10000000000000000 AS DOUBLE),
+ | CAST(-10000000000000000 AS DOUBLE),
+ | CAST(1 AS DOUBLE)
+ |) AS t(a, b, c)
+ |""".stripMargin
+ val unfusedBaseline = withSQLConf(
+ SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED.key -> "false") {
+ spark.sql(crossDistinctCollision).collect().toSeq
+ }
+ checkAnswer(spark.sql(crossDistinctCollision), unfusedBaseline)
+
+ checkAnswer(
+ spark.sql(
+ """SELECT
+ | percentile_approx(v, array(0.5D, 0.9D))
+ | FILTER (WHERE a + (b + c) = 0D),
+ | percentile_approx(v, 0.5D)
+ | FILTER (WHERE (a + b) + c = 0D),
+ | percentile_approx(v, 0.9D)
+ | FILTER (WHERE (a + b) + c = 0D)
+ |FROM VALUES (
+ | 7,
+ | CAST(10000000000000000 AS DOUBLE),
+ | CAST(-10000000000000000 AS DOUBLE),
+ | CAST(1 AS DOUBLE)
+ |) AS t(v, a, b, c)
+ |""".stripMargin),
+ Row(Seq(7, 7), null, null))
+ }
+
+ test("preserve canonically colliding parameters") {
+ val cases = Seq(
+ (
+ """SELECT
+ | percentile_approx(
+ | v, 0.5D, CAST((1e16D + -1e16D) + 3D AS INT)),
+ | percentile_approx(
+ | v, 0.5D, CAST(1e16D + (-1e16D + 3D) AS INT)),
+ | percentile_approx(
+ | v, 0.9D, CAST(1e16D + (-1e16D + 3D) AS INT)),
+ | percentile_approx(
+ | v, 0.9D, CAST((1e16D + -1e16D) + 3D AS INT))
+ |FROM range(100) AS t(v)
+ |""".stripMargin,
+ 2),
+ (
+ """SELECT
+ | percentile_approx(
+ | id, array((1e16D + -1e16D) + 0.5D, 0.9D)),
+ | percentile_approx(
+ | id, 1e16D + (-1e16D + 0.5D)),
+ | percentile_approx(id, 0.9D)
+ |FROM range(100)
+ |""".stripMargin,
+ 2))
+
+ cases.foreach { case (sql, expectedDigests) =>
+ checkMatchesUnfusedBaseline(sql, expectedDigests)
+ }
+ }
+
+ test("do not fuse with an existing array that collides after distinct
removal") {
+ checkAnswer(
+ spark.sql(
+ """SELECT
+ | percentile_approx(
+ | DISTINCT a + (b + c), array(0.5D, 0.9D)),
+ | percentile_approx(DISTINCT a + (b + c), 0.1D),
+ | percentile_approx((a + b) + c, 0.5D),
+ | percentile_approx((a + b) + c, 0.9D)
+ |FROM VALUES (
+ | CAST(10000000000000000 AS DOUBLE),
+ | CAST(-10000000000000000 AS DOUBLE),
+ | CAST(1 AS DOUBLE)
+ |) AS t(a, b, c)
+ |""".stripMargin),
+ Row(Seq(0.0d, 0.0d), 0.0d, 1.0d, 1.0d))
+ }
+
+ test("fused percentiles use fresh result IDs across CTE references") {
Review Comment:
Fixed in `03ca69ee`. The CTE regression now opts in through `fusionTest`,
uses `range(1, 6)` so the input cannot disappear into a folded local relation,
and explicitly asserts that an array-valued `ApproximatePercentile` exists in
`optimizedPlan.subqueriesAll`. The subquery traversal is important here:
`MergeSubplans` moves the fused aggregate into scalar subqueries, so
`optimizedPlan.exists(...)` alone misses it. The result check and separate
fresh-`resultId` Catalyst assertion remain.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.spark.sql.catalyst.optimizer
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, ExprId, GetArrayItem, LeafExpression, Literal, NamedExpression}
+import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
AggregateMode, ApproximatePercentile}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DoubleType}
+
+private[optimizer] case class PercentileFusionIdentity(
+ aggregateFunctions: Seq[Expression],
+ mode: AggregateMode,
+ isDistinct: Boolean,
+ filter: Option[Expression],
+ percentageBits: Seq[Long])
+
+/**
+ * Foldable percentage array that retains the original scalar aggregate
structures in equality.
+ *
+ * Fusion removes those structures from the physical aggregate. Keeping them
here prevents
+ * subquery or exchange reuse from equating plans that were distinct before
fusion.
+ */
+private[optimizer] case class PercentileFusionArray(identity:
PercentileFusionIdentity)
+ extends LeafExpression with CodegenFallback {
+ override def foldable: Boolean = true
+ override def nullable: Boolean = false
+ override def dataType: ArrayType = ArrayType(DoubleType, containsNull =
false)
+
+ private lazy val value = new GenericArrayData(
+ identity.percentageBits.map(java.lang.Double.longBitsToDouble))
+ private lazy val literal = Literal(value, dataType)
+
+ override def eval(input: InternalRow): Any = value
+ override def toString: String = literal.toString
+ override def sql: String = literal.sql
+}
+
+/**
+ * Combines scalar approximate percentiles that can share the same percentile
digest.
+ *
+ * An approximate percentile digest depends on its input, accuracy, filter,
distinctness, and
+ * aggregate mode, but not on the percentile requested from the completed
digest. Consequently,
+ * compatible scalar percentiles can be calculated by one array-valued
aggregate and projected
+ * back to their original scalar outputs.
+ *
+ * Inputs and filters must retain their original expression structure so that
floating-point
+ * evaluation and ANSI overflow behavior are preserved. Streaming aggregates
are left unchanged
+ * to preserve the value schemas of existing checkpoints.
+ */
+object CombineApproximatePercentiles extends Rule[LogicalPlan] {
+
+ private case class CompatibilityKey(
+ child: Expression,
+ accuracy: Long,
+ mode: AggregateMode,
+ isDistinct: Boolean,
+ filter: Option[Expression])
+
+ private case class PhysicalCompatibilityKey(
+ child: Expression,
+ accuracy: Expression,
+ mode: AggregateMode,
+ isDistinct: Boolean,
+ filter: Option[Expression])
+
+ private def structurallyNormalize(
+ expression: Expression,
+ inputOrdinals: scala.collection.Map[ExprId, Int]): Expression =
expression.transformUp {
+ case attribute: AttributeReference =>
+ inputOrdinals.get(attribute.exprId) match {
+ case Some(ordinal) => AttributeReference("none",
attribute.dataType)(ExprId(ordinal))
+ case None => attribute
+ }
+ }
+
+ private def physicalCompatibilityKey(
+ key: CompatibilityKey,
+ accuracy: Expression): PhysicalCompatibilityKey =
PhysicalCompatibilityKey(
+ key.child.canonicalized,
+ accuracy.canonicalized,
+ key.mode,
+ key.isDistinct,
+ key.filter.map(_.canonicalized))
+
+ private def hasSafePhysicalFusion(
+ expressions: scala.collection.Iterable[AggregateExpression]): Boolean = {
+ val physicalGroups = expressions.groupBy(_.canonicalized)
+ // PhysicalAggregation already shares a digest within each canonical
group. Fusion must both
+ // remove a digest and preserve cases where canonical percentages evaluate
differently.
+ physicalGroups.sizeCompare(1) > 0 && physicalGroups.values.forall { group
=>
+ group.iterator.map { expression =>
+ expression.aggregateFunction
+ .asInstanceOf[ApproximatePercentile]
+ .percentageExpression
+ .eval()
+ }.toSet.sizeCompare(1) == 0
+ }
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!conf.getConf(SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED)) return
plan
+
+ plan.transformUpWithPruning(_.containsPattern(AGGREGATE), ruleId) {
+ case aggregate: Aggregate if aggregate.resolved &&
!aggregate.isStreaming =>
+ combine(aggregate)
+ }
+ }
+
+ private def combine(aggregate: Aggregate): Aggregate = {
+ val compatible = mutable.LinkedHashMap.empty[
+ CompatibilityKey, mutable.ArrayBuffer[AggregateExpression]]
+ // PhysicalAggregation deduplicates semantically equivalent aggregates.
Track every logical
+ // key that shares a physical key so fusion does not change that existing
deduplication.
+ val physicalCompatibilityKeys = mutable.HashMap.empty[
+ PhysicalCompatibilityKey, mutable.HashSet[CompatibilityKey]]
+
+ aggregate.aggregateExpressions.foreach(_.foreach {
+ case expression @ AggregateExpression(
+ percentile: ApproximatePercentile, mode, isDistinct, filter, _)
+ if percentile.child.deterministic &&
+ filter.forall(_.deterministic) =>
+ val key = CompatibilityKey(
+ percentile.child,
+ // Analysis already validates that accuracy is foldable, non-null,
and in range.
+ percentile.accuracyExpression.eval().asInstanceOf[Number].longValue,
+ mode,
+ isDistinct,
+ filter)
+ physicalCompatibilityKeys.getOrElseUpdate(
+ physicalCompatibilityKey(key, percentile.accuracyExpression),
+ mutable.HashSet.empty) += key
+ if (percentile.percentageExpression.dataType == DoubleType) {
+ compatible.getOrElseUpdate(key, mutable.ArrayBuffer.empty) +=
expression
+ }
+ case _ =>
+ })
+
+ val replacements = mutable.HashMap.empty[ExprId, (AggregateExpression,
Int)]
+ lazy val inputOrdinals = {
+ val ordinals = mutable.HashMap.empty[ExprId, Int]
+ aggregate.child.output.zipWithIndex.foreach { case (attribute, ordinal)
=>
+ ordinals.getOrElseUpdate(attribute.exprId, ordinal)
+ }
+ ordinals
+ }
+ compatible.iterator.map { case (key, expressions) =>
+ key -> expressions.distinctBy(_.resultId)
+ }.filter { case (key, expressions) =>
+ hasSafePhysicalFusion(expressions) && expressions.forall { expression =>
+ val percentile =
expression.aggregateFunction.asInstanceOf[ApproximatePercentile]
+ val physicalKey = physicalCompatibilityKey(key,
percentile.accuracyExpression)
+ // OptimizeOneRowPlan can erase DISTINCT after fusion. Across
distinctness boundaries,
+ // canonical matches are safe only when their original inputs and
filters also match.
+ physicalCompatibilityKeys(physicalKey).sizeCompare(1) == 0 &&
+ physicalCompatibilityKeys
+ .get(physicalKey.copy(isDistinct = !physicalKey.isDistinct))
+ .forall(_.forall(other => other.child == key.child && other.filter
== key.filter))
+ }
+ }.foreach { case (key, expressions) =>
+ val first = expressions.head
+ val percentile =
first.aggregateFunction.asInstanceOf[ApproximatePercentile]
+ val percentages = expressions.map { expression =>
+ expression.aggregateFunction
+ .asInstanceOf[ApproximatePercentile]
+ .percentageExpression
+ }
+ val percentageValues =
percentages.map(_.eval().asInstanceOf[Double]).toSeq
+ val identity = PercentileFusionIdentity(
Review Comment:
Addressed both points. The PR description now explicitly says that, when
enabled, preserving the original aggregate structures can prevent pre-existing
unsafe exchange/subquery reuse and thereby correct a result that was already
wrong without fusion.
I also added an end-to-end regression over `range(0, 50000, 1, 4)` with
accuracy `100` and p10/p50/p90. It compares the fused results against the same
query with the flag disabled and asserts that the fused plan uses one physical
digest, exercising both compressed summaries and merged partial digests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
