[WIP] perf(core): Add a new mutation pipeline for per predicate runners

[ghost]

Per-predicate mutation pipeline — revival of #9467

This branch revives the abandoned per-predicate mutation pipeline started by @harshil-goel. The original work added an alternative path through applyMutations that fans mutation edges out by predicate and processes each predicate's batch concurrently — exposing parallelism the legacy serial path can't. It was abandoned with red CI (most notably TestCountIndexConcurrentSetDelScalarPredicate) and a trail of WIP debris; this work picks it up and gets it green.

The bulk of the diff is correctness fixes the original author didn't get to, each committed individually with a full trace narrative. TL;DR: scalar Del-of-old-value was overwriting new Sets inside both ProcessCount and ProcessSingle's data-list write, leaving counts off-by-one and post-mutation reads returning "no value"; InsertTokenizerIndexes deadlocked on uids ≥ 2⁶³ via a signed-int dispatch hash; reused key buffers in ReadPostingList and the async rollup queue silently corrupted unrelated reverse lists under the 21M live load; IterateDisk's IsEmpty had been stubbed to false and broke has(...) after star-deletes; the pipeline's star-delete fast path skipped index Del generation during background indexing because it forgot to set the write-context flag; vector inserts were fanned across 10 goroutines that raced on shared HNSW graph state. There are also four new in-process regression harnesses for [uid] @reverse @count and an exhaustive count(~genre) check across all 592 non-empty Genre entities in the 21M dataset.

The on/off feature flag became --feature-flags="mutations-pipeline-threshold=N" — N=0 disables the pipeline entirely (legacy path), N=1 always uses it, N>1 only routes mutations with ≥ N edges to it. The pipeline pays a per-predicate goroutine spin-up cost, so tiny mutations are slightly slower on it; bulk multi-predicate mutations are faster (crossover ≈ 100 edges in in-process benches; live-loading the 1M dataset is ~1.5× faster end-to-end at threshold=1 vs. legacy).

Status: local ./t --suite=systest-baseline, ./t --suite=load (incl. systest/bgindex), and the unit tests are green; the default is currently flipped to 1 so CI exercises the pipeline path on every mutation. Before merge, reset the default to 0 (off-by-default, opt-in via the feature flag) once the canonical Linux CI has been clean for a soak window. Operators who want the bulk-load speedup can opt in with a higher threshold (e.g. 200) so only large mutations take the new path.

Commits in this PR are stacked for easier review.

TODO: set the default mutations-pipeline-threshold to either zero or 100 (following more testing).

[trunk-io]

     

Failed Test	Failure Summary	Logs
TestOrderAndOffset	The test failed due to an unexpected EOF error when trying to send a POST request to the GraphQL endpoint.	Logs ↗︎
TestSystestSuite/TestCountIndexConcurrentSetDelScalarPredicate	The test failed because the expected value was 1, but the actual value was 0, indicating a mismatch in the scalar predicate count.	Logs ↗︎
TestSystestSuite	The test suite failed to complete successfully.	Logs ↗︎

View Full Report ↗︎ ⋅ Docs

[matthewmcneely]

Revival of harshil-goel/mutation-pipeline — summary of changes on top of 978a0d41d … b0c464598

The original WIP added the per-predicate mutation pipeline but was abandoned with a red CI (most notably TestCountIndexConcurrentSetDelScalarPredicate) and lots of WIP debris. This work picks it up and gets it green end-to-end against ./t --suite=systest-baseline, the acl integration suite, and systest/21million/live (the bulk-loader-shaped workload that motivated the original PR).

Phase 1 — cleanup (no behavior change)

Commit	What
436b55f8b	Resolve the merge of main into the branch (import paths hypermodeinc/dgraph → dgraph-io/dgraph, license header, two-line fix in mvcc_test.go::TestRegression9597 for the new Deltas type).
a2fcc09f7	Strip ~30 unconditional fmt.Println debug calls from hot paths (posting/lists.go, index.go, list.go, mvcc.go). They fired on every read/mutation/rollup.
4fe27e1d4	Remove dead commented-out code: ~175 lines of an alternate InsertTokenizerIndexes impl, plus the 101 commented-out lines of BenchmarkProcessListIndex in worker/draft_test.go (referenced methods that don't exist).
8fbb0bdf2	Replace the hardcoded featureFlag := true in applyMutations with a real superflag knob mutations-use-pipeline. (Subsequently superseded by the threshold form below.)

Phase 2 — correctness fixes (real bugs)

Commit	What
2d2afb190	Document why TestCount in worker/sort_test.go is t.Skip()'d. The test bypasses Oracle conflict checking; both legacy and new paths fail it.
43b826d0a	Scalar Del-of-old-value wiping the new Set in ProcessCount. handleOldDeleteForSingle appends a synthetic Del oldVal posting next to the user's Set newVal so index/count diffing can see the prior value. ProcessCount then iterated postings and called updateMutationLayer(post, singleUidUpdate=true) on each. For non-Lang scalars fingerprintEdge returns math.MaxUint64, so both postings collide on the same Uid; the second iteration's Del unconditionally rewrote currentEntries to [DeleteAll], dropping the new value. Fix: in the !isListEdge branch, when the postings list contains a Set/Ovr, treat any Del as synthetic and skip it for the data-list update.
792923279	InsertTokenizerIndexes deadlock on uids ≥ 2^63. The dispatcher in posting/index.go did chMap[int(uid)%numGo] <- uid. int(uid) is signed, so for any uid in [2^63, 2^64) the modulo result is negative; chMap[-N] returns Go's zero value — a nil channel — and sending on a nil channel blocks forever. Surfaced by live load on 1million.rdf.gz: the very first batch wedged. Two-character fix: chMap[int(uid%uint64(numGo))] <- uid.
8ccdae630	IterateDisk had IsEmpty stubbed to false (introduced by branch commit 41d6445ce1 "fixed some bug" with no comment). That forced every key the iterator found to be reported as non-empty. Broke has(<predicate>) for any uid whose value had been removed via star-deletion. Surfaced by TestSystestSuite/TestHasDeletedEdge in systest/mutations-and-queries. Restored.
de90d97e0	Nil-pointer deref in ProcessSingle on multi-Del-per-uid mutations + strip two leftover fmt.Println("READING…") calls in worker/task.go that the Phase 1B sweep missed. The GraphQL deleteTask(filter: {}) cleanup expands into multiple Del edges per entity (one per predicate); when ≥2 land on the same uid in one batch, findSingleValueInPostingList returns nil on the second iteration, then string(edge.Value) == string(oldVal.Value) panics. Fix: nil-guard the deref and move oldVal per-iteration so stale values can't bleed across edges. Surfaced by graphql/e2e/auth/TestOrderAndOffset.
0a9ffc174	Reused-key-buffer aliasing across ReadPostingList, the in-memory cache, and the rollup queue. The pipeline's ProcessCount loop allocates one dataKey buffer per call and mutates its trailing 8 bytes per uid. Two distant sites captured the slice header rather than the bytes: (a) ReadPostingList's l.key = key aliased the caller's buffer, and saveInCache stored a copyList that shared the alias — by the time async rollup pulled the cached list and built kv.Key = alloc.Copy(l.key), the bytes had collapsed to whatever uid was processed last, so a BitCompletePosting (with WithDiscard) got written to a different list's key, silently overwriting an unrelated reverse list with this list's contents; (b) addKeyToBatch appended the slice header verbatim, so all queued rollup targets from one ProcessCount goroutine ended up pointing at the same final-iteration key. Both fixed by taking ownership of the bytes. Legacy runMutation allocates a fresh key per call so it never aliased anything; the bug was only visible when a caller deliberately reused one buffer the way ProcessCount does. This was the cause of systest/21million/live's consistent failure.

Feature flag — threshold instead of on/off

Commit	What
c8f5fbcf3	Replace MutationsUsePipeline bool with MutationsPipelineThreshold int. A mutation runs through the pipeline only when threshold > 0 && len(m.Edges) >= threshold; 0 = pipeline off, 1 = always on, larger values opt small interactive mutations out (benches show ~2× slowdown for ≤10-edge txns). CLI: dgraph alpha --feature-flags="mutations-pipeline-threshold=200".
86ad1eea9	Flip the default to 1 so test suites exercise the pipeline path on every mutation. Reset to 0 before merging once the canonical CI has been green for a soak window.

Tests / regressions

Commit	What
7c95203b8	Four in-process regression harnesses for [uid] @reverse @count: single-txn, multi-batch sequential, multi-batch multi-predicate (parallel goroutines), and multi-batch concurrent through a fakeOracle conflict-checking harness. None reproduce the buffer-aliasing bug fixed in 0a9ffc1 — that needs the async rollup path that only fires on a real running cluster — but they pin down the in-memory mutation path's reverse-list bookkeeping.
d0c631a16	New systest/21million/queries/query-073: an exhaustive count(~genre) check across every one of the 592 non-empty Genre entities, with the exact expected count pinned. Catches any future regression that mis-routes reverse edges, which the existing query-017 only does indirectly.

Benchmarks

In-process matrix (worker/pipeline_bench_test.go, included in 792923279) on Apple M4 Pro, 30 iterations:

Config	Legacy ns/op	Pipeline ns/op	Pipeline / Legacy
1 pred × 1 edge	39,753	89,581	2.25× slower
10 preds × 1 edge	190,672	451,854	2.37× slower
1 pred × 100 edges	654,775	727,931	1.11× slower
50 preds × 100 edges	31,630,750	29,959,582	1.06× faster
10 preds × 100 edges	9,448,342	7,441,912	1.27× faster
1 pred × 1000 edges	5,753,151	4,957,482	1.16× faster
10 preds × 1000 edges	69,132,992	58,908,736	1.17× faster
10 preds × 1000 edges, no index	31,308,575	20,238,396	1.55× faster
50 preds × 1000 edges	403,867,619	375,424,165	1.07× faster

Crossover is around 100 total edges.

End-to-end live load against the 1M dataset (1million.rdf.gz, official 1M schema with mixed [uid] @reverse @count, [uid] @count, datetime @index(year), string @index(hash, term, trigram, fulltext) @lang, geo @index(geo), string @index(exact) @upsert):

mutations-pipeline-threshold	Avg time	Avg N-Quads/s
0 (legacy)	~14.3 s	~77K
1 (always-on pipeline)	~9.5 s	~116K

~1.50× faster on a realistic bulk-shaped workload — the case the per-predicate runner was designed for.

Test results

With mutations-pipeline-threshold=1:

• posting/, worker/, types/, schema/ — all unit tests pass, including the new TestPipelineCountIndexConcurrent and four TestPipelineReverseListCount* harnesses under -count=20 -race.
• ./t --suite=systest-baseline — all 13 packages green: systest, systest/acl/restore, systest/audit, systest/audit_encrypted, systest/backup/filesystem, systest/cdc, systest/cloud, systest/export, systest/group-delete, systest/loader, systest/multi-tenancy, systest/mutations-and-queries (66 tests including the previously-failing TestHasDeletedEdge), systest/plugin. Total wall clock 7m38s.
• ./t --suite=core,integration — acl (150.8s) green plus four large systest/vector tests (TestVectorIncrBackupRestore 124s, TestVectorBackupRestore 72s, TestVectorBackupRestoreDropIndex 23s, TestVectorBackupRestoreReIndexing 110s). Suite then died on a pre-existing Mac-only bug in dgraphtest (commit ef6f27da79 "Fix running tests on mac" on main) that's unrelated to the pipeline.
• systest/21million/live TestQueries — green. Sweep across all 764 Genre entities reports count(~genre) == count(forward edges) for every genre. Pre-fix the same sweep showed sets of 4 mismatched genres with thousands of stale or missing reverse entries each load, with the specific affected genres varying run-to-run.

Out of scope, worth noting

• The bulk loader (dgraph/cmd/bulk/) doesn't import any of the pipeline machinery. The "speed up bulk loader" framing in the PR's lineage doesn't survive code inspection — the bulk loader has its own map-reduce path.
• TestCount in worker/sort_test.go is left t.Skip()'d with a clear reason (harness lacks Oracle conflict checking; both legacy and pipeline fail it).

Recommended deployment

• Reset the default to 0 before merging.
• Operators who want the speedup on bulk-only workloads can opt in with --feature-flags="mutations-pipeline-threshold=200" (or whatever the per-mutation crossover is for their workload).
• Setting 1 always-on is fine for systest CI runs and is a useful "ramp-up" knob.