perf(cudf): Bound distributed TopN and Grace restore

[winningsix]

Summary

• Preserve the upstream community CudfTopNRowNumber implementation and tests byte-for-byte.
• Automatically select a bounded-memory backend for the partitioned row_number, limit=1 shape used by Job 144. It remains the same public CudfTopNRowNumber operator and requires no private enable switch.
• Add local-device exchange, packed micro-bucket spill/restore, and pinned bounce-buffer paths for Grace join restore.
• Stabilize both UCX receive-credit and ReadyToReceive backpressure wakeups.
• Keep the implementation inside Velox; no external cuDF source patch is included.

Validation

• Job 144 r923 end-to-end after the latest-dev rebase: 459.640 s total; exact Iceberg output of 9,053,014 rows (4 files, 1,517,554,252 bytes).
• Plan shape is identical to r917: 37 fragments, 36 exchanges, 230 main tasks, 1,866 operator records; 46 scans read exactly 2,691,810,553 rows in 2,837 vectors.
• r921 reproduced a stable fragment-23/25 exchange stall at the 1,000 s timeout. The queue check and dormant-flag publication race was closed under the queue mutex; r923 crossed the same topology and completed.
• No CUDA 719, OOM, executor failure, ENOSPC, or Xid in the valid r923 run.
• The r923 artifact was built from dev c123471. The branch is rebased onto dev fc40512.
• Current head b08075a: cuDF exec, TopN, config, packed-spill, OrderBy, HashJoin, BatchConcat, FilterProject, and UCX targets all compile and link.
• Current head b08075a: upstream TopN suite 7/7 and spillable-backend suite 12/12 passed (19/19 total; one manual benchmark remains disabled).
• git diff --check passes. The current source-separation head has not yet had another four-GPU full18 run; r923 remains the end-to-end evidence for the pre-separation implementation.

Performance finding

• Against r917, r923 is +10.316 s with the same plan, scan cardinality, and exchange payload. Grace increased from 23 drains / 184 partitions to 28 / 224, adding 44,592,557,312 drain-input bytes and 46,440,199,616 resident-plus-disk restore bytes.
• TopN did not regress: finalize restore changed from 352.884 s to 352.272 s and concatenate from 102.645 s to 96.562 s. Scan CPU also fell despite higher aggregate wall time. The remaining performance target is Grace admission/victim selection, not scan batch size or TopN restore.
• r918's narrower reclaim-wave experiment increased Grace drains to 33/264 and was rejected; it is not in this PR.

Follow-up

This is the native half required by Spark-Gluten PR #98. Further performance work should target Grace admission/restore scheduling without fragmenting reclaim waves.

[winningsix]

Job 144 validation update for head 9c1b4a5:

• Rebased/merged current dev; GitHub reports MERGEABLE (no remaining code conflict).
• Fixed the external native UDF ABI regression exposed by the dev merge while preserving the existing CudfFunction/CudfExpression virtual layouts.
• Focused tests: 87/87 passed (function registry 18, expression selection 27, CudfTopNRowNumber 20, UCX queue manager 22).
• 4-GPU full18 Job 144 r931: SUCCESS in 448.592s, versus r928 458.873s (-10.281s, -2.24%).
• Exact output: 9,053,014 rows in 4 data files; Iceberg snapshot 7608099636557673001.
• Execution shape unchanged: 37 fragments / 36 exchanges / 230 tasks; scan 2,691,810,553 rows / 2,837 vectors.
• No SIGSEGV, CUDA 700/719, OOM, ENOSPC, ExecutorLostFailure, or Xid signature in the successful run.

The current BLOCKED label is from the repository-wide pre-commit --all-files job, not a merge conflict. PR-title validation and labeler both pass.

[winningsix]

Updated in 20b680e with the generalized replayable spill/admission control plane.

Key changes:

• Extract ReplayableDeviceMemoryWorkspace (request, Driver future, cancellation, retry timing).
• Make resident HashJoin build finalization replayable after the one-shot peer barrier instead of immediately falling back to Grace on transient pressure.
• Use the shared lifecycle in HashJoin Grace restore, TopNRowNumber, BatchConcat, and OrderBy; execution operators no longer call the raw workspace API directly.
• Fix OrderBy charging a materialized in-memory output from the configured max run size (the regression test previously requested 4 TiB on a 32-GiB GPU) and avoid a redundant EOS admission.
• Document the shared control-plane contract and operator-owned spill-policy boundary in the cuDF README.

Validation:

• HashJoin: 140/140 enabled tests passed.
• TopNRowNumber: 20/20 passed.
• BatchConcat: 14/14 passed.
• OrderBy: 11/13 passed. The remaining two are pre-existing local run-count assertions (sourceChunks 1 vs required 23/7) caused by the upstream Values/BatchConcat path combining inputs; query correctness passes and the same assertions fail with the old raw request/future code.
• 4-GPU full18 Job 144 r968: 477.086s, exact 9,053,014 rows / 4 files, 37 fragments / 230 tasks, 16 Grace drains / 201.394 GiB, two successful replay waits, zero immediate fallback, no CUDA/OOM/ENOSPC/Xid/executor failure.
• Exact same-binary off control r967: 482.338s and 18 Grace drains / 203.048 GiB. The replayable path saved 11.074s in the cleaner r966 on/off comparison and removed both immediate fragment-23 fallbacks.
• r969 adverse victim repeat also completed exactly (479.152s); its extra 83.188-GiB fragment-7 Grace victim did not overlap the finalize-wait fragments, confirming it was not introduced by replay.

[winningsix]

Final PR-head validation update (c0c7c90826a12b09ac33d82044af7d4f59ff7ba8):

• Moved the reusable ReplayableDeviceMemoryWorkspace into CudfOperatorBase, which now owns one cancellation-safe coordinator per operator and resets it before derived close. HashJoin build/probe, TopNRowNumber, BatchConcat, and OrderBy use the protected base API; execution operators do not directly call the raw workspace-admission API.
• Full affected library rebuild passed. Focused tests remain HashJoin 140/140, TopNRowNumber 20/20, BatchConcat 14/14, and OrderBy 11/13 (the same two pre-existing local source-run-count assertions; query correctness passes).
• Four-GPU full18 Job 144 r970 completed in 481.294s from the final head.
• Exact correctness: Iceberg snapshot 8801636660452064210, 4 files, exactly 9,053,014 rows.
• Shape unchanged: 37 fragments / 36 exchanges / 230 tasks / 1,994 native operator records; scan 2,691,810,553 rows / 2,837 vectors.
• The shared coordinator replayed 5 transient resident-build finalize pressure waves in 205ms aggregate, with zero immediate fallback. Grace remained at the clean 16-build / 128-partition / 201.394-GiB probe-input shape.
• No CUDA 719, OOM, executor loss, ENOSPC, Xid, or UCX fatal signature.
• Main MPP interval was 333.053s. Mean four-card utilization was 39.65%; all cards were zero in only 0.96% of samples. The remaining bubble is still mostly per-card/fragment supply imbalance and host restore/decompression rather than admission waiting or GPU copy synchronization.

GitHub reports the head as MERGEABLE with no code conflict. The remaining pre-commit --all-files failure is from current dev baseline files outside this PR diff (Jenkinsfile executable bit, generated README index, and workflow formatting/zizmor findings); changed-file hooks for this PR pass.

[winningsix]

Review-architecture update in 7b7b34e710ecb3a99336458e50d161e1730c7bfd:

1. Memory reclaim ownership: removed the targetPool argument from the public Operator::MemoryReclaimer::create(DriverCtx*, Operator*) API. The default CPU reclaimer again targets op->pool(). CudfOperatorBase installs a cuDF-owned reclaimer subclass on its device leaf and only overrides pool selection; the device resource continues to own its allocator/arbitrator.
2. Join reuse: cuDF continues to inherit the Velox JoinBridge start/cancel/promise lifecycle, and CPU/cuDF bridges now share the result-or-waiter transition. Recursive Grace partition ordering already uses the shared IterableSpillPartitionSetBase. The payload remains deliberately separate: CPU HashJoinBridge carries BaseHashTable/RowVector spill shards, whereas cuDF carries CUDA hash tables and packed host/disk batches, avoiding a device-to-RowVector conversion.
3. Utilities: moved makeZeroOffsetsColumn into cuDF exec utilities and reused it from empty/all-null nested-column construction. Also consolidated the duplicated HashJoin/NestedLoopJoin expression-column append helper in ExpressionEvaluator.h.

Validation:

• Full affected cuDF exec archive and Gluten relink succeeded.
• HashJoin 140/140 passed.
• FilterProject 122/127 and NestedLoopJoin 57/59; the remaining failures reproduce existing unsupported-expression/local assertion cases and are outside these helpers.
• Two fresh 4-GPU full18 Job 144 runs succeeded: r977 434.022s / 287.712s main and r978 435.086s / 288.171s main. Both wrote exactly 9,053,014 rows in 4 files with 37 fragments, 36 exchanges, 230 tasks, 1,994 main operator records, and scan shape 2,691,810,553 rows / 2,837 vectors. No CUDA/OOM/executor/ENOSPC/Xid/UCX fatal error.
• Interleaved pre-refactor same-host control r979 was 428.724s / 285.135s main (the historical r974 was 420.349s / 276.152s). Thus most historical delta is host/scheduling drift; the residual new-vs-control main-stage delta is 2.577-3.036s (0.90-1.06%), with no increase in stable scan/UCX/TopN/Grace work. This is recorded explicitly rather than claiming a 420s reproduction.

GitHub currently reports the new head MERGEABLE. Fresh checks have been triggered.