fix: keep a late-materialized spill's column concatenation within the pool

[msafonov]

Problem

On em32 (9g heap / 12g off-heap / 5g overhead, 26 GiB pod) parse_parquet device_default with Comet joom-1.1-9 lost 5 of 11 executors to exit 137. At a death, Spark had granted 8 tasks 9 GiB off-heap, but Comet's native pools held 12.8 GiB ("allocated 12926 MiB, reserved 12810 MiB (8 plans)"). The pools carried the difference as overcommit.

A diagnostic build logged every 256 MiB of overcommit with its consumer and native stack. Every entry was ExternalSorter → SpillWorkspace::cover → Gather::concatenate → Gather::try_new, under sort_and_spill (the dedup sort, ~600–1000 GB of spill).

Cause

29b0bb0 made a spill's SpillWorkspace grow infallibly. Its reasoning was that everything the merge's children record already exists. That is true for the streaming merge of sorted runs. It is not true for late materialization:

• Gather::merge calls try_grow before it concatenates a group of column pieces, and the copy is new memory.
• In the spill workspace that try_grow could never fail.

Batches decoded from shuffle share one IPC body allocation per batch. So no piece is freed until the batch's last column has been concatenated, and a spill held up to about twice its input past Spark's grant.

Fix

• The sorted runs of a late-materialized sort register with SpillWorkspace::bounded(). This is a view whose try_grow stays under the execution pool's limits. grow still records existing memory without a limit.
• If the pool refuses a concatenation, it is skipped and the column is gathered from its pieces.
• Gather::merge also reserves the concatenated array's actual size (get_array_memory_size) before keeping it, not just its sliced size.
• The non-late streaming merge keeps the infallible workspace, so the merchants_lifecycle failure that 29b0bb0 fixed does not come back.

Tests

• spill_workspace::a_bounded_view_of_a_spill_workspace_refuses_growth_past_the_pool
• late_materialize::a_spill_concatenates_only_what_the_pool_grants: batches sharing one allocation per batch (IPC round trip), 5 orderings × 3 pool sizes. It asserts peak ≤ pool limit, output matches the reference, and the pool is empty at the end.
  • Before the fix: peak 2 997 665 of 2 283 968 bytes.
  • With only the bounded view: 2 293 006, the excess coming from the capacity vs sliced size of the concatenated array.
  • With both changes: within the limit.
• The vendored datafusion-physical-plan lib passes: 1760 tests.
• Comet core sort tests pass: 27.
• JVM suites: see comment.

🤖 Generated with Claude Code

[msafonov]

Local JVM suites (Spark 3.5, release lib from this branch): CometExecSuite, CometWindowExecSuite, CometPartitionAggregateWindowSuite, CometJoinSuite, WideRowSortFallbackSuite, CometTaskMetricsSuite and CometTPCDSV1_4/V2_7 PlanStabilitySuite: 532 succeeded, 0 failed, 2 canceled (Spark 4.x-only tests).

GCP, parse_parquet device_default for MSK day 2026-10-12, em32 9g/12g/5g (26 GiB pod), 11 executors:

run	task-h	wall, s	spill, GB	exit 137	failed tasks
joom-1.1-9	22.38	1732	845	5	104
joom-1.1-9 + overcommit logging	23.15	1788	839	11	218
this branch	21.07	1712	864	0	0
vanilla (21g heap)	27.47	1801	535	0	0

Output compared with the 1.1-9 run: 897 event types and 815 472 571 rows in both, with the same event_id key sets in every type. The value differences are the dedup tie-break among duplicate ids, which is nondeterministic in vanilla as well.

🤖 Generated with Claude Code