peter-toth commented on code in PR #57576: URL: https://github.com/apache/spark/pull/57576#discussion_r3690100532
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala: ########## @@ -0,0 +1,205 @@ +/* + * 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.{Attribute, 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.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, + input: Seq[Attribute]): Expression = expression.transformUp { + case attribute: AttributeReference => + val ordinal = input.indexWhere(_.exprId == attribute.exprId) + if (ordinal < 0) { + attribute + } else { + AttributeReference("none", attribute.dataType)(ExprId(ordinal)) + } + } + + private def physicalCompatibilityKey( + key: CompatibilityKey, + accuracy: Expression): PhysicalCompatibilityKey = PhysicalCompatibilityKey( + key.child.canonicalized, + accuracy.canonicalized, + key.mode, + key.isDistinct, Review Comment: **Finding 8.** Keying on `key.isDistinct` was the right move for the case I gave in finding 5, but dropping the normalization outright also removed the protection your original comment was reaching for. `OptimizeOneRowPlan` (`OptimizeOneRowPlan.scala:56-61`) sets `isDistinct = false` on *every* aggregate expression when the child has at most one row, and it runs after this batch in both the main `Optimizer` and `AQEOptimizer`. Once that happens a DISTINCT percentile can become semantically equal to a non-DISTINCT one and inherit its `PhysicalAggregation` dedup partner — which is exactly what `physicalCompatibilityKeys` exists to preserve. My finding 5 argument only covered the case where the two groups have a *structurally identical* input and filter. There the identity does cover it: the fused aggregates never dedup, base's scalars dedup pairwise, and both sides end at the same digest count with the same values. When the inputs collide only *canonically*, the fused form can no longer reproduce base's pairing, and the results change. Same shape as `do not fuse canonical input or filter collisions`, with DISTINCT on the first two columns: ```sql 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) ``` Run on `4bc0cf2`: ``` BASELINE (CombineApproximatePercentiles excluded) = [0.0, 0.0, 0.0, 0.0] FUSED = [0.0, 0.0, 1.0, 1.0] ``` Base strips DISTINCT, then dedups columns 3 and 4 onto 1 and 2, so all four read the `a + (b + c)` digest. Fused, the DISTINCT group becomes an array aggregate that can never dedup with the two scalars, so columns 3 and 4 evaluate `(a + b) + c` on their own digest. `do not fuse with an existing array that collides after distinct removal` doesn't catch it because its DISTINCT group asks for `0.1D`, which no non-DISTINCT column shares. The `maxRows <= 1` precondition makes a blunt fix (skip fusion whenever `aggregate.child.maxRows.exists(_ <= 1L)`) too coarse for the catalyst suite, whose `LocalRelation` is empty. A precise version keeps a second, `isDistinct`-insensitive map and only rejects when the colliding key differs structurally, so the mixed-DISTINCT query from finding 5 still fuses: ```scala val distinctInsensitiveKeys = mutable.HashMap.empty[ PhysicalCompatibilityKey, mutable.HashSet[CompatibilityKey]] ``` ```scala physicalCompatibilityKeys.getOrElseUpdate( physicalCompatibilityKey(key, percentile.accuracyExpression), mutable.HashSet.empty) += key distinctInsensitiveKeys.getOrElseUpdate( physicalCompatibilityKey(key, percentile.accuracyExpression) .copy(isDistinct = false), mutable.HashSet.empty) += key ``` ```scala hasSafePhysicalFusion(expressions) && expressions.forall { expression => val percentile = expression.aggregateFunction.asInstanceOf[ApproximatePercentile] val physicalKey = physicalCompatibilityKey(key, percentile.accuracyExpression) physicalCompatibilityKeys(physicalKey).sizeCompare(1) == 0 && // OptimizeOneRowPlan can drop isDistinct later, also during AQE, which lets a DISTINCT // percentile pick up the PhysicalAggregation dedup partner of a non-DISTINCT one. That // only changes results when the two differ structurally. distinctInsensitiveKeys(physicalKey.copy(isDistinct = false)).forall { other => other.child == key.child && other.filter == key.filter } } ``` With that applied the query above returns the baseline `[0.0, 0.0, 0.0, 0.0]`, and both suites stay green (`CombineApproximatePercentilesSuite` 8/8, `ApproximatePercentileQuerySuite` 31/31) — including `respect basic compatibility boundaries`, whose DISTINCT and non-DISTINCT groups share a structurally identical input and filter. Worth adding the query above as a regression. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala: ########## @@ -0,0 +1,205 @@ +/* + * 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.{Attribute, 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.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, + input: Seq[Attribute]): Expression = expression.transformUp { + case attribute: AttributeReference => + val ordinal = input.indexWhere(_.exprId == attribute.exprId) + if (ordinal < 0) { + attribute + } else { + AttributeReference("none", attribute.dataType)(ExprId(ordinal)) + } + } + + 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 = 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)] + 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] + physicalCompatibilityKeys( + physicalCompatibilityKey(key, percentile.accuracyExpression)).sizeCompare(1) == 0 + } + }.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( + expressions.map { expression => + structurallyNormalize(expression.aggregateFunction, aggregate.child.output) + }.toSeq, + key.mode, + key.isDistinct, + key.filter.map(structurallyNormalize(_, aggregate.child.output)), + percentageValues.map(java.lang.Double.doubleToRawLongBits)) + val combined = first.copy(aggregateFunction = percentile.copy( Review Comment: **Finding 9.** `percentile.copy(...)` is the compiler-generated case-class copy, which does not carry tree-node tags over — only `TreeNode#withNewChildren` / `mapChildren` call `copyTagsFrom`. `ApproximatePercentile.prettyName` reads `FunctionRegistry.FUNC_ALIAS` (`ApproximatePercentile.scala:249`), and the registry sets that tag because `approx_percentile` is registered as an alias (`FunctionRegistry.scala:541`, `setAlias = true`), so the fused aggregate silently renames itself in EXPLAIN. On `4bc0cf2`: ``` SELECT approx_percentile(id, 0.5D) FROM range(10) Aggregate [approx_percentile(id#0L, 0.5, 10000, 0, 0) AS approx_percentile(id, 0.5, 10000)#2L] SELECT approx_percentile(id, 0.5D), approx_percentile(id, 0.9D) FROM range(10) Aggregate [percentile_approx(id#3L, [0.5,0.9], 10000, 0, 0)[0] AS approx_percentile(id, 0.5, 10000)#6L, percentile_approx(id#3L, [0.5,0.9], 10000, 0, 0)[1] AS approx_percentile(id, 0.9, 10000)#7L] ``` Output column names are unaffected — `ResolveAliases` fixed them at analysis time — so this is EXPLAIN text only. One line: ```scala val combinedPercentile = percentile.copy( percentageExpression = PercentileFusionArray(identity)) combinedPercentile.copyTagsFrom(percentile) val combined = first.copy(aggregateFunction = combinedPercentile) ``` -- 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]
