You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/07/23 14:23:19 UTC
(spark) branch master updated: [SPARK-58164][SQL] approx_top_k_estimate crashes or returns wrong results on sketches whose item type was widened by type coercion
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 490f2bbf1d78 [SPARK-58164][SQL] approx_top_k_estimate crashes or returns wrong results on sketches whose item type was widened by type coercion
490f2bbf1d78 is described below
commit 490f2bbf1d783c9fd4531d0dcbbedbcfc795b3a3
Author: Eric Yang <ji...@gmail.com>
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 <ji...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org