peter-toth commented on code in PR #57576: URL: https://github.com/apache/spark/pull/57576#discussion_r3698282434
########## 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 Review Comment: **Finding 12.** The description doesn't reflect the flag, and since it becomes the commit message this is what `git log` will carry. Three concrete gaps: - **"What changes were proposed in this PR?"** never names `spark.sql.optimizer.combineApproximatePercentiles.enabled` or says the rule is opt-in. As written it reads as if fusion now applies to every eligible query. - **"Does this PR introduce _any_ user-facing change?"** answers "No", then claims "Eligible batch queries use fewer percentile aggregation buffers and can perform less sketch update, serialization, shuffle, and merge work." With the default `false` that second sentence is false — nothing changes unless a user opts in. And the config itself is user-facing: it isn't `.internal()`, so it lands in the generated SQL config table. The honest answer is "Yes: a new config, default `false`, no behavior change unless enabled", plus a line on why it's off (I assume soak time before flipping — worth stating either way). - **"How was this PR tested?"** still says "the focused Catalyst suite passes all 8 tests, the full approximate-percentile SQL suite passes all 31 tests"; those are 9 and 33 now. The paragraph also still describes disabling fusion by rule exclusion, which the flag replaced in the tests. Same class as finding 3 from round 2 — the code grew a moving part the description doesn't mention. ########## 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: **Finding 13.** This enables fusion for the whole suite, including the 20 tests that predate this PR, so the shipped default is no longer covered here at all. It isn't a no-op for those tests. `percentile_approx, single percentile value` (`:443`) asks for seven scalar percentiles over one column; after constant folding `0.0`/`0` and `1.0`/`1` collapse, leaving five distinct canonical groups, so `hasSafePhysicalFusion` passes and the query now runs as one array-valued aggregate. `percentile_approx, with different accuracies`, the `null`-input cases and the interval/TIME cases are all affected the same way. Before this PR they exercised the unfused path; now nothing in the suite does. Since the default is `false`, I'd rather the suite keep testing the default and the 13 fusion tests opt in locally — two of them already do (`approximate percentile fusion can be disabled`, and `checkMatchesUnfusedBaseline`), so it's the same shape: ```scala // drop the sparkConf override, and in each fusion test: withSQLConf(SQLConf.COMBINE_APPROXIMATE_PERCENTILES_ENABLED.key -> "true") { ... } ``` A `private def withFusion(f: => Unit)` wrapper would keep that from being noisy. Alternatively keep the override and move the fusion tests into their own suite, which also stops them from drifting when someone edits the shared ones. ########## 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: **Finding 14.** This test only asserts the result row, and `Row(3, 5, 3)` is what the query returns with fusion disabled too — so it passes whether or not the fix under test is in effect. That matters more here than usual: it's the only regression for finding 10, and the fusion it depends on comes entirely from the suite-level `sparkConf` override at `:45` rather than from anything in the test. If that override moves (see finding 13), or a future eligibility condition stops this plan shape from fusing, the test keeps passing and the `resultId` collision is unguarded again. Please pin down that the `c1` branch actually fused: ```scala val query = spark.sql( """WITH c AS ( ... |""".stripMargin) checkAnswer(query, Row(3, 5, 3)) assert(query.queryExecution.optimizedPlan.exists(_.expressions.exists(_.exists { case percentile: ApproximatePercentile => percentile.percentageExpression.dataType.isInstanceOf[ArrayType] case _ => false }))) ``` (needs `ArrayType` added to the `org.apache.spark.sql.types` import). `assertPercentileDigestCount(query, 1)` also distinguishes the two cases — `c1`'s aggregate holds two `ApproximatePercentile`s unfused and one fused — but it's indirect enough that the explicit check reads better as the guard for this specific bug. -- 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]
