cloud-fan commented on code in PR #59094:
URL: https://github.com/apache/spark/pull/59094#discussion_r4139328069
##########
sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala:
##########
@@ -549,6 +549,71 @@ 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)
+ )
+ }
+ withTempView("estimate_invalid_k") {
+ sql(
+ s"""SELECT approx_top_k_accumulate(expr) AS state
Review Comment:
**Nit (P3):** This fixture includes a populated sketch that can raise
APPROX_TOP_K_NON_POSITIVE_ARG on its own. If k validation moves behind the
NULL-state return again, collect() can therefore satisfy this assertion without
proving that the row-evaluated NULL state rejects k = 0, so the claimed runtime
NULL-path regression coverage can remain green.
**Verification:**
- **Regression:** Exercise a non-positive k against a row-evaluated,
non-foldable NULL sketch state with no alternate row able to throw the expected
validation error.
##########
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
Review Comment:
**Non-blocking (P2):** Returning the untouched buffer is not type-neutral
once this NULL row is the only input to a partial aggregate: serialization
fills the unset itemDataType from the coercion-widened state schema, and final
merge can reject it against the real sketch's DDL type. That makes a supported
combine over the same NULL and non-NULL states fail with
APPROX_TOP_K_SKETCH_TYPE_NOT_MATCH solely because those rows landed in
different partial aggregates.
**Verification:**
- **Regression:** Combine a real sketch with a coercion-widened NULL state
isolated in a different partial aggregate and verify the result is independent
of partition placement.
- **Compatibility:** Verify that empty or NULL-only final results retain
valid concrete state metadata while incompatible non-NULL sketch types remain
rejected.
--
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]
