This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 350fa45aa4db [SPARK-58164][SQL] approx_top_k_estimate crashes or
returns wrong results on sketches whose item type was widened by type coercion
350fa45aa4db is described below
commit 350fa45aa4db9757a6304b499b96db8d0022814e
Author: Eric Yang <[email protected]>
AuthorDate: Thu Jul 23 22:23:02 2026 +0800
[SPARK-58164][SQL] approx_top_k_estimate crashes or returns wrong results
on sketches whose item type was widened by type coercion
### What changes were proposed in this pull request?
`approx_top_k_estimate` picked its sketch deserialization SerDe from the
static item type in the sketch-state struct (field 2). A `UNION ALL` of two
sketch columns with coercible item types (e.g. INT + BIGINT) widens that static
type while the serialized bytes and the per-row DDL (field 3) keep the original
type. Estimation then decoded the bytes with the wrong SerDe.
This PR makes `estimate` deserialize using the true item type recorded
per-row in the state DDL (recovering collation from the static type, exactly as
`combine` already does), then widens each decoded item to the declared output
type via a lossless cast so results still match the statically-declared schema.
### Why are the changes needed?
A query like:
```sql
SELECT approx_top_k_estimate(approx_top_k_combine(sketch, 5))
FROM (SELECT approx_top_k_accumulate(c) AS sketch, 'int' AS tag
FROM VALUES (1), (1), (2) AS t(c)
UNION ALL
SELECT approx_top_k_accumulate(c) AS sketch, 'long' AS tag
FROM VALUES (CAST(9 AS BIGINT)) AS t(c))
WHERE tag = 'int';
```
fails at runtime with SketchesArgumentException: Bounds Violation, e.g.,
```Java
26/07/16 22:29:33 ERROR Executor: Exception in task 0.0 in stage 2.0 (TID 3)
org.apache.datasketches.common.SketchesArgumentException: Bounds Violation:
reqOffset: 0, reqLength: 16, (reqOff + reqLen): 16, allocSize: 10
at org.apache.datasketches.common.Util.checkBounds(Util.java:659)
```
Or, depending on the widening it can also throw ClassCastException or, when
byte sizes
happen to match (e.g. BIGINT→DOUBLE), silently return wrong results.
### Does this PR introduce _any_ user-facing change?
Yes. A query that may crash or return wrong results now gets fixed.
### How was this patch tested?
Added UT.
### Was this patch authored or co-authored using generative AI tooling?
Yes
Closes #57327 from jiwen624/SPARK-58164.
Authored-by: Eric Yang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 490f2bbf1d783c9fd4531d0dcbbedbcfc795b3a3)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../expressions/ApproxTopKExpressions.scala | 7 ++-
.../aggregate/ApproxTopKAggregates.scala | 18 +++++--
.../org/apache/spark/sql/ApproxTopKSuite.scala | 63 ++++++++++++++++++++++
3 files changed, 82 insertions(+), 6 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApproxTopKExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApproxTopKExpressions.scala
index 629b73543377..8d4f0a18d570 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApproxTopKExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApproxTopKExpressions.scala
@@ -108,10 +108,13 @@ case class ApproxTopKEstimate(state: Expression, k:
Expression)
val kVal = kEval.asInstanceOf[Int]
ApproxTopK.checkK(kVal)
ApproxTopK.checkMaxItemsTracked(maxItemsTrackedVal, kVal)
+ val sketchItemType = ApproxTopK.withCollationOf(
+
ApproxTopK.DDLToDataType(stateEval.asInstanceOf[InternalRow].getUTF8String(3).toString),
+ itemDataType)
val approxTopKAggregateBuffer = ApproxTopKAggregateBuffer.deserialize(
dataSketchBytes,
- ApproxTopK.genSketchSerDe(itemDataType))
- approxTopKAggregateBuffer.eval(kVal, itemDataType)
+ ApproxTopK.genSketchSerDe(sketchItemType))
+ approxTopKAggregateBuffer.eval(kVal, sketchItemType, itemDataType)
}
override protected def withNewChildrenInternal(newState: Expression, newK:
Expression)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
index ac1d57803905..bdd3f50fc1ff 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
@@ -28,10 +28,11 @@ import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry,
TypeCheckResult}
import
org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure,
TypeCheckSuccess}
-import org.apache.spark.sql.catalyst.expressions.{ArrayOfCollatedStringsSerDe,
ArrayOfDecimalsSerDe, CollatedString, Expression, ExpressionDescription,
ImplicitCastInputTypes, Literal}
+import org.apache.spark.sql.catalyst.expressions.{ArrayOfCollatedStringsSerDe,
ArrayOfDecimalsSerDe, Cast, CollatedString, Expression, ExpressionDescription,
ImplicitCastInputTypes, Literal}
import org.apache.spark.sql.catalyst.trees.{BinaryLike, TernaryLike}
import org.apache.spark.sql.catalyst.util.{CollationFactory, GenericArrayData,
UnsafeRowUtils}
import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -169,7 +170,7 @@ case class ApproxTopK(
buffer.merge(input)
override def eval(buffer: ApproxTopKAggregateBuffer[Any]): GenericArrayData =
- buffer.eval(kVal, itemDataType)
+ buffer.eval(kVal, itemDataType, itemDataType)
override def serialize(buffer: ApproxTopKAggregateBuffer[Any]): Array[Byte] =
buffer.serialize(ApproxTopK.genSketchSerDe(itemDataType))
@@ -355,6 +356,14 @@ object ApproxTopK {
TypeCheckSuccess
}
}
+
+ def widenItemValue(item: Any, sketchType: DataType, outputType: DataType):
Any =
+ (sketchType, outputType) match {
+ case _ if sketchType == outputType => item
+ case (_: StringType, _: StringType) => item
+ case _ =>
+ Cast(Literal(item, sketchType), outputType,
Some(SQLConf.get.sessionLocalTimeZone)).eval()
+ }
}
/**
@@ -432,7 +441,7 @@ class ApproxTopKAggregateBuffer[T](val sketch:
ItemsSketch[T], private var nullC
* Evaluate the buffer and return top K items (including null) with their
estimated frequency.
* The result is sorted by frequency in descending order.
*/
- def eval(k: Int, itemDataType: DataType): GenericArrayData = {
+ def eval(k: Int, sketchItemType: DataType, outputItemType: DataType):
GenericArrayData = {
// frequent items from sketch
val frequentItems = sketch.getFrequentItems(ErrorType.NO_FALSE_POSITIVES)
// total number of frequent items (including null, if any)
@@ -462,7 +471,7 @@ class ApproxTopKAggregateBuffer[T](val sketch:
ItemsSketch[T], private var nullC
(null, nullCount.toLong)
} else {
// insert frequent item into result
- val item: Any = itemDataType match {
+ val rawItem: Any = sketchItemType match {
case _: BooleanType | _: ByteType | _: ShortType | _: IntegerType |
_: LongType | _: FloatType | _: DoubleType | _: DecimalType |
_: DateType | _: TimestampType | _: TimestampNTZType | _:
TimeType =>
@@ -475,6 +484,7 @@ class ApproxTopKAggregateBuffer[T](val sketch:
ItemsSketch[T], private var nullC
s"Unexpected sketch item type for a string column:
${other.getClass.getName}")
}
}
+ val item = ApproxTopK.widenItemValue(rawItem, sketchItemType,
outputItemType)
fiIndex += 1 // move to next frequent item
(item, itemEstimate)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
index 98aba80682e6..62ae39a3616c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
@@ -707,6 +707,69 @@ class ApproxTopKSuite extends SharedSparkSession {
}
}
+ test("SPARK-58164: estimate over combine of a coercion-widened sketch (INT
-> BIGINT)") {
+ val res = sql(
+ """SELECT approx_top_k_estimate(approx_top_k_combine(sketch, 5))
+ |FROM (SELECT approx_top_k_accumulate(c) AS sketch, 'int' AS tag
+ | FROM VALUES (1), (1), (2) AS t(c)
+ | UNION ALL
+ | SELECT approx_top_k_accumulate(c) AS sketch, 'long' AS tag
+ | FROM VALUES (CAST(9 AS BIGINT)) AS t(c))
+ |WHERE tag = 'int'""".stripMargin)
+ checkAnswer(res, Row(Seq(Row(1L, 2L), Row(2L, 1L))))
+ }
+
+ test("SPARK-58164: estimate over a coercion-widened accumulate sketch (INT
-> BIGINT)") {
+ val res = sql(
+ """SELECT approx_top_k_estimate(sketch, 5)
+ |FROM (SELECT approx_top_k_accumulate(c) AS sketch, 'int' AS tag
+ | FROM VALUES (1), (1), (2) AS t(c)
+ | UNION ALL
+ | SELECT approx_top_k_accumulate(c) AS sketch, 'long' AS tag
+ | FROM VALUES (CAST(9 AS BIGINT)) AS t(c))
+ |WHERE tag = 'int'""".stripMargin)
+ checkAnswer(res, Row(Seq(Row(1L, 2L), Row(2L, 1L))))
+ }
+
+ test("SPARK-58164: estimate over combine of a coercion-widened sketch across
SerDe " +
+ "families (INT -> DOUBLE)") {
+ val res = sql(
+ """SELECT approx_top_k_estimate(approx_top_k_combine(sketch, 5))
+ |FROM (SELECT approx_top_k_accumulate(c) AS sketch, 'int' AS tag
+ | FROM VALUES (1), (1), (2) AS t(c)
+ | UNION ALL
+ | SELECT approx_top_k_accumulate(c) AS sketch, 'double' AS tag
+ | FROM VALUES (CAST(9 AS DOUBLE)) AS t(c))
+ |WHERE tag = 'int'""".stripMargin)
+ checkAnswer(res, Row(Seq(Row(1.0d, 2L), Row(2.0d, 1L))))
+ }
+
+ test("SPARK-58164: estimate over combine of a coercion-widened sketch within
a SerDe " +
+ "family (TINYINT -> INT)") {
+ val res = sql(
+ """SELECT approx_top_k_estimate(approx_top_k_combine(sketch, 5))
+ |FROM (SELECT approx_top_k_accumulate(c) AS sketch, 'byte' AS tag
+ | FROM VALUES (CAST(1 AS BYTE)), (CAST(1 AS BYTE)), (CAST(2 AS
BYTE)) AS t(c)
+ | UNION ALL
+ | SELECT approx_top_k_accumulate(c) AS sketch, 'int' AS tag
+ | FROM VALUES (9) AS t(c))
+ |WHERE tag = 'byte'""".stripMargin)
+ checkAnswer(res, Row(Seq(Row(1, 2L), Row(2, 1L))))
+ }
+
+ test("SPARK-58164: estimate widens each row independently across a coerced
union") {
+ val res = sql(
+ """SELECT approx_top_k_estimate(sketch, 5)
+ |FROM (SELECT approx_top_k_accumulate(c) AS sketch
+ | FROM VALUES (1), (1), (2) AS t(c)
+ | UNION ALL
+ | SELECT approx_top_k_accumulate(c) AS sketch
+ | FROM VALUES (CAST(9 AS BIGINT)), (CAST(9 AS BIGINT)) AS
t(c))""".stripMargin)
+ checkAnswer(res, Seq(
+ Row(Seq(Row(1L, 2L), Row(2L, 1L))),
+ Row(Seq(Row(9L, 2L)))))
+ }
+
// enumerate all combinations of number and datetime types
gridTest("SPARK-52798: number vs datetime - fail on UNION")(
for {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
