sunchao commented on code in PR #57576:
URL: https://github.com/apache/spark/pull/57576#discussion_r3687082450
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -84,6 +84,9 @@ object ConstantFolding extends Rule[LogicalPlan] {
// object and running eval unnecessarily.
case l: Literal => l
+ // This foldable expression carries planning identity that must survive
later optimizer batches.
+ case p: PercentileFusionArray => p
Review Comment:
Updated the PR description to explain the profitability gate,
`PercentileFusionArray`/`PercentileFusionIdentity`, and the `ConstantFolding`
exception. It now also uses the concrete `100` versus `100L` example to explain
why preserving discarded aggregate structure is necessary for safe
exchange/subquery reuse.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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, CreateArray, 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 contextIndependentFoldable: 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,
+ // OptimizeOneRowPlan can remove DISTINCT later, including during AQE.
+ isDistinct = false,
+ 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]]
+ val arrayPercentiles = mutable.ArrayBuffer.empty[AggregateExpression]
+
+ 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
+ } else {
+ arrayPercentiles += 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 combinedPercentile = percentile.copy(
+ percentageExpression = CreateArray(percentages.toSeq))
+ val combined = first.copy(aggregateFunction = combinedPercentile)
+ if (!arrayPercentiles.exists(_.semanticEquals(combined))) {
Review Comment:
Fixed in `4bc0cf22`: removed `arrayPercentiles` and the old
`semanticEquals(combined)` veto. Fusion now directly constructs the
identity-preserving aggregate. The existing Catalyst and SQL regressions verify
that a native `[0.5, 0.9]` aggregate remains separate while colliding scalar
`[0.0, 0.9]` percentiles safely share their own sketch; the SQL case now uses
two digests instead of three.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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, CreateArray, 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 contextIndependentFoldable: Boolean = true
Review Comment:
Fixed in `4bc0cf22`: removed the `contextIndependentFoldable` override. The
expression remains `foldable` for `ApproximatePercentile` validation while
inheriting `contextIndependentFoldable = false`; the existing Catalyst
regression now asserts that property.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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, CreateArray, 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)
Review Comment:
Thanks for suggesting this. I investigated it, but literal-only eligibility
is not sufficient: accuracy accepts any `IntegralType`, so `100`
(`IntegerType`) and `100L` (`LongType`) are both `Literal`s and evaluate to the
same `long` while retaining different Catalyst identities.
```sql
-- Branch A
percentile_approx(1D, 0.5D, 100)
percentile_approx(1D, 0.9D, 100L)
-- Branch B
percentile_approx(1D, 0.5D, 100)
percentile_approx(1D, 0.9D, 100)
```
Before fusion, these branches cannot reuse the same exchange because the
second aggregate's accuracy literal has a different data type. Fusion retains
only the first aggregate's `100`, so removing `PercentileFusionIdentity` makes
both fused aggregates identical. Combined with grouping expressions such as `(a
+ b) + c` and `a + (b + c)`, which canonicalize identically but evaluate
differently, that newly enables incorrect exchange reuse.
This remains a counterexample with `ConstantFolding` enabled because all
parameters are already literals. I extended the existing exchange-reuse
regression with the `100L` case, so I kept the identity wrapper and the
`ConstantFolding` exception while removing the unnecessary guards you
identified.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CombineApproximatePercentiles.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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, CreateArray, 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 contextIndependentFoldable: 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,
+ // OptimizeOneRowPlan can remove DISTINCT later, including during AQE.
+ isDistinct = false,
Review Comment:
Fixed in `4bc0cf22`: `PhysicalCompatibilityKey` now preserves
`key.isDistinct`. The existing compatibility regression uses identical input
and filter for both DISTINCT and non-DISTINCT percentile groups and verifies
that each group fuses independently. `PercentileFusionIdentity` still retains
distinctness, so later DISTINCT removal cannot collapse the fused aggregates.
--
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]
