[SPARK-57908][SQL] Replace generated fast-hash-map TaskCompletionListener with a close-hook helper

[gengliangwang]

What changes were proposed in this pull request?

Janino cannot compile lambdas, so the per-map close hook that HashAggregateExec whole-stage
codegen emits for the fast hash map was written as an anonymous TaskCompletionListener class --
which Janino compiles into a separate generated inner class, one per fast hash map.

Before -- the generated code emits an anonymous class implementing TaskCompletionListener:

$thisPlan.getTaskContext().addTaskCompletionListener(
  new org.apache.spark.util.TaskCompletionListener() {
    @Override
    public void onTaskCompletion(org.apache.spark.TaskContext context) {
      fastHashMap.close();
    }
});

After -- a one-line call to a new compiled helper
HashAggregateExec.addFastHashMapCloseHook(AutoCloseable) (the listener is a lambda in compiled
Scala, which Janino never sees), and the generated map class (row-based and vectorized) implements
java.lang.AutoCloseable:

$thisPlan.addFastHashMapCloseHook(fastHashMap);

The hook registers at the same point with the same effect, once per task; only the anonymous
generated inner class is removed.

Why are the changes needed?

Part of SPARK-56908 (reduce generated Java size
in whole-stage codegen). On a TPC-DS codegen dump (150 queries, 1,572 whole-stage-codegen subtrees),
generated inner classes drop from 542 to 271 (-50%) -- the anonymous listener accounts for
exactly half of all generated inner classes. Fewer generated classes means less Janino work, less
class loading/verification, and less metaspace per query per executor.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests (SingleLevelAggregateHashMapSuite, TwoLevelAggregateHashMapSuite,
TwoLevelAggregateHashMapWithVectorizedMapSuite, WholeStageCodegenSuite,
DataFrameAggregateSuite) exercise the generated fast hash map through both the row-based and
vectorized paths; the close hook still fires once per task.

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

Generated-by: Claude Code (Opus 4.8)

[uros-b]

The last line reads a bit awkward in this prose. Can we make this consistent with some other scaladocs? e.g. This is called by <XYZ>, should be public.

[gengliangwang]

Thanks @uros-b! Reworded the scaladoc so the trailing sentence follows the conventional This is called by <caller>, should be public. phrasing.

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits.
Clean, well-scoped codegen-size optimization: the per-map fast-hash-map close hook moves from an anonymous TaskCompletionListener (a separate Janino-generated inner class per map) to a one-line call into a compiled Scala helper, with the generated map now implements AutoCloseable.

Verification

Confirmed the refactor preserves behavior: fastHashMapTerm is assigned once in the create block and never reassigned, so passing it by value at registration is equivalent to the old field-capture-at-completion; same registration point, same task thread (TaskContext.get()), once per task, unchanged close(). Both the row-based and vectorized generated maps inherit the shared generate()/close(), so the single AutoCloseable header covers both paths.

[gengliangwang]

Merge Summary:

• merged into master 78d458b
• merged into branch-4.x 7a83a4d

Posted by merge_spark_pr.py

[LuciferYang]

late LGTM