jubins opened a new pull request, #59094: URL: https://github.com/apache/spark/pull/59094
### 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 incorrectly about the expression. **`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. These are changes compared to released versions: `approx_top_k` is annotated `since = "4.1.0"`. ### How was this patch tested? Four 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 is nullable when its state is nullable` - `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. All four tests were confirmed to fail without the fix. Reverting the two source commits and re-running them produces exactly the reported failures — `Cannot invoke "InternalRow.getBinary(int)" because "stateEval" is null` for the three estimate tests and `... because "inputState" is null`, aborting the stage, for the combine test — and all four pass with the fix applied. The full `ApproxTopKSuite` passes (200 tests: 196 existing plus the 4 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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
