etl: replay sorted non-overlapping runs sequentially instead of heap-merging

[AskAlexSharov]

Adds a sorted-input fast path to the ETL collector: when every spilled run's keys arrived already sorted and consecutive runs do not overlap, Load replays the runs sequentially instead of building the k-way merge heap.

Motivation

Audit of the 58 collector call sites in the tree found two producers that emit globally sorted keys:

• db/state/inverted_index.go — w.indexKeys (key = BE8 txNum, monotone, fed on every state write during exec; the highest-volume ETL user in the node)
• cl/antiquary/beacon_states_collector.go — the antiquary collectors (ascending slot keys)

For these producers the buffer sort was already free (sortableBuffer.Sort short-circuits on slices.IsSortedFunc), but Load still paid for the merge heap: heapify plus a heapPop+heapPush (with key comparisons) per element. The spill-to-tmp-file and mmap read-back are unavoidable — the data must survive between Collect and Load under a RAM bound — so this PR removes only the per-element heap work, not the spill.

How it works

• sortableBuffer tracks input sortedness in Put by comparing against the previous key (comparison stops once broken).
• On every flush the collector records a sortedRun{first, last, sorted} aligned 1:1 with its data providers. For an async flush the boundaries are captured before the buffer is handed to the background sorter, which is why they are only valid when the input arrived sorted.
• canReplaySequentially(runs, providers) — a pure decision function — requires every run sorted and run[i-1].last <= run[i].first. When it holds, mergeSortFiles iterates the providers in order; otherwise it falls back to the existing heap merge. Both drivers share the same per-element processing (quit checks, SortableAppendBuffer concatenation, SortableOldestAppearedBuffer dedup), so semantics cannot diverge.
• A single in-RAM run is marked sorted after Sort() regardless of input order, so the common everything-fits-in-RAM collector also skips the heap.

No call-site changes: the II writer and the caplin antiquary collectors benefit purely through the collector.

Dup-ordering equivalence

The heap orders elements by (key, run index) and each run is stably sorted by (key, insertion order). With non-overlapping runs, an equal key at a run boundary drains run i completely before run i+1 — exactly what sequential replay does — and within a run insertion order is preserved by the stable sort in both paths. So the sequential replay's output stream is byte-identical to the heap's, including duplicate keys under DupSort tables. Tests pin this: byte-identical Load streams fast-path vs forced-heap on the same input (equal keys spanning run boundaries included), identical DupSort table contents, and fall-back tests that break sortedness at the second Put, mid-run, at a run boundary, and scattered.

AppendDup for the II load path: deferred

The optional follow-up (using AppendDup instead of Put in the II loadFunc when the stream is sorted) is not included. For indexKeys, values under one txNum arrive in state-write order, not in byte order, and MDBX AppendDup requires ascending dup values — so it would change semantics. Left as follow-up work.

Benchmarks

Apple M4 Max, -count=6, benchstat old=main new=this branch (after-side used a bounded -benchtime 5x). LoadMonotoneDupKeys mimics the indexKeys shape (BE8 monotone keys, 4 values per key, ~10 spilled runs). sorted_* take the new sequential path; random_* pin the unchanged heap path; PutOnly isolates the new per-Put sortedness compare.

│     sec/op      │    sec/op     vs base                 │
SortableBufferPutOnly/random_100k-16     678.4µ ±  1%     860.4µ ± 21%  +26.82% (p=0.002 n=6)
SortableBufferPutOnly/random_500k-16     3.440m ±  1%     3.513m ±  2%   +2.13% (p=0.002 n=6)
SortableBufferPutOnly/sorted_100k-16     622.0µ ±  1%     699.9µ ±  3%  +12.52% (p=0.002 n=6)
SortableBufferPutOnly/sorted_500k-16     3.150m ±  1%     3.479m ±  5%  +10.44% (p=0.002 n=6)
SortableBufferLoadOnly/random_100k-16    3.632m ±  4%     3.728m ±  7%        ~ (p=0.132 n=6)
SortableBufferLoadOnly/random_500k-16    19.00m ± 29%     18.95m ±  5%        ~ (p=0.310 n=6)
SortableBufferLoadOnly/sorted_100k-16    3.031m ±  8%     1.718m ±  3%  -43.32% (p=0.002 n=6)
SortableBufferLoadOnly/sorted_500k-16   13.819m ±  7%     8.574m ±  2%  -37.95% (p=0.002 n=6)
LoadMonotoneDupKeys/100k_x4-16          12.613m ±  8%     5.717m ±  5%  -54.67% (p=0.002 n=6)
LoadMonotoneDupKeys/500k_x4-16           61.44m ±   ∞ ¹   28.49m ±  2%  -53.64% (p=0.004 n=5+6)
geomean                                  5.024m           4.070m        -18.99%
¹ need >= 6 samples for confidence interval at level 0.95

The PutOnly/random_100k +26.8% is bounded-benchtime noise (±21%): a recheck at -benchtime 1000x gives 700µs vs the 678µs baseline (+3%), consistent with random_500k. The honest cost summary: the pure-Put microbenchmark pays ~2-12% for the sortedness compare (32-byte keys; real II keys are 8 bytes), while the Load side of the same sorted workload drops 38-55% — in absolute terms for sorted_500k, Put +0.3ms vs Load -5.2ms per pass.

Notes

• TDD: red-first via the new API (canReplaySequentially, run tracking) — see db/etl/sorted_replay_test.go.
• CLAUDE.md conventions apply to this PR.

https://claude.ai/code/session_01UYCsHj9HYTJnqUCy8a7W91

[Copilot]

Pull request overview

This PR optimizes db/etl load-time merging by adding a fast path for the common case where spilled runs are already globally ordered and non-overlapping, allowing sequential replay instead of k-way heap merging (targeting high-volume ETL producers like the inverted index key stream and Caplin antiquary collectors).

Changes:

• Track whether keys arrived in non-decreasing order inside sortableBuffer.Put, and record per-flush run boundaries (first/last) when that condition holds.
• Refactor mergeSortFiles to choose between sequential replay (when runs are sorted and non-overlapping) and the existing heap merge, sharing the same per-element processing logic.
• Add equivalence/unit tests and a benchmark to validate correctness and quantify the performance impact.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
db/etl/collector.go	Records flushed-run metadata and adds sequential replay path in mergeSortFiles (fallback to heap when needed).
db/etl/buffers.go	Adds input-sortedness tracking to sortableBuffer during Put.
db/etl/etl_test.go	Updates tests for the mergeSortFiles signature change (runs metadata now passed separately).
db/etl/sorted_replay_test.go	New tests covering run decision logic, stream equivalence vs heap merge, DupSort behavior, and fallback cases.
db/etl/sorted_bench_test.go	New benchmark simulating monotone dup-key workload similar to inverted index collection.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[AskAlexSharov]

Review follow-up: removed the per-Put sortedness compare entirely. Every run is sorted after Sort() by construction, so eligibility for sequential replay needs only run boundaries — now captured post-Sort inside the flush (async-safe: the flush goroutine fills its run slot before provider.Wait() in mergeSortFiles). Effects: zero hot-path overhead on Put (the earlier +2–12% microbench cost is gone), and the fast path now also fires for any producer whose runs happen not to overlap, not just perfectly-sorted input. Output equivalence is unchanged — the equal-boundary tie still drains run i before run i+1, matching the heap's (key, run index) order. Sortedness-break tests became pure output-equivalence tests since intra-run breaks can now legally take the sequential path.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (3)

db/etl/collector.go:401

• The two load drivers handle 'first read of a run failed' differently: replaySequentially returns an error, while mergeViaHeap panics. That means a truncated/corrupt spill can crash the process on the heap path but only error on the sequential path. Consider unifying behavior (preferably return an error in both paths, since this is a library/data-layer utility) so failure semantics are consistent regardless of which driver is selected.

func replaySequentially(logPrefix string, providers []dataProvider, process func(key, value []byte) error) error {
	for i, provider := range providers {
		// A recorded run is never empty, so EOF on the first read means a
		// truncated or corrupt spill; fail loudly like the heap path does.
		key, value, err := provider.Next()
		if err != nil {
			return fmt.Errorf("%s: reading first element of run %d of %d (provider=%s): %w",
				logPrefix, i, len(providers), provider, err)
		}
		for {

db/etl/collector.go:410

• This error message is reused for all dataProvider implementations, including in-memory providers, so 'from disk' can be misleading. Consider changing it to 'from provider' (or include the provider via %s) to make debugging accurate in both RAM and file-backed cases.

				return fmt.Errorf("%s: error while reading next element from disk: %w", logPrefix, err)

db/etl/collector.go:73

• The docs say 'every run must be sorted', but the sortedRun struct doesn’t track sortedness (only valid), and the current boundary recording happens after Sort(), making runs inherently sorted by construction. To avoid confusion, consider renaming valid to something like filled/recorded, and adjust the docstring to reflect that the decision is based on (1) boundaries being recorded and (2) non-overlap (last <= next.first).

// sortedRun holds one flushed run's key boundaries, captured after Sort —
// every run is internally sorted by construction. Runs align 1:1 with
// Collector.dataProviders; an async flush fills its run in the flush goroutine,
// which completes before mergeSortFiles reads it (provider.Wait).
type sortedRun struct {
	first []byte
	last  []byte
	valid bool
}

// canReplaySequentially reports whether runs can be replayed one after another
// instead of heap-merged: every run must be sorted and consecutive runs must not
// overlap. An equal boundary key is fine — the heap breaks key ties by run order,
// which sequential replay preserves.
func canReplaySequentially(runs []*sortedRun, providers int) bool {

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.