fix(top): bound distributed large Top-N merge

[XuPeng-SH]

Problem and fixes

Fixes #28285 with a version-53-gated ordered Top-N merge hierarchy. The coordinator consumes one ordered stream per worker/CN and emits bounded chunks instead of materializing all P*K candidates.

This revision addresses review 5127191041, the failed CI checks, and an additional correctness defect found in full-scale validation:

• Large OFFSET: restore external Order fallback above the 16,384-row prefix rewrite bound. Spilling payload does not eliminate Top's O(K) key/reference memory.
• Small varlen Top: use byte-bounded resident winner admission, compact dead replacement history, and migrate survivors to spill only under real pressure. Fully rejected spill batches perform no payload writes. Keep the fixed-width fast path and all allocation accounting.
• Remote DOP correctness: preserve the ordered single-producer contract when the remote server adds its write-back merge after decoding. Otherwise worker batches interleave on the wire and produce an incorrect global prefix. Added decode/write-back/DOP and exact multi-CN SQL regressions.
• CI: explicitly check rows.Err() at each DML query. The coverage timeout was an internal LIMIT 1 snapshot lookup unnecessarily spilling into a nonexistent LOCAL file service; the resident admission fix resolves the actual cause without raising timeouts or hiding errors.

Output remains bounded to 8,192 rows and a 64 MiB logical payload window, reduced with a smaller allocator ceiling. Actual capacity, sidecars and growth overlap remain accounted. Large-K ordering keys/references are still O(K), not fully externalized. Protocol and expression-key fallbacks remain in place; #27999's vector ownership safeguards are retained.

Validation

Base e60b3eb3bdfd8ad5f1b5ae3eb8745037371f44d8; reviewed head 7764c28ecdf038b4c7ea0edf26870c52de52103c.

• Full Top, MergeTop and compile packages pass; Top/MergeTop race checks pass. New resident-transition and remote-write-back regressions each pass 100 race repetitions. Relevant SCA (golangci-lint 2.6.2), vet and service build pass.
• Review's 4,194,304-row OFFSET reproducer returns exactly 4,194,303 under a 160 MiB shared budget, peak 103,374,880 accounted bytes, zero residual memory/disk/FD charges.
• Matched 1M-row/64-byte-varlen LIMIT 1 benchmark, five samples: main median 7.682 ms, fixed 7.738 ms; both zero spill bytes.
• CI's unchanged TestExecuteIteration1 passes once and five repetitions. The whole engine/test package passes in 130.732 s with -short -tags matrixone_test -covermode=set -coverpkg=./pkg/.... The failed downstream Coverage job was a consequence of its failed UT producer, not a measured coverage-percentage failure.
• Forced multi-CN DML passes twice in one process, now including 100K source rows and exact 1K ordered keys/payloads, alongside prepared reuse and existing delete/cast checks.
• Final LIMIT BVT passes twice against the same three-CN instance: 35/35 statements each, zero t1 residue after each run. The performance source survives both final protected cleanups. Test-harness startup/configuration failures were resolved separately and are not counted as passing tests.

Issue-shaped 100M-row acceptance

Final production binary SHA-256: c593f182705aef2aa3d0609c600870584068bae4f0d6f5b5a02a5e069185cfe8.

On host 55, same NVMe data/configuration, three CN services in one launch process, GOMAXPROCS=8 and 128-byte varchar payload:

Build	100M-row ORDER BY / LIMIT 5M INSERT
Main e60b3eb3	Still running beyond 300 s; explicitly canceled, returned after 328.25 s; target stayed empty
Fixed, first run	156.078 s; exact result
Fixed, same-instance repeat	145.954 s; exact result

Both fixed results: count and distinct count 5,000,000, key range 0..4,999,999, sum 12,499,997,500,000, payload bytes 640,000,000. Same-instance LIMIT 10 controls return exact 0..9 twice.

The initially tested implementation finished in about 150 s but returned wrong keys; those runs are explicitly rejected as acceptance evidence and led to the remote write-back fix above. Historical author timings are not substituted for this revision's measurements.

Remaining rollout boundaries

This is an issue-shaped, co-located three-CN comparison, not the original three-host TKE deployment or a historical-good build comparison. Main's result is censored, so no exact completion-time speedup is claimed. Mixed-version deployment validation, a full topology/width scale matrix, and human design approval remain separate rollout gates. CI on the new pushed commit must be checked independently of the old failed run.

Design, ownership/change map and detailed evidence: bounded_distributed_topn.md.

[qodo-code-review]

Qodo reviews are paused for this user.

Troubleshooting steps vary by plan Learn more →

On a Teams plan?
Reviews resume once this user has a paid seat and their Git account is linked in Qodo.
Link Git account →

Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center?
These require an Enterprise plan - Contact us
Contact us →

[XuPeng-SH]

Deep re-review of b65946fd08133ad82eb46eab98dc5189643c39c8 against base e60b3eb3bdfd8ad5f1b5ae3eb8745037371f44d8.

Recommendation: approve the code changes. No blocking correctness, performance, or lifecycle finding was confirmed in this pass. Both reproduced problems from review 5127191041 are resolved. This is a self-review COMMENT because the authenticated account is the PR author; GitHub does not permit a self-APPROVE.

The design addresses the identified bottleneck: consolidate actual worker streams in order before sending them across a CN boundary, then merge those streams without retaining the coordinator's entire P*K candidate payload. The remote write-back correction is essential: the newly inserted local edge now preserves the ordered flag and single-producer requirement, so runtime DOP expansion installs the ordered gather. Small-result admission is correctly separated from this topology choice.

Executed verification on this exact head (Darwin/arm64, source-matched native libraries):

• Full Top, MergeTop, compile and process package tests passed; full Top/MergeTop race tests passed. Build, vet and diff-check passed.
• Reran the original 4,194,304-row LIMIT 1 OFFSET 4,194,303 counterexample under the same 160 MiB account: correct result 4,194,303; peak 103,374,880 accounted bytes; cleanup assertions passed.
• Small-varlen LIMIT 1, 1,048,576 rows with 64-byte payload, five samples: head median 6.04 ms, ordered-output head median 5.92 ms; every sample wrote zero spill bytes. A fresh base run also wrote zero bytes (median 7.25 ms). These short runs shared the machine with compilation, so I do not claim a precise speedup; the previous unconditional ~100.7 MB spill per execution is gone.
• Added an independent review-only sort oracle covering a non-leading varchar key, numeric tie-breaker, ASC/DESC, NULLS FIRST/LAST, variable payload widths, 23-row input batches, three resident windows (512/4096/1 MiB), and two execution generations: all 24 combinations returned exact row identities/payloads and released memory/disk/FD charges. The cases cover resident execution and mid-batch spill transition.
• Executed the actual two-CN TestForcedMultiCNDeleteAndInsertIgnore: passed in 12.51 s. Its 100K-row ordered query returned exactly the expected 1000 keys, IDs and payloads; remote topology, OFFSET, prepared reuse, delete and cast controls passed.
• The latest CI run 34076017582 has successful substantive checks, including UT, SCA, UT coverage, both active BVT variants and merged coverage.

Lifecycle review:

Boundary	Result
Resident compaction / spill migration	Old state remains owned until replacement is complete; tested failures and reset release reservations.
Ordered receivers / remote DOP	One ordered producer per edge; remote write-back preserves the invariant; exercised protocol/topology and SQL tests.
Cancellation / cleanup	Existing context and terminal-error propagation retained; receiver cleanup uses one shared bounded deadline. No new closed wait cycle confirmed.
Memory growth	Resident winner history compacts or migrates; output windows are bounded. Local Top still retains O(K) keys/references, as documented.

One non-blocking documentation correction is inline: the design still says every ordered local Top must spill, contradicting this revision's resident admission.

Scope of this recommendation: I did not independently rerun the 100M/5M workload, a historical-good build, or a mixed-binary deployment. The PR body's co-located 100M results remain author-provided evidence; they should not be described as a reproduction on the original three-host TKE environment. Keep the documented O(K) local state and deployment protocol-version boundary visible.

[XuPeng-SH]

[P3] Update the design to match the final admission policy. This paragraph still says every local Top uses external storage once the ordered hierarchy is selected, including small K. At this head, Prepare no longer spills for OrderedOutput, and processBatchResident admits/compacts fitting winners before migration. Please describe that behavior here and clarify that the 8192-row chunk limit applies to spill reconstruction and ordered gather output (the resident Top path can return its fitting result in one batch). This is a documentation correction, not a blocking code finding.

[aunjgr]

Review of commit b65946fd08133ad82eb46eab98dc5189643c39c8.

[design gate] The ordered Top-N merge hierarchy changes the distributed stream contract, remote write-back ownership, memory/spill behavior, and cumulative MORPC capability. This is a major refactor requiring an approved versioned design before implementation approval. The PR explicitly leaves human design approval outstanding and the linked design records validation progress without an approval record. Close the independent design review for this exact ownership/protocol/resource model and record implementation conformance and remaining rollout gates. The issue-shaped measurements and green CI are useful evidence, but do not themselves supply that decision. This is a design-gate review, not a claim of a new runtime defect or completed downstream implementation review.

Evidence: source review and existing passing CI for this commit; no new local test run was performed for this review.

[XuPeng-SH]

Addressed the design-record gap in f8d40ae8ed (documentation-only; runtime code is unchanged).

The versioned design now records:

• the exact implementation revision reviewed (b65946fd08133ad82eb46eab98dc5189643c39c8);
• the selected protocol/ownership/resource decisions and their accepted limits;
• implementation conformance across protocol-v53 fallback, ordered producer edges, bounded gather output, resident winner admission, spill migration, and the large-OFFSET fallback;
• remaining independent-design and mixed-version rollout gates.

It also removes stale statements that said every ordered local Top always spills. The document now matches the implementation: fitting small winners remain resident, dead varlen history is compacted, and survivors migrate to spill only under actual pressure; large K begins in spill mode.

git diff --check passes. The prior exact-head implementation verification and green substantive CI remain applicable because this commit changes documentation only.

@aunjgr please re-review this exact design revision. The independent approval gate is deliberately still marked pending; this update does not self-assert that approval.

[LeftHandCold]

Deep review of f8d40ae8ed29895a459798a60cafe2c7c3e0bc4a against base e60b3eb3bdfd8ad5f1b5ae3eb8745037371f44d8.

APPROVE. Independent design review: PASS for this exact design revision and the implementation through b65946fd08133ad82eb46eab98dc5189643c39c8.

I approve the selected v1 boundaries: protocol-v53 fail-closed negotiation; one producer per ordered edge after runtime DOP expansion; CN-local and coordinator hierarchical MergeTop; row- and byte-bounded reconstruction/gather output; actual-byte resident admission with compaction or spill migration; the large-OFFSET external-order fallback; accounted O(K) local key/reference state; and finite producer draining instead of a new prefix-stop protocol.

No blocking correctness, performance, ownership, wait-termination, or growth-bound defect was confirmed. In particular:

• Q1 ownership: resident/compacted/spilled batches, migration temporaries, expression vectors, merge input heads, and output batches have explicit cleanup owners; failure paths retain the old owner until transition publication and terminal cleanup returns the tested accounts/resources to zero.
• Q2 termination: every ordered receiver has one real producer; runtime worker gathers start concurrently; empty/error/cancel paths preserve a terminal signal; cleanup uses one shared deadline and does not let a late edge-delivery failure mask the query cause.
• Q3 bounds: rejected spill batches write no payload; resident replacement history is admitted/compacted; reconstructed and gathered output is chunked by rows and actual bytes. The documented O(K) per-producer selection metadata is an explicit v1 limit, not an unbounded or hidden payload claim.

Fresh verification on the implementation tree:

• exact-head native CGo build;
• focused Top, MergeTop, compile/remote-topology, protocol-edge, and cleanup regressions;
• full top, mergetop, compile, and process packages in normal and race modes;
• four highest-risk race tests repeated 100 times each;
• the real two-CN DML fixture, including Remote + MergeTop physical topology, exact ordered results, OFFSET, and prepared reuse;
• direct-package build, vet, protobuf/defines tests, and git diff --check.

The final f8d40ae8e increment is documentation-only; the runtime tree is byte-identical to the tested b65946fd0 implementation. A merge-tree check against current main is clean; the only overlapping main change is unrelated compile retry-state initialization.

The documented mixed-version deployment exercise and original three-host TKE/historical-good comparison remain rollout gates. They are not inferred from unit tests and are not source-level merge blockers. The branch is currently behind main and should follow the repository's update policy.