[SPARK-58069][SQL][FOLLOWUP] Handle empty approx_top_k combine buffers

[cloud-fan]

What changes were proposed in this pull request?

Followup to #57177.

This change makes CombineInternal serialize and deserialize the placeholder buffer created for
an empty partition. It encodes a missing item type as a zero-length type section and uses a default
string serde for the necessarily empty sketch.

Why are the changes needed?

An untouched approx_top_k_combine buffer has no item type. It can nevertheless be serialized
between aggregation stages, where selecting a type-specific serde or serializing the type currently
throws. Preserving the empty placeholder allows the later merge to initialize it from real input.

Does this PR introduce any user-facing change?

Yes. approx_top_k_combine no longer fails when an empty partition contributes an untouched
placeholder buffer between aggregation stages.

How was this patch tested?

Added a regression test to ApproxTopKSuite that round-trips an empty combine buffer and verifies
that its missing item type, sketch, and configured size are preserved. The staged workflow will run
the required fail-before and pass-after suite cycle before opening the pull request.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Codex (GPT-5)

[viirya]

The fix is well-targeted. genSketchSerDe has no null/fallthrough case, so an untouched placeholder buffer (itemDataType == null) hit a MatchError on serialize between stages — branching on null before that call is exactly the right spot. I traced the round-trip and the downstream merge, and it holds up:

• Defaulting to ArrayOfStringsSerDe for the null-type case is safe because of the invariant itemDataType == null <=> the sketch is empty (every update sets the type from a non-null DDL; merging an empty input only carries an empty sketch). An empty sketch serializes no item bytes, so the serde choice is immaterial to the content.
• On the later merge, the placeholder adopts the real type via the existing updateItemDataType null-handling (and the reverse direction — real buffer, empty input — is a no-op), and the sketch merge runs on in-memory objects, independent of the serde used at (de)serialize time.
• The zero-length type section round-trips symmetrically, and the sketchBytes length math stays correct at typeLength == 0.

One non-blocking thought on testing: the new test exercises the serde in isolation (constructs CombineInternal and round-trips serialize()/deserialize()), which is a fine regression guard and fails before the fix. It doesn't reproduce the actual trigger — an empty partition contributing a placeholder buffer through a real multi-stage aggregation — but making that deterministic is awkward, so the unit-level test seems a reasonable call. If it's cheap, asserting the restored nullCount (0) alongside the type/size/sketch would round out the round-trip coverage.

LGTM.

[uros-b]

The new regression test is a pure in-JVM round-trip of serialize()/deserialize(); it never drives the reported failure through a real SQL query with a shuffle-forced empty partition (e.g. a REPARTITION or UNION of empty + non-empty approx_top_k_accumulate). It also exercises only the concrete maxItemsTracked = 100 variant, not the VOID_MAX_ITEMS_TRACKED = -1 placeholder that the size-unspecified production path (createAggregationBuffer) actually creates. An end-to-end SQL test (and a VOID variant) would faithfully guard the failure this PR claims to fix.

[uros-b]

Please investigate a bit more and coordinate with #57289 accordingly: The fix closes the intermediate serialize/deserialize crash and correctly handles the mixed case (some empty some non-empty partitions resolve the null type during merge/update, so eval sees a concrete type). But a combine over an entirely empty input leaves the final buffer with itemDataType == null, and eval then calls ApproxTopK.genSketchSerDe(null) (line 900) and dataTypeToDDL(null) (line 902); neither has a null/wildcard case (verified against genSketchSerDe, lines 273-291), so the query still fails with a scala.MatchError. This is pre-existing (the query failed earlier, at serialize, before this PR) and arguably outside the declared "between stages" scope, and is separately handled by draft #57289. The author should decide whether to also cover it here (a fallback to uncheckedItemDataType in eval would close it) or explicitly document the remaining gap and defer to #57289; not leave a still-crashing realistic path silently behind a narrower fix.