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 20814f1a5b27 [SPARK-58095][SQL] approx_top_k_combine throws
scala.MatchError on empty input
20814f1a5b27 is described below
commit 20814f1a5b27e6bf759932d9526d54e53d18224e
Author: Eric Yang <[email protected]>
AuthorDate: Sat Jul 25 02:00:13 2026 +0800
[SPARK-58095][SQL] approx_top_k_combine throws scala.MatchError on empty
input
### What changes were proposed in this pull request?
Fix the MatchError on empty input with changes below:
- Fall back to the statically-known item data type carried by the state
struct (`uncheckedItemDataType`) when no input sketch resolved it, in both
`eval` and `serialize`.
- When combine size is unspecified and no input was consumed,
`maxItemsTracked` is still the internal `VOID` sentinel (-1); resolve it to the
default in `eval` only (a partial buffer must keep `VOID` for the final merge),
so the empty output sketch carries a valid size that downstream functions such
as `approx_top_k_estimate` accept.
### Why are the changes needed?
The query aborts with an internal error instead of returning an empty
result, unlike the sibling functions (`approx_top_k`,
`approx_top_k_accumulate`, `approx_top_k_estimate`) which handle empty input
correctly:
```sql
SELECT approx_top_k_combine(sketch, 5)
FROM (SELECT approx_top_k_accumulate(expr) AS sketch
FROM VALUES (1), (2) AS t(expr))
WHERE false;
-- before: org.apache.spark.SparkException ... scala.MatchError: null
-- after: returns a single empty sketch (estimates to [])
```
### Does this PR introduce _any_ user-facing change?
Yes. Fixed an query failing issue when the input is empty.
### How was this patch tested?
Added UT case
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Closes #57289 from jiwen624/SPARK-58095.
Authored-by: Eric Yang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../aggregate/ApproxTopKAggregates.scala | 20 +++++++++--
.../org/apache/spark/sql/ApproxTopKSuite.scala | 39 ++++++++++++++++++++++
2 files changed, 57 insertions(+), 2 deletions(-)
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 bdd3f50fc1ff..b646f6e6e59d 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
@@ -672,7 +672,7 @@ class CombineInternal[T](
def getMaxItemsTracked: Int = maxItemsTracked
def updateMaxItemsTracked(combineSizeSpecified: Boolean, newMaxItemsTracked:
Int): Unit = {
- if (!combineSizeSpecified) {
+ if (!combineSizeSpecified && newMaxItemsTracked !=
ApproxTopK.VOID_MAX_ITEMS_TRACKED) {
// check size
if (this.maxItemsTracked == ApproxTopK.VOID_MAX_ITEMS_TRACKED) {
// If buffer's maxItemsTracked VOID_MAX_ITEMS_TRACKED, it means the
buffer is a placeholder
@@ -905,10 +905,25 @@ case class ApproxTopKCombine(
buffer
}
+ private def resolveEmptyBufferItemDataType(buffer: CombineInternal[Any]):
Unit = {
+ if (buffer.getItemDataType == null) {
+ buffer.updateItemDataType(uncheckedItemDataType)
+ }
+ }
+
override def eval(buffer: CombineInternal[Any]): Any = {
+ resolveEmptyBufferItemDataType(buffer)
val sketchBytes = buffer.getSketchWithNullCount
.serialize(ApproxTopK.genSketchSerDe(buffer.getItemDataType))
- val maxItemsTracked = buffer.getMaxItemsTracked
+ // A buffer that saw no input (e.g. a size-unspecified combine over empty
input) still carries
+ // the VOID sentinel here. eval must emit a concrete size: the downstream
estimate rejects any
+ // non-positive maxItemsTracked, so fall back to
DEFAULT_MAX_ITEMS_TRACKED. This differs from
+ // serialize, which keeps VOID so an empty partial stays neutral for the
final merge's size
+ // check.
+ val maxItemsTracked = buffer.getMaxItemsTracked match {
+ case ApproxTopK.VOID_MAX_ITEMS_TRACKED =>
ApproxTopK.DEFAULT_MAX_ITEMS_TRACKED
+ case other => other
+ }
val itemDataTypeDDL = ApproxTopK.dataTypeToDDL(buffer.getItemDataType)
InternalRow.apply(
sketchBytes,
@@ -918,6 +933,7 @@ case class ApproxTopKCombine(
}
override def serialize(buffer: CombineInternal[Any]): Array[Byte] = {
+ resolveEmptyBufferItemDataType(buffer)
buffer.serialize()
}
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 62ae39a3616c..0832262e64ff 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
@@ -631,6 +631,45 @@ class ApproxTopKSuite extends SharedSparkSession {
}
}
+ test("SPARK-58095: approx_top_k_combine on empty input returns an empty
sketch") {
+ // The inner aggregate produces a single sketch row, which WHERE false
filters out, so
+ // approx_top_k_combine sees zero input rows. It must return a single
empty sketch that
+ // estimates to an empty top-k result, instead of failing the query with a
MatchError,
+ // whether or not the combine size is specified. When the size is
specified, the empty
+ // result carries that size; when it is unspecified, the empty buffer
never learns a size
+ // from input and eval falls back to DEFAULT_MAX_ITEMS_TRACKED (10000).
+ Seq(("approx_top_k_combine(sketch, 5)", 5),
("approx_top_k_combine(sketch)", 10000))
+ .foreach { case (combine, expectedMaxItemsTracked) =>
+ withView("combined") {
+ sql(
+ s"SELECT $combine AS com " +
+ "FROM (SELECT approx_top_k_accumulate(expr) AS sketch " +
+ " FROM VALUES (1), (2) AS t(expr)) " +
+ "WHERE false")
+ .createOrReplaceTempView("combined")
+ checkAnswer(sql("SELECT approx_top_k_estimate(com) FROM combined"),
Row(Seq.empty))
+ checkAnswer(
+ sql("SELECT com.maxItemsTracked FROM combined"),
Row(expectedMaxItemsTracked))
+ }
+ }
+ }
+
+ test("SPARK-58095: approx_top_k_combine over mixed empty and non-empty
partitions") {
+ Seq("approx_top_k_combine(sketch, 5)",
"approx_top_k_combine(sketch)").foreach { combine =>
+ withView("combined") {
+ sql(
+ s"SELECT $combine AS com " +
+ "FROM (SELECT /*+ REPARTITION(4) */ sketch " +
+ " FROM (SELECT approx_top_k_accumulate(expr) AS sketch " +
+ " FROM VALUES (1), (1), (2) AS t(expr)))")
+ .createOrReplaceTempView("combined")
+ checkAnswer(
+ sql("SELECT approx_top_k_estimate(com) FROM combined"),
+ Row(Seq(Row(1, 2), Row(2, 1))))
+ }
+ }
+ }
+
test("SPARK-52798: same type, different size, unspecified combine size -
fail") {
withView("accumulation1", "accumulation2", "unioned") {
setupMixedSizeAccumulations(10, 20)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
