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/24 18:00:45 UTC
(spark) branch branch-4.x updated: [SPARK-58095][SQL] approx_top_k_combine throws scala.MatchError on empty input
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 5d45b9041205 [SPARK-58095][SQL] approx_top_k_combine throws scala.MatchError on empty input
5d45b9041205 is described below
commit 5d45b90412057046907f4431a92f52dade630009
Author: Eric Yang <ji...@gmail.com>
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 <ji...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 20814f1a5b27e6bf759932d9526d54e53d18224e)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org