peter-toth commented on code in PR #57576:
URL: https://github.com/apache/spark/pull/57576#discussion_r3681315158


##########
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:
   **Finding 4.** `contextIndependentFoldable = true` breaks the post-condition 
that `ConstantFolding.constantFolding` returns a `Literal` for such an 
expression, and there is an existing caller that relies on that for termination.
   
   
`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala:97`:
   
   ```scala
   case _ if expr.contextIndependentFoldable
       && 
SQLConf.get.getConfByKeyStrict[Boolean]("spark.sql.optimizer.datasourceV2ExprFolding")
 =>
     // If the expression is context independent foldable, we can convert it to 
a literal.
     val constantExpr = ConstantFolding.constantFolding(expr)
     generateExpression(constantExpr, isPredicate)
   ```
   
   With the new skip in `ConstantFolding` (`optimizer/expressions.scala:88`), 
`constantFolding` hands back the same `PercentileFusionArray`, so 
`generateExpression` re-enters the same branch on an identical argument — 
unbounded recursion, `StackOverflowError`. `Literal` is safe there only because 
`case literal: Literal` matches first, and that conf defaults to `true`.
   
   I could not find a reachable path today: `V2ScanRelationPushDown` runs in 
`earlyScanPushDownRules`, well before the `Combine Approximate Percentiles` 
batch, and the other `V2ExpressionBuilder` callers are column defaults, 
generated columns, check constraints and predicate pushdown — none of which can 
contain an `AggregateExpression`. So this is latent rather than live. But the 
override buys nothing either: the doc on `contextIndependentFoldable` is about 
evaluating expressions during DDL (`CREATE TABLE` defaults, views, 
constraints), and an optimizer-internal marker never appears in a DDL 
expression. Simplest fix is to delete the line and let it default to `false` — 
`foldable = true` is all `ApproximatePercentile.checkInputDataTypes` needs.
   



##########
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:
   **Finding 3.** This line and the `PercentileFusionArray` / 
`PercentileFusionIdentity` classes are the two most consequential parts of the 
PR, and neither appears in the description — "What changes were proposed in 
this PR?" describes only the fusion and the compatibility gate.
   
   `ConstantFolding` is a rule every Spark developer has a mental model of, and 
this adds a bespoke exception to it. Since the description becomes the commit 
message, a reader of `git log` gets no signal that the rule was touched at all. 
Please extend the description to cover:
   
   - the new `PercentileFusionArray` expression, what identity it carries, and 
why the fused percentage array cannot be a plain `Literal` (the fused aggregate 
keeps only `first.aggregateFunction`'s `accuracyExpression`, so the discarded 
structures have to live somewhere that participates in equality);
   - this `ConstantFolding` skip, and why folding it would re-open exchange / 
subquery reuse between plans that were distinct before fusion;
   - the profitability gate added since the last revision — fusion now fires 
only when it removes a digest, so `percentile_approx(col, 0.5D)` twice stays 
unfused.
   



##########
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:
   **Finding 5.** Erasing `isDistinct` here makes a DISTINCT group and a 
non-DISTINCT group over the same input, accuracy and filter collapse onto one 
`PhysicalCompatibilityKey`, so the size-1 check at `:170` rejects **both** and 
nothing fuses:
   
   ```sql
   SELECT percentile_approx(v, 0.5D), percentile_approx(v, 0.9D),
          percentile_approx(DISTINCT v, 0.5D), percentile_approx(DISTINCT v, 
0.9D)
   FROM t
   ```
   
   Four digests on base, two are achievable, it stays at four. The existing 
`respect basic compatibility boundaries` test doesn't catch this because its 
distinct and filtered groups differ in `filter` as well, so they land on 
different physical keys.
   
   The stated reason doesn't hold up. If `OptimizeOneRowPlan` later drops 
DISTINCT (`OptimizeOneRowPlan.scala:59`), the two fused aggregates still don't 
dedup, because `PercentileFusionIdentity` records `isDistinct` (`:191`) — the 
identities differ no matter what happens to `AggregateExpression.isDistinct`. 
That leaves two digests, which is exactly what base ends up with (its four 
scalars dedup pairwise after DISTINCT removal), with the same values. So 
allowing fusion here is a win when DISTINCT survives and neutral when it 
doesn't.
   
   Suggest keying on `key.isDistinct` and adding the mixed-DISTINCT query as a 
regression — or, if there is a divergence I'm not seeing, please post it, 
because as written the comment describes a hazard the identity already covers.
   



##########
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:
   **Finding 6.** This guard can no longer block a dedup — it only blocks 
fusion.
   
   Two reasons. It compares `combined`, whose percentage is a `CreateArray`, 
but the expression that actually reaches the plan is `identifiedCombined` 
carrying a `PercentileFusionArray`, which can never be semantically equal to a 
natively written array percentile. And `checkInputDataTypes` requires the 
percentage to be foldable, so with `ConstantFolding` on (the default) a native 
`percentile_approx(v, array(0.5D, 0.9D))` carries a folded `Literal(ArrayData)` 
while `combined` carries a `CreateArray` — different classes, `semanticEquals` 
false. The guard therefore only fires when `ConstantFolding` is excluded, i.e. 
in the catalyst suite, where it now suppresses a fusion the identity has 
already made safe.
   
   Concretely, the `arrayResult` case in `do not fuse canonically equal 
percentages that evaluate differently` 
(`CombineApproximatePercentilesSuite.scala:236`) is safe to fuse now: the fused 
aggregate would carry `PercentileFusionArray` with bits `[0.0, 0.9]`, the 
pre-existing array aggregate keeps its own digest, every value is unchanged, 
and the plan drops from three digests to two.
   
   Suggest dropping the check and relaxing that assertion. If it is meant as 
belt-and-braces against a future change, compare `identifiedCombined` and say 
so in the comment, because as written it reads like a live safety guard.
   



##########
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:
   **Finding 7 (alternative).** The identity machinery could be replaced by a 
one-line eligibility condition, at the cost of not firing when 
`ConstantFolding` is excluded.
   
   Every collision this class exists to prevent comes from an expression that 
canonicalizes equal to another but evaluates differently. For the two 
parameters the identity protects beyond what `physicalCompatibilityKeys` 
already covers — percentage and accuracy — 
`ApproximatePercentile.checkInputDataTypes` requires both to be **foldable**, 
so on the default path `ConstantFolding` has replaced them with `Literal`s 
inside `operatorOptimizationBatch`, long before the `Combine Approximate 
Percentiles` batch. Once they are literals, canonical equality implies equal 
values and the collision cannot exist. That is why `preserve pre-fusion 
identity across exchange reuse` has to exclude `ConstantFolding` to construct 
one.
   
   Input and filter collisions are a different class and stay covered without 
the identity: within one `Aggregate` the `physicalCompatibilityKeys` size-1 
check blocks them, and across `Aggregate`s base already canonicalizes `(a+b)+c` 
and `a+(b+c)` to the same form for the *scalar* percentiles too, so fusion is 
at parity there.
   
   So the gate at `:167` gains:
   
   ```scala
   // Only fuse literal parameters, so that a canonical match implies an equal 
value and fused
   // aggregates cannot newly collide under exchange or subquery reuse.
   val parametersAreLiteral = expressions.forall { expression =>
     val percentile = 
expression.aggregateFunction.asInstanceOf[ApproximatePercentile]
     percentile.percentageExpression.isInstanceOf[Literal] &&
       percentile.accuracyExpression.isInstanceOf[Literal]
   }
   ```
   
   and `PercentileFusionArray`, `PercentileFusionIdentity`, 
`structurallyNormalize`, the `ConstantFolding` case at 
`optimizer/expressions.scala:88` and finding 6's guard all go away — roughly 90 
lines, plus no change to a rule the whole optimizer depends on. The three 
"canonically equal but differently evaluated" negative tests keep passing 
unchanged, since their percentages and accuracies are `Add`/`Cast` trees rather 
than literals.
   
   The honest counter-argument: fusion silently stops firing for anyone who 
puts `ConstantFolding` in `spark.sql.optimizer.excludedRules`, the baseline 
comparisons in `ApproximatePercentileQuerySuite` (which exclude it to get an 
unfused reference) would need reworking to exclude only 
`CombineApproximatePercentiles`, and a couple of `toString`/`sql` assertions 
would move to the `CreateArray` rendering. Your call whether that trade is 
worth it — I'd take it, but it's a scope question, not a defect.
   



-- 
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]

Reply via email to