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/10/22 11:42:10 UTC
(spark) branch branch-4.x updated: [SPARK-59818][SQL] Handle a NULL sketch state in approx_top_k_estimate and approx_top_k_combine
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 f6b98369cdca [SPARK-59818][SQL] Handle a NULL sketch state in approx_top_k_estimate and approx_top_k_combine
f6b98369cdca is described below
commit f6b98369cdca394ba5d0ffbc5e93495a2bac3364
Author: Jubin Soni <ju...@gmail.com>
AuthorDate: Thu Oct 22 11:41:37 2026 +0000
[SPARK-59818][SQL] Handle a NULL sketch state in approx_top_k_estimate and approx_top_k_combine
### What changes were proposed in this pull request?
Fixes [SPARK-59818](https://issues.apache.org/jira/browse/SPARK-59818) — makes `approx_top_k_estimate` and `approx_top_k_combine` handle a NULL sketch state instead of throwing `NullPointerException`. Two related gaps are addressed:
**`approx_top_k_estimate` dereferenced the state unconditionally.** `ApproxTopKEstimate` overrides `BinaryExpression.eval` directly rather than implementing `nullSafeEval`, so the inherited null handling is bypassed, and `eval` calls `stateEval.asInstanceOf[InternalRow].getBinary(0)` with no null check. It also declared `override def nullable: Boolean = false` even though the result is NULL whenever the state is, so the reported output schema was wrong and the optimizer reasoned incor [...]
**`approx_top_k_combine` had the same unguarded dereference in `update`.** `ApproxTopKCombine.update` calls `inputState.getBinary(0)` directly, so a NULL sketch fails on the executors and aborts the running job, rather than being ignored the way aggregates normally ignore NULL inputs.
Analysis does not prevent either case: `checkStateFieldAndType` validates only the *schema* of the state struct, not its nullness, so a correctly-typed nullable struct column passes the analyzer and then fails at optimization or execution time. A NULL sketch arises naturally from an outer join, a `UNION` with a missing side, or a partially-populated column.
Both are surfaced and locked down by new tests.
**Brief change log**
- `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ApproxTopKExpressions.scala`: `ApproxTopKEstimate.eval` returns `null` when the state evaluates to `null`, and `nullable` is now `state.nullable` instead of a hardcoded `false`, so the reported schema matches the values actually produced.
- `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala`: `ApproxTopKCombine.update` returns the buffer untouched for a NULL sketch, so combine ignores NULL inputs like any other aggregate.
- `sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala`: added four tests covering the foldable NULL state, the non-foldable NULL state, the reported nullability, and combine skipping NULL sketches.
### Why are the changes needed?
Both paths fail on a valid query.
`approx_top_k_estimate` raises the NPE inside `ConstantFolding`, which reports it to the user as an internal error:
```sql
SELECT approx_top_k_estimate(s, 5) FROM (
SELECT CAST(NULL AS STRUCT<sketch: BINARY, maxItemsTracked: INT,
itemDataType: INT, itemDataTypeDDL: STRING>) AS s
);
```
```
[INTERNAL_ERROR] The Spark SQL phase optimization failed with an internal error.
You hit a bug in Spark or the Spark plugins you use. SQLSTATE: XX000
Caused by: java.lang.NullPointerException: Cannot invoke
"org.apache.spark.sql.catalyst.InternalRow.getBinary(int)" because "stateEval" is null
at ApproxTopKEstimate.eval(ApproxTopKExpressions.scala:113)
at org.apache.spark.sql.catalyst.optimizer.ConstantFolding$.tryFold(expressions.scala:65)
```
`approx_top_k_combine` fails on the executors and aborts the job:
```sql
SELECT approx_top_k_estimate(approx_top_k_combine(s), 5) FROM (
SELECT CAST(NULL AS STRUCT<sketch: BINARY, maxItemsTracked: INT,
itemDataType: INT, itemDataTypeDDL: STRING>) AS s
);
```
```
org.apache.spark.SparkException: Job aborted due to stage failure:
Task 0 in stage 0.0 failed 1 times ...
Caused by: java.lang.NullPointerException
at ApproxTopKCombine.update(ApproxTopKAggregates.scala:908)
at TypedImperativeAggregate.update(interfaces.scala:583)
```
The new behaviour follows ordinary SQL semantics for a nullable input column: `approx_top_k_estimate(NULL, k)` returns NULL, and `approx_top_k_combine` skips NULL sketches.
This is distinct from the existing `APPROX_TOP_K_NULL_ARG` error, which is raised for NULL *foldable configuration* arguments (`k`, `maxItemsTracked`). That check, `ApproxTopK.checkExpressionNotNull`, calls `expr.eval()` with no input row, so it applies only to literals and is not applicable to a per-row data column.
It is also distinct from earlier NULL-related work on these functions. SPARK-53960 added handling for NULLs *within* the accumulated data (counting them as items) and did not touch the NULL-state path or the `nullable` declaration; SPARK-58095 concerns *empty* input (zero rows) in combine's `eval`/`serialize`, which is a different code path and trigger.
### Does this PR introduce _any_ user-facing change?
Yes, in three ways, all aligning the functions with ordinary SQL null semantics:
1. `approx_top_k_estimate` on a NULL sketch state returns NULL instead of failing the query with `[INTERNAL_ERROR]`.
2. `approx_top_k_combine` skips NULL sketches instead of aborting the job with a `NullPointerException`.
3. `approx_top_k_estimate` now reports itself as nullable when its state argument is nullable. Previously it always reported `nullable = false`, which was incorrect for a nullable state. The reported schema is unchanged for a non-nullable state, which is the common case — the `sql-expression-schema.md` golden file is unaffected, as its example applies the function to an aggregate result.
The existing `APPROX_TOP_K_NON_POSITIVE_ARG` validation is unaffected: `k` is evaluated and validated before the NULL-state short-circuit, so a non-positive `k` is rejected whether or not the sketch is NULL.
These are changes compared to released versions: `approx_top_k` is annotated `since = "4.1.0"`.
### How was this patch tested?
Five new tests in `ApproxTopKSuite`, covering both functions:
- `SPARK-59818: estimate of a foldable NULL state returns NULL`
- `SPARK-59818: estimate of a non-foldable NULL state returns NULL`
- `SPARK-59818: estimate nullability follows the nullability of its state`
- `SPARK-59818: estimate rejects a non-positive k even for a NULL state`
- `SPARK-59818: combine skips NULL sketches`
The foldable and non-foldable state are covered separately because only the former is reached by `ConstantFolding`; the non-foldable case is built with a `UNION ALL` so the state stays a per-row column.
Each test was confirmed to fail without the corresponding fix:
- Reverting the two NULL-state commits reproduces the reported failures — `Cannot invoke "InternalRow.getBinary(int)" because "stateEval" is null` for the estimate tests and `... because "inputState" is null`, aborting the stage, for the combine test.
- Restoring the previous evaluation order, where the NULL-state return ran before `ApproxTopK.checkK`, makes the non-positive-`k` test fail with "Expected exception org.apache.spark.SparkRuntimeException to be thrown, but no exception was thrown".
- Hardcoding `nullable` to `true` makes the nullability test fail, confirming it pins both branches of the contract rather than only the nullable one.
The full `ApproxTopKSuite` passes (201 tests: 196 existing plus the 5 new). `ExpressionInfoSuite`, `NullabilitySuite`, `CanonicalizeSuite`, and `ExpressionsSchemaSuite` also pass, confirming the `nullable` change leaves the expression schema golden file unchanged.
Run:
```
build/sbt 'sql/testOnly org.apache.spark.sql.ApproxTopKSuite'
build/sbt 'catalyst/testOnly *ExpressionInfoSuite *NullabilitySuite *CanonicalizeSuite'
build/sbt 'sql/testOnly *ExpressionsSchemaSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes - Claude Code was used as a pair-programming assistant. All changes were reviewed and verified by the author. Generated-by: Claude Opus 5
Closes #59094 from jubins/j-spark-59818-fix-null-sketch-state.
Authored-by: Jubin Soni <ju...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 329c057478b5c30fcdee14e1ee3524bffc0d6436)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../expressions/ApproxTopKExpressions.scala | 16 ++-
.../aggregate/ApproxTopKAggregates.scala | 9 +-
.../org/apache/spark/sql/ApproxTopKSuite.scala | 152 +++++++++++++++++++++
3 files changed, 172 insertions(+), 5 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 fc647c860208..b028d5118d07 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,12 +108,16 @@ case class ApproxTopKEstimate(state: Expression, k: Expression)
// null check
ApproxTopK.checkExpressionNotNull(k, "k")
// eval
+ val kVal = right.eval(input).asInstanceOf[Int]
+ // Validate `k` before short-circuiting on a NULL state, so that an invalid `k` is
+ // rejected regardless of whether the sketch happens to be NULL.
+ ApproxTopK.checkK(kVal)
val stateEval = left.eval(input)
- val kEval = right.eval(input)
+ if (stateEval == null) {
+ return null
+ }
val dataSketchBytes = stateEval.asInstanceOf[InternalRow].getBinary(0)
val maxItemsTrackedVal = stateEval.asInstanceOf[InternalRow].getInt(1)
- val kVal = kEval.asInstanceOf[Int]
- ApproxTopK.checkK(kVal)
ApproxTopK.checkMaxItemsTracked(maxItemsTrackedVal, kVal)
val sketchItemType = ApproxTopK.withCollationOf(
ApproxTopK.DDLToDataType(stateEval.asInstanceOf[InternalRow].getUTF8String(3).toString),
@@ -127,7 +131,11 @@ case class ApproxTopKEstimate(state: Expression, k: Expression)
override protected def withNewChildrenInternal(newState: Expression, newK: Expression)
: Expression = copy(state = newState, k = newK)
- override def nullable: Boolean = false
+ // The sketch state is an ordinary nullable input column: `approx_top_k_estimate(NULL, k)`
+ // returns NULL rather than failing, so the result is nullable exactly when the state is.
+ // A sketch produced by `approx_top_k_accumulate` is non-nullable, so the common case is
+ // unchanged.
+ override def nullable: Boolean = state.nullable
override def prettyName: String =
getTagValue(FunctionRegistry.FUNC_ALIAS).getOrElse("approx_top_k_estimate")
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 8ed6590793da..1d0df5a2b120 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
@@ -905,6 +905,10 @@ case class ApproxTopKCombine(
*/
override def update(buffer: CombineInternal[Any], input: InternalRow): CombineInternal[Any] = {
val inputState = state.eval(input).asInstanceOf[InternalRow]
+ if (inputState == null) {
+ // A NULL sketch carries no items, so it contributes nothing to the combined sketch.
+ return buffer
+ }
val inputSketchBytes = inputState.getBinary(0)
val inputMaxItemsTracked = inputState.getInt(1)
val inputItemDataTypeDDL = inputState.getUTF8String(3).toString
@@ -961,7 +965,10 @@ case class ApproxTopKCombine(
}
override def serialize(buffer: CombineInternal[Any]): Array[Byte] = {
- resolveEmptyBufferItemDataType(buffer)
+ // Keep a partial buffer that saw no sketch (empty input or only NULL sketches) untyped, so it
+ // stays neutral in the final merge. Resolving it to uncheckedItemDataType here would give it
+ // the coercion-widened type, which need not match the real sketches' item type (SPARK-58164).
+ // eval resolves the type of a buffer that is still untyped at the end.
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 dbccbcce7393..ad8447220e12 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
@@ -549,6 +549,69 @@ class ApproxTopKSuite extends SharedSparkSession {
)
}
+
+ private val nullSketchState =
+ """CAST(NULL AS STRUCT<sketch: BINARY, maxItemsTracked: INT,
+ | itemDataType: INT, itemDataTypeDDL: STRING>)""".stripMargin
+
+ test("SPARK-59818: estimate of a foldable NULL state returns NULL") {
+ checkAnswer(sql(s"SELECT approx_top_k_estimate($nullSketchState, 5)"), Row(null))
+ }
+
+ test("SPARK-59818: estimate of a non-foldable NULL state returns NULL") {
+ withTempView("estimate_null_state") {
+ sql(
+ s"""SELECT approx_top_k_accumulate(expr) AS state
+ |FROM VALUES 0, 1, 1 AS tab(expr)
+ |UNION ALL
+ |SELECT $nullSketchState AS state""".stripMargin)
+ .createOrReplaceTempView("estimate_null_state")
+ val res = sql("SELECT approx_top_k_estimate(state, 2) FROM estimate_null_state")
+ checkAnswer(res, Seq(Row(Seq(Row(1, 2), Row(0, 1))), Row(null)))
+ }
+ }
+
+ test("SPARK-59818: estimate nullability follows the nullability of its state") {
+ withTempView("estimate_nullable_state") {
+ sql(s"SELECT $nullSketchState AS state")
+ .createOrReplaceTempView("estimate_nullable_state")
+ val nullableRes =
+ sql("SELECT approx_top_k_estimate(state, 2) FROM estimate_nullable_state")
+ assert(nullableRes.schema.fields.head.nullable)
+ }
+ // The common case: a sketch produced by approx_top_k_accumulate is non-nullable, so the
+ // estimate over it must stay non-nullable.
+ val nonNullableRes = sql(
+ """SELECT approx_top_k_estimate(approx_top_k_accumulate(expr), 2)
+ |FROM VALUES 0, 1, 1 AS tab(expr)""".stripMargin)
+ assert(!nonNullableRes.schema.fields.head.nullable)
+ }
+
+ test("SPARK-59818: estimate rejects a non-positive k even for a NULL state") {
+ Seq(0, -1).foreach { invalidK =>
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql(s"SELECT approx_top_k_estimate($nullSketchState, $invalidK)").collect()
+ },
+ condition = "APPROX_TOP_K_NON_POSITIVE_ARG",
+ parameters = Map("argName" -> "`k`", "argValue" -> invalidK.toString)
+ )
+ }
+ // Every row carries a NULL state, so no other row can raise the error on its own: the
+ // assertion holds only if the row-evaluated NULL state itself validates `k`.
+ withTempView("estimate_invalid_k") {
+ spark.range(2).selectExpr(s"$nullSketchState AS state")
+ .createOrReplaceTempView("estimate_invalid_k")
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("SELECT approx_top_k_estimate(state, 0) FROM estimate_invalid_k").collect()
+ },
+ condition = "APPROX_TOP_K_NON_POSITIVE_ARG",
+ parameters = Map("argName" -> "`k`", "argValue" -> "0")
+ )
+ }
+ }
+
/////////////////////////////////
// approx_top_k_combine
/////////////////////////////////
@@ -1296,4 +1359,93 @@ class ApproxTopKSuite extends SharedSparkSession {
checkAnswer(est, Row(Seq(Row(null, 5))))
}
}
+
+ test("SPARK-59818: combine over NULL sketches is independent of partition placement") {
+ withTempView("combine_null_partitions") {
+ // Many NULL rows alongside a single real sketch, so that some partial aggregates see
+ // only NULL rows and have to serialize a buffer no update ever populated.
+ sql(
+ """SELECT approx_top_k_accumulate(CAST(expr AS BIGINT)) AS state
+ |FROM VALUES 0, 1, 1 AS tab(expr)""".stripMargin)
+ .union(
+ spark.range(200).selectExpr(
+ """CAST(NULL AS STRUCT<sketch: BINARY, maxItemsTracked: INT,
+ | itemDataType: BIGINT, itemDataTypeDDL: STRING>) AS state""".stripMargin))
+ .createOrReplaceTempView("combine_null_partitions")
+ Seq("1", "8", "64").foreach { partitions =>
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> partitions) {
+ val res = sql(
+ """SELECT approx_top_k_estimate(approx_top_k_combine(state), 2)
+ |FROM combine_null_partitions""".stripMargin)
+ checkAnswer(res, Row(Seq(Row(1L, 2), Row(0L, 1))))
+ }
+ }
+ }
+ }
+
+ test("SPARK-59818: combine of an INT sketch with widened BIGINT NULL states is independent " +
+ "of partition placement") {
+ withTempView("combine_widened_null_partitions") {
+ // The UNION widens the state's item type to BIGINT, while the real sketch still records INT
+ // (SPARK-58164). Partial aggregates that see only NULL rows must stay type-neutral instead
+ // of taking the widened type, which would not match the real sketch at the final merge.
+ sql("SELECT approx_top_k_accumulate(expr) AS state FROM VALUES 0, 1, 1 AS tab(expr)")
+ .union(
+ spark.range(200).selectExpr(
+ """CAST(NULL AS STRUCT<sketch: BINARY, maxItemsTracked: INT,
+ | itemDataType: BIGINT, itemDataTypeDDL: STRING>) AS state""".stripMargin))
+ .createOrReplaceTempView("combine_widened_null_partitions")
+ Seq("1", "8", "64").foreach { partitions =>
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> partitions) {
+ val res = sql(
+ """SELECT approx_top_k_estimate(approx_top_k_combine(state), 2)
+ |FROM combine_widened_null_partitions""".stripMargin)
+ checkAnswer(res, Row(Seq(Row(1L, 2L), Row(0L, 1L))))
+ }
+ }
+ }
+ }
+
+ test("SPARK-59818: combine over only NULL sketches keeps usable state metadata") {
+ withTempView("combine_all_null") {
+ spark.range(4).selectExpr(s"$nullSketchState AS state")
+ .createOrReplaceTempView("combine_all_null")
+ val combined = sql("SELECT approx_top_k_combine(state) AS state FROM combine_all_null")
+ // The combined state must still carry a concrete item type, so that it can be estimated.
+ assert(combined.collect().head.getStruct(0).getString(3) === "item INT NOT NULL")
+ checkAnswer(
+ sql("""SELECT approx_top_k_estimate(approx_top_k_combine(state), 3)
+ |FROM combine_all_null""".stripMargin),
+ Row(Seq.empty))
+ }
+ }
+
+ test("SPARK-59818: combine still rejects sketches of different item types") {
+ withTempView("combine_int", "combine_string") {
+ sql("SELECT approx_top_k_accumulate(expr) AS state FROM VALUES 0, 1 AS tab(expr)")
+ .createOrReplaceTempView("combine_int")
+ sql("SELECT approx_top_k_accumulate(expr) AS state FROM VALUES 'x', 'y' AS tab(expr)")
+ .createOrReplaceTempView("combine_string")
+ intercept[SparkRuntimeException] {
+ sql(
+ """SELECT approx_top_k_combine(state) FROM (
+ | SELECT state FROM combine_int UNION ALL SELECT state FROM combine_string)"""
+ .stripMargin).collect()
+ }
+ }
+ }
+
+ test("SPARK-59818: combine skips NULL sketches") {
+ withTempView("combine_null_state") {
+ sql(
+ s"""SELECT approx_top_k_accumulate(expr) AS state
+ |FROM VALUES 0, 1, 1 AS tab(expr)
+ |UNION ALL
+ |SELECT $nullSketchState AS state""".stripMargin)
+ .createOrReplaceTempView("combine_null_state")
+ val res = sql(
+ "SELECT approx_top_k_estimate(approx_top_k_combine(state), 2) FROM combine_null_state")
+ checkAnswer(res, Row(Seq(Row(1, 2), Row(0, 1))))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org