peter-toth commented on code in PR #57576:
URL: https://github.com/apache/spark/pull/57576#discussion_r3699982181
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -255,6 +255,10 @@ abstract class Optimizer(catalogManager: CatalogManager)
Batch("Eliminate Sorts", Once,
EliminateSorts,
RemoveRedundantSorts),
+ // Run after operator optimization normally folds accuracy expressions and
before
+ // RewriteDistinctAggregates so fused distinct percentiles are rewritten
correctly.
+ Batch("Combine Approximate Percentiles", Once,
Review Comment:
**Finding 16.** This batch sits in `Optimizer.defaultBatches`, but
`SparkOptimizer` appends its own batches after `super.defaultBatches` —
including `Batch("MergeSubplans", Once, MergeSubplans,
RewriteDistinctAggregates)` at `SparkOptimizer.scala:72-74`. `MergeSubplans` is
on by default and its job is to fold one-row subplans into a single
`Aggregate`, so the most natural spelling of the query this PR exists for is
created *after* fusion has already run and never gets fused.
Measured on this head, flag on, AQE off:
```sql
SELECT (SELECT percentile_approx(v, 0.5D) FROM t), (SELECT
percentile_approx(v, 0.9D) FROM t)
```
optimized plan:
```
Aggregate [percentile_approx(v#70, 0.5, 10000, 0, 0) AS ...,
percentile_approx(v#70, 0.9, 10000, 0, 0) AS ...]
```
Two scalar percentiles, same input, same accuracy, in one `Aggregate` —
exactly the shape the rule collapses — left untouched.
That batch already re-runs `RewriteDistinctAggregates` for precisely this
reason, so listing the rule there too is the established pattern:
```scala
Batch("MergeSubplans", Once,
MergeSubplans,
CombineApproximatePercentiles,
RewriteDistinctAggregates),
```
Worth checking the ordering constraint in the comment above still reads
correctly if you do this (the rule wants to run before
`RewriteDistinctAggregates`, which that batch satisfies).
##########
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:
**Finding 15.** `PercentileFusionIdentity` keeps the originals verbatim —
`structurallyNormalize` only rewrites `AttributeReference` exprIds to
child-output ordinals — so identity equality is strictly *finer* than plan
canonicalization. That is deliberate, and it does block the hazard the
`100`/`100L` pair of `preserve pre-fusion identity across exchange reuse`
covers. But it cuts the other way too: it removes reuse that base already had,
and that changes results.
`Add` reorders commutative operands even for `DoubleType`, so `(a+b)+c` and
`a+(b+c)` canonicalize equal and base reuses one subquery for the other.
Measured on this head with `spark.sql.adaptive.enabled=false`,
`spark.sql.subquery.reuse=true`, `spark.sql.exchange.reuse=false`:
```sql
SELECT
(SELECT named_struct('p50', percentile_approx((a + b) + c, 0.5D),
'p90', percentile_approx((a + b) + c, 0.9D)) FROM t),
(SELECT named_struct('p50', percentile_approx(a + (b + c), 0.5D),
'p90', percentile_approx(a + (b + c), 0.9D)) FROM t)
```
over the single row `(1e16, -1e16, 1)`:
```
flag off: [[1.0,1.0],[1.0,1.0]]
flag on : [[1.0,1.0],[0.0,0.0]]
```
I think the fused answer is the **better** one — `(a+b)+c` is `1.0` and
`a+(b+c)` is `0.0`, so base is reusing a plan that is not equivalent — so I am
not asking you to change the behaviour. The ask is narrower: "No SQL syntax,
public API, output schema, or percentile result changes" is not true with the
flag on, and a result change is worth being a written-down trade rather than a
surprise. One sentence in the description, alongside finding 12.
Related, and probably why this went unnoticed: **no test in the suite could
detect a percentile value change at all.** The largest input is `(1 to 1000)`
with the default accuracy, so `compress()`'s `mergeThreshold` is `2 * 1e-4 *
1000 = 0.2`, and `compressImmut` only merges when `g + head.g + head.delta <
mergeThreshold` with `g >= 1` — so no compression ever happens and every digest
in the suite is mathematically exact. Everything else is a one-row `VALUES`,
`range(10)` or `range(100)`. One test on a genuinely compressed and merged
digest (say ~1e6 rows, `accuracy` 100, several partitions) compared against the
same query with the flag off would actually back the no-result-change claim.
##########
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(
Review Comment:
**Finding 18.** This `checkAnswer` and the one in `do not fuse with an
existing array that collides after distinct removal` (`:277`) both assert only
result values, and in both the values are the same whether the guard fires or
not — so each would pass against a broken guard. This is the same class as
finding 14, on the guard side rather than the fusion side.
Measured on this head, with the flag on and off:
- Here, both scalars sit under `FILTER ((a + b) + c = 0D)`, which is false
for the single input row, so both are `null` regardless of fusion:
`[[7,7],null,null]` either way.
- At `:277` the input is one row, so every percentile over `(a + b) + c` is
`1.0` whether columns 3-4 come from two digests or one fused array:
`[[0.0,0.0],0.0,1.0,1.0]` either way.
The suite already has the discriminating pattern in two places —
`crossDistinctCollision` compares against a dynamically collected
fusion-disabled baseline, and `preserve canonically colliding parameters`
asserts `expectedDigests`. Either add an `assertPercentileDigestCount` to these
two, or choose an input where the guard actually changes the answer.
##########
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(
Review Comment:
**Finding 17.** `PhysicalCompatibilityKey` deliberately omits
`percentageExpression`, and the size-1 check at `:177` then rejects fusion
whenever two `CompatibilityKey`s collapse onto one physical key. But what that
guard has to preserve is `PhysicalAggregation`'s dedup of *canonically equal
`AggregateExpression`s*, and canonical equality includes the percentage.
Without it the guard fires when there is no dedup pair to protect.
`Add` is commutative under canonicalization, so:
```sql
SELECT percentile_approx(a + b, 0.5D), percentile_approx(a + b, 0.9D),
percentile_approx(b + a, 0.25D), percentile_approx(b + a, 0.75D) FROM
t
```
produces two logical keys (`a+b` and `b+a`) on one physical key, so **both**
groups are rejected. Measured with the flag on: four scalar
`ApproximatePercentile`s survive where two array-valued ones would do, and the
rows are identical with the flag on and off — a pure lost optimization. There
is also nothing to preserve here: all four percentages differ, so no two of the
four aggregates are canonically equal and base has no dedup either.
The physical key is already built per expression at `:174`, so this is a
field plus one argument:
```suggestion
private case class PhysicalCompatibilityKey(
child: Expression,
percentage: Expression,
accuracy: Expression,
mode: AggregateMode,
isDistinct: Boolean,
filter: Option[Expression])
```
passing `percentile.percentageExpression.canonicalized` from
`physicalCompatibilityKey`.
I applied that change locally and ran it. The query above goes from four
scalar percentiles to two fused array-valued ones and returns the identical
rows (`[7.0, 11.0, 3.0, 11.0]` either way), and both suites stay green:
`CombineApproximatePercentilesSuite` (9 tests) and
`ApproximatePercentileQuerySuite` (33 tests). So the existing guards still
block everything they should — the colliding pairs in `require structural
equality for inputs and filters` and in the first `checkAnswer` of `do not fuse
canonical input or filter collisions` *share* percentages `0.5`/`0.9`, so they
continue to land on one physical key.
--
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]
