SPARK-1432: Make sure that all metadata fields are properly cleaned - #338

While working on spark-1337 with @pwendell, we noticed that not all of the metadata maps in JobProgessListener were being properly cleaned. This could lead to a (hypothetical) memory leak issue should a job run long enough. This patch aims to address the issue.

Author: Davis Shepherd <[email protected]>

Closes #338 from dgshep/master and squashes the following commits:

a77b65c [Davis Shepherd] In the contex of SPARK-1337: Make sure that all metadata fields are properly cleaned
(cherry picked from commit a3c51c6)

Conflicts:

	core/src/main/scala/org/apache/spark/ui/jobs/JobProgressListener.scala

SPARK-1005 Ning upgrade

While working on spark-1337 with @pwendell, we noticed that not all of the metadata maps in JobProgessListener were being properly cleaned. This could lead to a (hypothetical) memory leak issue should a job run long enough. This patch aims to address the issue.

Author: Davis Shepherd <[email protected]>

Closes apache#338 from dgshep/master and squashes the following commits:

a77b65c [Davis Shepherd] In the contex of SPARK-1337: Make sure that all metadata fields are properly cleaned

Run cluster-api-provider-openstack-image-build in sigs.k8s.io

### What changes were proposed in this pull request?

Step 5 of task 87 (`sql/varka/plans/PLAN_TASK_87.md`, section 3.1 step 2): `methodByteBudget` becomes the limit its name says. Stacked on apache#338 (step 4); the diff here is the last commit once that merges.

After the class is built, the emitter measures it in the units the JVM enforces (`VarkaEmittedClass`, from apache#336). A group with a loop or epilogue method over the budget is split at its middle output (`groupOutputs` takes forced group starts) and the class is built again, until every group's methods fit or the groups over budget are single outputs. Weight groups first because weight is known before anything is built; bytes decide because bytes are what the JVM reads. A shape still over a limit when no split is left declines with a new `VarkaEmitDeclined`: an `IllegalArgumentException`, so every caller's fallback contract holds unchanged, carrying the reason - the method, its bytes, the budget - and the indices of the outputs whose own group cannot fit, so that a compiler can fuse the rest without them. The driver is measured too and cannot be regrouped (it sets up every output and gains a call per group), so a driver over the budget declines naming no output. The class-file caps, 65535 bytes of code in one method and 65535 constant pool entries, are read from the same measurement.

At the production limit of 8000 bytes nothing changes on the `make_date` ladder: after steps 3a and 4 its weight groups already fit, so no rung regroups and none declines. The tests see the regroup by lowering the budget. At 1500 bytes the sixteen-output ladder regroups from weight's four groups to sixteen single outputs (each method 1067 to 1236 bytes), every method under the budget, the answers unchanged, the emission byte-identical run to run. At 300 bytes every output is stuck and the decline names all sixteen. At 2000 bytes over sixty outputs the single-output groups fit and the 2806-byte driver does not. The shape that declines at 8000 is a single heavy output - a balanced `greatest` over thirty-two `add_months`, one group whose masked loop method reads 25629 bytes - which the legacy form emits into methods HotSpot never compiles. Splitting inside one output is deliberately not attempted; it would forfeit the register residency that is the point of fusing.

The class-file caps are out of reach of any shape the IR caps admit, in either form (sixty-two `add_months` outputs put the legacy single epilogue at 49339 bytes), so their readings are pinned on a measurement built by hand and the test says so rather than claiming a shape covers them.

Also here: the fuzzer treats a size decline under the budget as an outcome and asserts it carries a size; `dev/varka_emit.sh` prints a decline instead of a stack trace; the oracle's opt-in option audit records a declining arm as its answer; `VarkaEmitOptions.methodByteBudget`'s doc describes the whole mechanism; and section 9.4 of the plan records the step, including what it leaves to task 169: the compiler still admits by the weight caps alone, so a heavy single output is fused at plan time and declines on the executor through the evaluator's existing catch, and making that a plan-time decline EXPLAIN shows - demoting exactly the outputs `VarkaEmitDeclined` names - is that task's contract, built on this type.

### Why are the changes needed?

Steps 3a and 4 made a group's methods a function of the group; this step makes the budget hold for every shape, not only the family measured so far: a group that is over the limit on its own is split until it fits, and a shape no split can fit is refused with a reason rather than emitted into a method the JVM will never compile.

### Does this PR introduce _any_ user-facing change?

No. Without the switch, which is still off by default, the emitted classes are byte-identical and nothing is measured.

### How was this patch tested?

Three new tests in `VarkaEmitterBudgetSuite`: the regroup at a 1500-byte budget (group count, every method under the budget, determinism method by method, answers against the reference evaluator at lengths 1, 1024 and 1031); the declines at 300 bytes on one output and on sixteen, and at the production limit on the heavy single output, checking the message names the method, the bytes, the budget and the outputs; and the driver's decline at 2000 bytes over sixty outputs plus the class-file cap and constant pool readings on hand-built measurements. The whole `codegen.varka` test package passes locally (282 tests, 27 cancelled as opt-in or hsdis-gated). `dev/scalastyle`, `dev/lint-java`, `dev/varka_quote_check.py` and the non-ASCII and 100-column scans pass.

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

Generated-by: Claude Code (Claude Fable 5.1)

### What changes were proposed in this pull request?

Step 6a of task 87 (`sql/varka/plans/PLAN_TASK_87.md`, section 6): `VarkaMethodSizeBenchmark` gains its second arm and its committed files are regenerated with both forms in one table per rung, banded over ten runs. Stacked on apache#339 (step 5); the diff here is the last two commits once that merges.

The two forms are named explicitly so the labels survive the default changing: `single epilogue`, the legacy emission (`methodByteBudget` 0), and `epilogue per group`, the emission under the budget at `HugeMethodLimit` - a group's methods set up only their group (apache#337), the epilogue is one method per group (apache#338), and the class is measured and regrouped (apache#339). `VarkaEmitBudget` is public so a benchmark outside the emitter's package can name the limit. The results, in M rows/s at 512 bits on the masked arm (the arm that reaches the single epilogue's crossing at thirteen outputs):

| outputs | even batches | ragged batches |
| ---: | :--- | :--- |
| 4 | 441.9 -> 434.1 | 431.2 -> 442.8 |
| 12 | 134.0 -> 143.1 | 135.2 -> 142.3 |
| 13 | 120.9 -> 132.9 | 40.7 -> **131.7** |
| 16 | 94.4 -> 106.3 | 32.9 -> 104.9 |
| 60 | 24.1 -> 29.4 | 8.7 -> 28.9 |

The cliff is gone: on ragged batches the per-group form reads within a few percent of its own even-batch rate at every rung, where the single form had lost three quarters of its rate from thirteen outputs. And from twelve outputs the per-group form is faster on even batches too, by 7% at twelve and 22% at sixty, which no prediction had: each loop method now sets up only its group's segments and literals, and that outweighs the driver's extra epilogue calls. The 128-bit file reads the same way (13 outputs ragged: 24.9 -> 44.0). Section 9.5 of the plan scores the registered predictions: 3 held and understated the gain, 4 fell on the fast side, 5 is superseded by the design; the default's rule - nowhere worse beyond its band on the masked arm at any rung, length or width, and on the null-free arm up to twelve outputs - is met, and the flip is step 6b.

One lead for task 189: in this fork the single form's null-free rows at 128 bits from twelve outputs sit in the deoptimization cycle (0.3 to 0.7 M rows/s ragged) and the per-group form's do not (40 to 56). One fork proves nothing about a bimodal effect; the plan records it as a lead, not a property.

### Why are the changes needed?

The plan's rule for flipping the default is a measurement, and this is it: both forms in one file, at both widths, with the band that says which differences mean anything.

### Does this PR introduce _any_ user-facing change?

No. The default is unchanged in this PR.

### How was this patch tested?

The benchmark is the test: `dev/varka_bench_regen.sh catalyst VarkaMethodSizeBenchmark` (pinned, load 0.57 at start, canary clean, invariants ok at both widths) and `dev/varka_bench_repeat.sh catalyst VarkaMethodSizeBenchmark 10 --band`. `dev/varka_quote_check.py` passes for every number quoted in the plan. The suites do not change.

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

Generated-by: Claude Code (Claude Fable 5.1)