perf(core): per-predicate mutation pipeline with intra-mutation parallel apply (revives #9467) - #9762
Open
rahst12 wants to merge 40 commits into
Open
perf(core): per-predicate mutation pipeline with intra-mutation parallel apply (revives #9467)#9762rahst12 wants to merge 40 commits into
rahst12 wants to merge 40 commits into
Conversation
Adds an alternative path through `applyMutations` that fans the edges of a single mutation out by predicate and processes each predicate's batch in its own goroutine, exposing parallelism the legacy serial-by-edge path cannot. The per-predicate runners drive the full mutation lifecycle (scalar/list, reverse, count, tokenized index, vector index) through new helpers in posting/index.go: `MutationPipeline.Process` → `ProcessPredicate` → `ProcessSingle` / `ProcessList` / `ProcessVectorIndex`, with `InsertTokenizerIndexes`, `ProcessReverse`, and `ProcessCount` shared across the predicate-shaped paths. Supporting infrastructure: - `posting.Deltas` — per-txn delta store split into a sharded raw-bytes map and a per-predicate `indexMap` for batched index writes. - `types.LockedShardedMap` — generic sharded RWMutex map used by Deltas and the in-memory index aggregation. - New mutation-time helpers on `posting.Txn` (`AddDelta`, `GetScalarList`, `addConflictKey`, `addConflictKeyWithUid`). - Test scaffolding in `worker/mutation_unit_test.go`, `worker/sort_test.go`, and `worker/draft_test.go`. This commit squashes the original WIP series authored by Harshil Goel (commits 978a0d4…41d6445ce on the abandoned branch) plus the merge into current `main` and the mechanical clean-ups required to compile against it: import paths `hypermodeinc/dgraph` → `dgraph-io/dgraph`, license headers, a `posting/mvcc_test.go::TestRegression9597` fix-up for the `LocalCache.deltas` map → `*Deltas` refactor, and a stray doc reference in TESTING.md. The pipeline ships gated and disabled; the on/off knob and subsequent correctness fixes follow in later commits in this series. Co-Authored-By: Matthew McNeely <matthew.mcneely@gmail.com> Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Removes the unconditional debug prints scattered through posting/ during the original WIP work — they fired on every read, mutation, rollup, and commit. None of them were guarded by a verbosity flag, so under load they would have produced megabytes of stdout noise per second. Sites stripped: - posting/lists.go: READING / READING SINGLE / GETTING KEY FROM DELTAS - posting/index.go: TOKENS, LOCAL MAP, INSERTING INDEX, UPDATE INDEX, ERRORRRING, "Inserting tokenizer indexes ... took" - posting/mvcc.go: COMMITTING (and unused fmt import) - posting/list.go: "Buidlding committed uids", "Setting mutation", PrintRollup helper (called once internally, never elsewhere) Left in place: printTreeStats() in index.go, which is already gated by the DEBUG_SHOW_HNSW_TREE env var and is an intentional opt-in HNSW debug helper. No behavior change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Two large dead-code blocks left over from the original WIP: - posting/index.go: ~175 lines of an alternate InsertTokenizerIndexes implementation, fully commented out. The live implementation directly above it supersedes it; keeping the commented variant just made the file harder to follow. Also drop the scattered "//fmt.Println(...)" leftovers next to live code. - worker/draft_test.go: BenchmarkProcessListIndex was added entirely commented out and references methods (DefaultPipeline, ProcessListWithoutIndex, ProcessListIndex) that don't exist on MutationPipeline. If we want a benchmark for the pipeline, we should write one against the real API. No behavior change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Replace the hardcoded \`featureFlag := true\` in applyMutations with a real superflag knob, defaulted off: - Add WorkerOptions.MutationsUsePipeline (bool) in x/config.go. - Extend the feature-flags superflag with mutations-use-pipeline=false and wire alpha to populate WorkerConfig.MutationsUsePipeline from it. - worker/draft.go applyMutations now branches on x.WorkerConfig.MutationsUsePipeline; default false routes mutations through the legacy path, preserving current behavior. Tests can opt into the new pipeline by setting x.WorkerConfig.MutationsUsePipeline = true. CLI usage: dgraph alpha --feature-flags="mutations-use-pipeline=true" Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
TestCount is t.Skip()'d on the branch, but the reason wasn't recorded. Investigation: the test launches concurrent transactions sharing entity uids and bypasses the Oracle's conflict-checking commit path — it just calls CommitToDisk() directly with disjoint commit timestamps. Both the legacy AddMutationWithIndex path and the new mutation pipeline fail it identically: with two threads adding edges to the same subject's [uid] @count predicate, neither path can serialize @count updates without real txn conflicts, so the count index ends up inconsistent and many subjects are missing from count(N). This is expected without conflict checking — the unit harness can't exercise the safety the Oracle provides. Re-enable when we wire either: (a) Oracle.WaitForTs/IsAborted into the harness, or (b) this test through worker.applyMutations() (which does invoke the Oracle conflict path). Single-thread TestCount passes, so the per-predicate pipeline's own count logic is correct in the absence of contention. The existing TestStringIndexWithLang covers the multithreaded happy path with disjoint uids. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…ssCount Bug: scalar @count writes were nondeterministically losing data under concurrent transactions. Roughly half the deltas committed by the new mutation pipeline contained only a [DeleteAll] posting and no Set, so reads at maxTs returned an empty value list. Root cause: in ProcessSingle, handleOldDeleteForSingle appends a synthetic Del-of-old-value to postings[uid] alongside the user's Set, so InsertTokenizerIndexes / ProcessReverse / count diffing can see the prior value. ProcessCount then iterates the postings and calls list.updateMutationLayer(post, singleUidUpdate=true, ...) on each. For non-Lang scalar predicates fingerprintEdge returns math.MaxUint64, so the synthetic Del and the user Set both have Uid == math.MaxUint64. The first iteration (Set new) leaves mutationMap.currentEntries = [DeleteAll, Set new]; the second iteration (Del old) finds the Set we just inserted via findPosting and applies updateMutationLayer in singleUidUpdate mode, which unconditionally rewrites currentEntries to [DeleteAll] (the Del branch never appends mpost) — wiping the new value. Fix: in ProcessCount, when iterating a !isListEdge predicate's postings, if the list contains a Set/Ovr posting, treat any Del as synthetic and skip it for the data-list update. Standalone user Dels (no accompanying Set) are still applied. Index/reverse/count diffing already happen before ProcessCount runs and aren't affected. Repro: TestPipelineCountIndexConcurrent in worker/sort_test.go is a new conflict-aware in-process harness that mirrors the systest TestCountIndexConcurrentSetDelScalarPredicate. It runs 200 contending transactions setting <0x1> <name> "name<rand>" against a "string @index(exact) @count" schema with a fakeOracle that implements the same hasConflict algorithm as dgraph/cmd/zero/oracle.go. Pre-fix the test fails roughly 50% of runs with an empty data list and the wrong count buckets; post-fix it is stable across 20+ -count iterations and under -race. Existing tests (TestScalarPredicateIntCount, *RevCount, *Count, TestSingleUidReplacement, TestDeleteSetWithVarEdgeCorruptsData, TestStringIndexWithLang, TestMultipleTxnListCount, TestGetScalarList, TestDatetime) all pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
When loading via dgraph live (or any mutation source whose uids span
the full uint64 range, including xidmap-assigned uids), the
per-predicate pipeline hung indefinitely on the very first batch with
zero forward progress. A goroutine dump showed the dispatcher
goroutine wedged on \`chan send (nil chan)\` at the line:
chMap[int(uid)%numGo] <- uid
uid is uint64. Casting directly to int produces a negative value for
uid >= 2^63, so int(uid)%10 can be in [-9, -1]. chMap[-3] returns the
zero value for a chan uint64, which is a nil channel; sending on a
nil channel blocks forever.
The 10 worker goroutines (also created here) were idle on
\`for uid := range uids\` since no uids ever reached them, so the
parent \`wg.Wait()\` and the surrounding errgroup never returned.
applyMutations therefore never released the txn, the alpha's old-txn
abort loop kept retrying every minute, and live-load showed
"Txns: 0 N-Quads: 0" indefinitely.
Fix: hash unsigned, then cast: \`chMap[int(uid%uint64(numGo))]\`.
Verified end-to-end with the live loader against the 1million.rdf.gz
benchmark dataset (1,041,684 n-quads, schema mixes [uid] @reverse
@count, [uid] @count, datetime @index(year), string @index(...) @lang,
geo @index(geo), string @index(exact) @upsert):
legacy : 13.85s / 14.74s (avg ~14.3s, ~77k n-quads/s)
pipeline : 9.65s / 9.36s (avg ~9.5s, ~116k n-quads/s)
That is ~1.50x faster on a realistic multi-predicate, multi-index
workload — i.e. the case the per-predicate runner pipeline is built
for.
Also adds worker/pipeline_bench_test.go: in-process Go benchmarks
comparing legacy runMutation vs newRunMutations across a matrix of
(predicates, edges-per-predicate, indexed/non-indexed) shapes. They
show the pipeline loses ~2x on tiny mutations (1-10 edges) and wins
1.2x-1.55x on bulk (10 preds x 100+ edges, indexed or not), which is
why the feature flag stays default-off and the live-loader speedup
above is the right place to evaluate this work.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The benchmark matrix in worker/pipeline_bench_test.go showed the
pipeline loses ~2x on small mutations (≤10 edges total) and wins
~1.5x on bulk (live-loader sized: 1000 edges per txn across many
predicates). A binary on/off flag forces an all-or-nothing choice,
penalising whichever side of that crossover the workload spends most
time on.
Replace MutationsUsePipeline (bool) with MutationsPipelineThreshold
(int):
threshold = 0 -> never use the pipeline (default; legacy behavior)
threshold = 1 -> always use the pipeline (any txn with ≥1 edge)
threshold = N -> use the pipeline only when len(m.Edges) >= N
The threshold compares against total edges in the proposal. From the
benches the crossover is around 100; the live-loader 1M dataset uses
~1000 edges per txn, so anything from 100-1000 will engage the
pipeline only on bulk-shaped mutations and leave small interactive
mutations on the legacy serial path.
Wiring:
- x.WorkerConfig.MutationsPipelineThreshold (int) replaces the
bool field.
- feature-flags superflag: "mutations-pipeline-threshold=0".
- alpha/run.go reads it via featureFlagsConf.GetInt64.
- worker/draft.go applyMutations branches on
`t > 0 && len(m.Edges) >= t`.
Verified end-to-end against the live-loader benchmark
(1million.rdf.gz, official 1M schema):
threshold=0 : 13.56s, 80,129 N-Quads/s (legacy, matches baseline)
threshold=1 : 9.92s, 115,742 N-Quads/s (always-on, matches prior)
CLI usage:
dgraph alpha --feature-flags="mutations-pipeline-threshold=200"
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Engages the per-predicate mutation pipeline by default so that systest runs (and any other test suites that don't override feature-flags) exercise the pipeline path on every mutation, not the legacy serial path. Threshold of 1 means "any mutation with ≥1 edge takes the pipeline" — i.e. always on. This is a deliberate ramp toward shipping the pipeline. Operators who want to opt small interactive mutations out of the pipeline (where benches showed ~2x slowdown for ≤10-edge txns) can set a higher threshold: dgraph alpha --feature-flags="mutations-pipeline-threshold=200" To turn the pipeline fully off, set 0. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The original branch commit 41d6445 ("fixed some bug") replaced the IsEmpty(readTs) call in IterateDisk with a hardcoded `false`, forcing every key found by the iterator to be reported as non-empty. That broke has(<predicate>) for any uid whose value had been removed via star-deletion (<uid> <pred> *): the data list still exists in badger with a DeleteAll marker on top, but the live posting list is empty at readTs — IsEmpty returns true and the uid should be skipped. Surfaced by systest TestSystestSuite/TestHasDeletedEdge in systest/mutations-and-queries: 3 nodes are created with <end> "", one is star-deleted, follow-up has(end) is expected to return 2 uids. With IsEmpty stubbed to false it returned 3. No comment was left on the original change. Restoring the call. The mutations-and-queries package is fully green with this in place (66/66 tests pass including TestHasDeletedEdge); if a real underlying issue motivated the original disable we'll chase it with a real diagnosis instead of silently dropping a safety check. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…p stale READING debug prints
Two bugs surfaced by graphql/e2e/auth's TestOrderAndOffset (cleanup
mutation \`mutation DelTask { deleteTask(filter: {}) }\` crashed the
alpha mid-request, causing the test client to see EOF on POST):
1. posting/index.go ProcessSingle SIGSEGV at line 675.
GraphQL deleteTask cleanup expands into multiple Del edges per
entity (one per predicate the entity has — uid, type, list edges,
etc.). When two Del edges to the same uid land in one transaction's
batch, the second iteration through ProcessSingle's per-edge loop
does:
pl, exists := postings[uid]
if exists {
if edge.Op == DEL {
oldVal = findSingleValueInPostingList(pl)
if string(edge.Value) == string(oldVal.Value) { ... }
^^^^^^ nil deref
}
}
findSingleValueInPostingList only returns Set postings; if the
accumulated list holds only Dels (from the prior iteration), it
returns nil and we panic dereferencing oldVal.Value.
Two fixes here:
- Guard the deref: \`if oldVal != nil && string(...) == ...\`.
- Move \`var oldVal *pb.Posting\` inside the loop. It was declared
at function scope, so a stale value from one edge could bleed
into the nil-guarded branch for a different uid on a later
iteration. Per-edge scope makes the intent explicit.
2. worker/task.go: two leftover \`fmt.Println("READING SINGLE", ...)\`
and \`fmt.Println("READING", ...)\` calls in the value-postings
read path. Same class of debug spew Phase 1B stripped from
posting/, missed because that sweep didn't include worker/.
Removed both. Safe — they were unconditional prints on every
query value read.
The graphql/e2e/auth and graphql/e2e/auth/debug_off packages now both
pass in 30s. \`./posting/\` and \`./worker/\` unit tests still green.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…addKeyToBatch The 21million live-load systest failed with `~genre @filter(uid(F))` returning wrong films for some genres — corruption pattern was that one genre's reverse list contained another genre's films verbatim, while a third genre's reverse list was missing thousands of entries. Different genres were affected on each load, but always: legacy runMutation produced correct reverse counts; the per-predicate pipeline produced corrupt ones. Root cause: the pipeline's ProcessCount per-uid loop allocates one key buffer per ProcessCount call and mutates its trailing 8 bytes each iteration via `binary.BigEndian.PutUint64(dataKey[len-8:], uid)`. Two distinct sites then captured the slice header rather than the bytes: 1. ReadPostingList's `l.key = key` aliased the caller's buffer. saveInCache stores a copyList of the freshly-read list, and copyList sets `key: l.key` — also an alias. The cached list's key field therefore points at a buffer that the pipeline keeps mutating; by the time the async rollup path retrieves the cached list and runs `kv.Key = alloc.Copy(l.key)` to build the rolled-up KV, the bytes are whatever the LAST iteration left behind. Rollup then writes a BitCompletePosting with WithDiscard() to the wrong key, overwriting an unrelated reverse list with this list's (rolled-up) contents. 2. ReadPostingList's defer `IncrRollup.addKeyToBatch(key, ...)` appended the slice header verbatim to the rollup queue. Every queued entry from one ProcessCount goroutine ended up pointing at the same shared dataKey buffer; by the time the rollup goroutine processed the batch the bytes had collapsed to the final iteration's uid, redirecting many distinct rollup targets to the same key. Both sites fixed by taking ownership of the bytes (`append nil` / explicit `make+copy`). Legacy runMutation hits ReadPostingList too but allocates a fresh key per call, so it never aliased anything; the bug is only visible when a caller deliberately reuses one buffer across many uids the way ProcessCount does. Verified end-to-end on the systest/21million/live load with mutations-pipeline-threshold=1: a fresh load + sweep across all 764 Genre entities now reports `count(~genre) == count(forward edges)` for every genre. Pre-fix the same sweep showed 4 mismatched genres with thousands of stale or missing reverse entries; on a different load it showed Documentary missing 23,325 of its 31,370 reverse entries while Children's/Family had 1,435 stale Crime-Thriller entries. Unit tests in posting/ and worker/ still pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Added in pursuit of the 21million live-load failure that turned out to be the dataKey buffer-aliasing bug fixed in 0a9ffc1. None of these tests reproduce that specific corruption — it needs the async rollup path that only triggers on a real running cluster — but they pin down what the per-predicate pipeline does correctly for the non-rollup cases: - TestPipelineReverseListCount: one transaction, multiple subjects pointing at multiple objects on a [uid] @reverse @count predicate. Verifies forward and reverse lists are both complete. - TestPipelineReverseListCountMultiBatch: 50 subjects x 20 objects spread across 143 sequential transactions in batches of 7. - TestPipelineReverseListCountMultiPred: 30 subjects x 12 objects across 3 distinct list-uid + reverse + count predicates, with shuffled edges and small batches so the per-predicate pipeline goroutines for each predicate run in parallel inside Process(). - TestPipelineReverseListCountConcurrent: same shape as the multi-batch case but with 10 worker goroutines submitting batches in parallel through the fakeOracle conflict-checking harness. All four pass cleanly. They guard against regressions in the in-memory mutation path's reverse-list bookkeeping; the rollup path that the 0a9ffc1 fix actually targets is exercised by the systest/21million/live integration test. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…empty genres A comprehensive `count(~genre)` check across every Genre entity in the 21million dataset. The query asks for each genre's name and the size of its reverse posting list, sorted by name; the fixture pins the expected count for all 592 genres that have at least one film. Motivation: the per-predicate mutation pipeline had a buffer-aliasing bug (fixed in 0a9ffc1) that produced wrong reverse-list contents on a different small subset of genres each load — sometimes Documentary lost most of its 31,370 entries, sometimes Children's/Family gained ~1,500 spurious Crime-Thriller entries, sometimes Backstage Musical and Indie film were the affected pair. The existing query-017 (Taraji-films-by-genre) only catches it when that specific actor's films happen to land on a corrupted genre, and query-016 / query-044 only test genres above 30,000 films. This new query exercises every genre's reverse list and pins the exact expected count, so any regression that mis-routes reverse edges for any genre will fail it directly. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Two related bugs in the per-predicate mutation pipeline that surface as stale index entries under TestStringIndex (systest/bgindex): 1. Process() (the star-delete fast path) was passing the caller's bare context to handleDeleteAll. schema.State().IsIndexed / IsReversed / HasCount only consult the pending mutSchema (the new schema with the index being built) when the context carries the isWrite flag. The legacy runMutation path calls schema.GetWriteContext at the top; the pipeline didn't. During a background index build, every star-delete saw isIndexed=false and skipped addIndexMutations(DEL) for the prior value, leaving stale uids in the index permanently. Lifting GetWriteContext into Process covers both the star-delete path and the later predicate-pipeline goroutines. 2. ProcessSingle's data-list write was passing the unfiltered postings list to AddDelta. handleOldDeleteForSingle appends a synthetic Del(oldVal) alongside the user's Set(newVal) so InsertTokenizerIndexes and ProcessReverse can emit Del-of-old entries. For scalar non-list non-lang predicates both postings share Uid == math.MaxUint64, and at read time pickPostings' equal-ts tie-break falls back to Go's unstable sort.Slice while setMutationAfterCommit overwrites committedUids[mpost.Uid] in append order. Either way Del can clobber the new Set and the data list reads as "no value." ProcessCount already strips the synthetic Del via skipSyntheticDel; mirror that filter in ProcessSingle's main data-list path. systest/bgindex now passes (TestStringIndex, TestReverseIndex, TestCountIndex, TestParallelIndexing). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HNSW Insert performs multi-key read-modify-write across the entry-pointer
key, the per-level edge lists, and back-edges into the neighbors of the
new node. Per-list locks make each key's update atomic, but the
cross-key sequences ("read entry, lock entry, read neighbor, modify
neighbor, append back-edge to neighbor's list…") are not. Running 10
worker goroutines concurrently against the same txn cache lets updates
stomp on each other, leaving nodes that have a data-list entry but are
unreachable from the entry point.
Surface symptom: similar_to(k=N) returns fewer than N hits even though
all N vectors committed to the data list. TestVectorTwoTxnWithoutCommit
reliably reproduced this on Linux CI (and not on the macOS dev box,
likely a scheduling artifact).
Legacy applyMutations arrives at single-threaded vector handling
indirectly via x.DivideAndRule: for any num < 256 the helper rounds
numGo down to 1, so a 5-edge vector mutation runs serially on main and
the bug never surfaces there. Mirror that here by dropping
numThreads to 1 — correctness over within-txn parallelism for vector
predicates. Cross-txn parallelism is unaffected.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1. Process() channel-send deadlock when a predicate goroutine errors
before the buffered channel drains. The bare `pred.edges <- edge`
send had no select on egCtx.Done(), so once the 1000-slot buffer
filled and the reader had exited the dispatcher blocked forever
and eg.Wait() was never reached. Guard the send with a select +
labeled break, still close every per-predicate channel and call
eg.Wait() so the goroutine's real error wins over context.Canceled.
2. ProcessSingle compared `string(edge.Value) == string(oldVal.Value)`
on uid postings whose Value is nil. nil==nil always matched, so a
same-batch DEL wrongly overwrote an accumulated SET in the
exists=true branch and synthesised a spurious DEL delta in the
exists=false branch. Introduce a sameTarget closure that compares
Uid/ValueId for uid postings and Value for scalars; use it at both
call sites. handleOldDeleteForSingle picked up the same fix on its
committed-vs-new comparison.
3. Same-batch SET after a star-delete restored the prior committed
values: handleDeleteAll wrote [{deleteAll}] into currentEntries via
setCurrentEntries; the pipeline's later AddDelta for the
accumulated SET overwrote currentEntries via the same path and
reset deleteAllMarker. populateDeleteAll at read time then found
no deleteAll and the previously committed friends survived.
Preserve any existing deleteAll posting when setCurrentEntries
overwrites — mirroring legacy's append-via-insertPosting behaviour.
4 & 5. ProcessList lost index emissions for scalar @lang predicates
in two ways: (a) it never read the committed posting list to emit
a DEL of the old value when SET-replacing — only the SET for the
new value was emitted, so the old "Alice"@en index entry persisted
after SET "Alicia"@en; (b) for scalar @lang mpost.Uid is
fingerprint(Lang), so SET-new + DEL-old@same-lang collided in the
per-uid slot inside mutations[uid] — the second write erased the
first and only one of the two index emissions survived. Legacy
avoids both because addMutationHelper generates index mutations
per edge before the same-uid overwrite takes effect.
Fix: maintain rawIndexPostings alongside mutations — every batched
posting is appended in arrival order, bypassing insertPosting's
collapse. New helper handleOldDeleteForList iterates the cached
posting list via l.Iterate, finds each committed SET matching a
lang fingerprint touched in this batch, and appends a synthetic
DEL of the old value to rawIndexPostings. Skipped when
info.isList=true because list-typed predicates append rather than
replace. Pass rawIndexPostings to InsertTokenizerIndexes;
postings (the collapsed view) is still used for the data write
and ProcessReverse, where collapse is correct.
6. When a SET and a DEL for the same uid tokenized into the same
index bucket (e.g. case-insensitive tokenizer where "Apple" and
"APPLE" share token "apple"), InsertTokenizerIndexes' last-write-
wins insertPosting on indexGenInThread left whichever op landed
last as the winner. For the scalar @index [DEL old, SET new] case
handleOldDeleteForSingle appended a third synthetic DEL, so
postings had three entries and the previous `isSingleEdge &&
len==2` reversal heuristic no longer fired — the resulting
DEL-SET-DEL emission order left the uid removed from a bucket she
should still belong to. Replace the order-dependent reversal with
a stable Del-first sort of the per-uid posting list. SETs land in
indexGenInThread after DELs for the same bucket so the SET wins;
for different buckets the sort is a no-op on correctness.
Adds posting/pipeline_test.go as the regression harness — 25 tests
covering 13 scenarios, each with a TestLegacy* mirror so any future
divergence between the pipeline and legacy runMutation paths is
caught directly. The header comment documents the startTs/commitTs
band layout used to isolate tests.
posting/ unit tests fully green; all Legacy* mirrors continue to
pass — these fixes are pipeline-specific.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1. handleOldDeleteForSingle cleared postings[uid] to an empty
PostingList whenever the new SET value matched the committed
value. The intent was "value unchanged, skip the write," but
clearing also bypassed InsertTokenizerIndexes / ProcessReverse —
so a SET that needed to populate a newly-added @index or
@reverse (data committed before the alter, then re-SET with the
same value under the new schema) emitted no index/reverse entry.
TestSchemaMutationIndexRemove caught this: query for
anyofterms(name, "Alice") returned [] after a same-value SET on
a term-indexed predicate. Legacy passes because runMutation
calls addIndexMutations per SET edge regardless of value change.
Fix: in both same-value branches (REF and scalar), continue
without clearing postings[uid]. The SET stays so the index /
reverse passes can populate the entry — idempotent when it
already existed, load-bearing when the index was newly added.
2. ProcessSingle's data-write loop called addConflictKey + AddDelta
for every uid in the postings map, including uids whose
accumulated PostingList was empty. An empty list happens when
setPosting was never invoked — e.g. a DEL whose value did not
match the committed value (a no-op per Dgraph semantics) or the
same-value SET no-op after fix dgraph-io#1. AddDelta with an empty input
marshals to empty bytes and then calls list.setMutation on the
txn-cached List, which routes through setCurrentEntries and
resets the mutable layer (currentEntries cleared, deleteAllMarker
reset). Any reader holding that cached List then reads the data
list as empty. CommitToDisk filters empty-PostingList deltas
before badger writes (mvcc.go ~L318), but UpdateCachedKeys still
propagates them to the global MemLayer via setMutationAfterCommit,
planting committedEntries[commitTs] = emptyPL with CommitTs set.
TestDeleteScalarValue and TestUpsertDeleteWrongValue surfaced
this: a DEL with a wrong value (val(updated_amt) where
updated_amt = amt + 1) deleted the data anyway. ProcessList
(~L401) and ProcessReverse (~L466) already had the empty-pl
guard; ProcessSingle did not.
Fix: at the top of ProcessSingle's data-write loop, skip uids
with an empty PostingList — matches the sibling paths.
Regression coverage in posting/pipeline_test.go:
- TestMutationPipelineSetThenAddIndexThenResetMirrorsAlphaFailure
- TestMutationPipelineSameValueSetKeepsIndex
- TestMutationPipelineDelWithWrongValueIsNoOp
- TestMutationPipelineDelWrongValueTwoUidsCloseMatch
All with Legacy mirrors that pass before the fix; the
TestMutationPipeline* variants fail without the fix.
…sList forward write Builds on the per-predicate mutation pipeline (dgraph-io#9467). Three additive, flag-gated changes that make a single hot/dominant predicate's apply actually parallelize: 1. Proportional goroutine budget across predicates (allocateWorkers): mutations-pipeline-goroutines=N distributes N workers by edge count (largest-remainder), so a dominant predicate gets >1 worker. 2. Auto mode (=-1): budget = min(GOMAXPROCS*fraction, edges/minEdgesPerWorker), derived at runtime. Tunables mutations-pipeline-goroutines-fraction (1.0) and mutations-pipeline-min-edges-per-worker (256). 3. Parallel forward data-write in ProcessList (the merge-light <pred,uid> pass), matching the existing ProcessSingle split, so [uid]/@reverse predicates' forward write parallelizes. ProcessReverse stays SERIAL. Default is off (0) => byte-identical to the current one-goroutine-per- predicate path. Byte-identical equivalence + conflict-key-set tests pass under -race for scalar, [uid] @reverse, and AUTO==fixed. Benchmarks (8-core dev box, ~20k-edge batch) included. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…chemas Confirms the intra-predicate goroutine budget (proportional, auto, and the ProcessList parallel forward write) produces byte-identical committed state and identical conflict-key sets vs. the legacy one-goroutine-per-predicate path, across string (exact/hash/term/fulltext/trigram), int(+count), float, dateTime, bool, geo, [uid]@reverse(+count), uid@reverse, [string], @upsert, @noconflict, and @lang predicates. Generic: scans every committed Badger key (data/index/reverse/count) per predicate and compares dumps. 36 subtests (12 schemas x {fixed8, fixed32, auto}); race-clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
matthewmcneely
self-requested a review
June 23, 2026 21:55
Author
|
TODO: The flags need consolidated to something more manageable. |
… grant ProcessReverse built reverseredMap[targetUid] then wrote each target serially via AddDelta, which holds the GLOBAL txn.cache.Lock() across proto.Marshal. With many predicates applying concurrently at L1 they all convoy on that one mutex. A live goroutine dump (25 [uid] @reverse predicates, 20k triples/txn) showed 61% of pipeline goroutines parked in ProcessReverse and 86% of those blocked on that lock. Two changes: 1. Split "use the lock-free store" from "split across k goroutines". Both decisions were tied to workers > 1, but allocateWorkers hands out mostly 1-worker grants when the budget is below ~2x the predicate count (the shipped default of 30 with 25 predicates yields {1:20, 2:5}), so nearly every predicate fell back to the global lock. Now any enabled budget uses concStore; workers == 0 keeps the legacy locked AddDelta so "budget off == byte-identical legacy" still holds. 2. Above reverseParallelMinTargets (256) distinct targets, partition the target-uid keyspace across the predicate's worker grant. reverseredMap is keyed BY target, so each <~pred,targetUid> key has exactly one writer (I1) — the same disjointness that makes ProcessList's forward write safe. Below the threshold the map holds a few hot targets with large lists; sharding that shape measured as a net loss, so it stays serial. concStore reproduces AddDelta(..., true, true) exactly: prior-delta prepend (a *Txn spans proposals, so dropping it silently loses an earlier proposal's reverse postings) and unconditional sort/dedup, reading only the sharded delta map — never (*Deltas).Get, which also touches indexMap under cache.Lock. Per-worker key buffers are required: addConflictKeyWithUid fingerprints the full buffer including the 8 uid bytes. The reverseredMap BUILD stays serial (many-to-one, shared edge struct); parallelising it needs a merge stage that costs more than it saves. The info.count path returns to ProcessCount before any of this, unchanged. Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn, 64 threads, badger_write_bytes_user: budget=30 (default): 9.63 -> 10.58 MB/s (+9.9%) budget=-1 (auto): 9.68 -> 11.29 MB/s (+16.6%) Reverse-write parking drops 61% -> 37%; global cache.Lock blocking drops 86% -> 41%. Tests: byte-identical committed state and conflict-key sets across budgets {0,8,32,auto} for 13 schemas, incl. two new high-cardinality rows covering the ProcessList and ProcessSingle legs of the parallel path, plus a cross-proposal regression test for the addToList prepend (verified to fail when the prepend is removed). Full posting package green under -race. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
ProcessList and ProcessSingle tied two independent decisions to the same
`workers > 1` test: whether to use the lock-free store, and whether to
split the uid space across goroutines. allocateWorkers hands out mostly
one-worker grants whenever the budget is below ~2x the predicate count —
the shipped default of 30 with 25 predicates yields {1:20, 2:5} — so 20 of
25 predicates fell back to AddDelta, which holds the GLOBAL
txn.cache.Lock() across proto.Marshal. With every predicate goroutine at
L1 doing that, they convoy on one mutex and the parallel forward write
added in 37c834a never engages for most of the batch.
Split the two decisions, mirroring 4a02d8e's fix for ProcessReverse:
workers == 0 keeps the legacy locked AddDelta so "budget off ==
byte-identical legacy" holds; any enabled budget uses the lock-free store
(concStore for ProcessList, which must reproduce AddDelta's addToList
prepend and info.isUid sort/dedup; plain AddDeltaConcurrent for
ProcessSingle, whose serial call already passes addToList=false,
doSortAndDedup=false). The k-way split still requires workers > 1.
Single-writer safety is unchanged: <pred,uid> forward keys are one-to-one,
and x.generateKey encodes attrLen plus a per-type discriminator byte, so
two predicates cannot produce the same key string.
Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse
predicates, 20k triples/txn, 64 threads, badger_write_bytes_user, 2 runs
per arm:
stock 9.63 MB/s
+ reverse (4a02d8e) 10.58 MB/s (+9.9%)
+ this commit 11.44 MB/s (+18.8%)
Budget stops mattering once the lock-free store is reachable at any grant:
auto measures 11.38 MB/s, within noise of the default's 11.44.
The global cache lock disappears from the blocking profile entirely —
goroutines parked on it go 86.4% (stock) -> 41-52% (reverse only) -> 0%,
and runnable goroutines rise to 57.9%. addConflictKeyWithUid's single
global txn.Lock() is now 71.2% of parked pipeline goroutines and is the
next bottleneck. CPU remains at 8% of 64 vCPU: processApplyCh applies one
Raft entry at a time, so a single Alpha cannot saturate a large box
regardless of intra-transaction parallelism.
Full posting package green under -race.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
After 4a02d8e and 612ae8c removed the txn.cache.Lock() convoy, Txn.conflicts became the only remaining mutex contention in the apply path: on a 32-core ingest run 25% of pipeline goroutines were blocked on it and 46% were running inside addConflictKeyWithUid. The map insert was never the problem. A line-level profile attributed 80% of that function's 6.80% of total CPU to acquiring and releasing the mutex (8.75s + 5.33s of 17.86s cum) and only 15% to the map writes (2.71s), with every worker in every predicate contending one lock once per key. So batch instead of shard. Each Process* accumulates its conflict keys in a conflictBuf and flushes them under a single txn.Lock(); parallel workers get their own buffers, merged after wg.Wait(), and never touch txn.Mutex at all. Sharding the container was considered and rejected: it only attacks the 2.71s of inserts (~1% of total CPU) while changing Txn.conflicts' type, nine test sites, FillContext, and NumShards — which is shared with the delta and index maps. Keys are expanded EAGERLY into uint64. Every call site reuses one scratch key buffer across iterations, so a buffer retaining the []byte and fingerprinting at flush time would hash the last uid's bytes for every entry — wrong keys, no panic, nothing for -race to catch. In InsertTokenizerIndexes the flush defer is registered ABOVE cache.Lock() so LIFO runs it after cache.Unlock(); registering it below compiles, passes every test, and silently keeps the txn.Lock()-inside- cache.Lock() nesting this removes. Workers break rather than return on error so already-buffered keys still flush. The pre-existing error path never rolled back emitted keys, so dropping them would change the set — and since Zero's hasConflict is a pure existential, a subset risks a lost update while a superset only costs a spurious abort. Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn, 64 threads, budget=30, 2 runs per arm: badger_write_bytes_user 11.44 -> 12.45 MB/s (+8.8%) triples/s 113001 -> 123732 (+9.5%) Cumulative over stock across the three commits: 9.63 -> 12.45 MB/s (+29%). The conflict-key path disappears from the goroutine profile entirely: share of pipeline goroutines 68.5% -> 0.0%, blocked on its mutex 23.0% -> 0.0%, and total pipeline goroutine observations fall 518 -> 216. The new top blocker is Deltas.AddToDeltas (57.6% of a much smaller blocked population) — shard contention on the delta map, where NumShards = 30 is thin for 64 vCPU. New TestFillContextKeysGolden pins the exact ctx.Keys sent to Zero against a golden captured at 612ae8c, and asserts they are startTs-independent. The six existing byte-identical tests only compare budget 0 vs N, so they cannot catch a regression that changes the serial and parallel paths the same way; nothing asserted FillContext's output before. Full posting package green under -race. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…nquad worker.MaxLeaseId takes a global RWMutex read lock on the membership state to return a single uint64. ExtractBlankUIDs called verifyUid — and thus MaxLeaseId — for EVERY subject and every object uid, including blank nodes, whose uid is 0 and always passes. A 20k-nquad mutation of uid edges therefore took roughly 40k acquisitions of shared cluster state to compare against a number that barely moves. Take one snapshot per ExtractBlankUIDs call and only fall through to verifyUid for uids above it. The lease is monotonically non-decreasing, so a uid at or below the snapshot was already leased and cannot become unleased; a uid above it still goes through the live re-read and the existing wait-for-lease loop, so a uid racing an in-flight lease extension is accepted exactly as before. A stale (lower) snapshot is conservative — it can only route a uid down the slow path, never wrongly accept one. Measured on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn with EXPLICIT uids, 64 threads, 2 runs per arm: ExtractBlankUIDs 3.67% -> absent from the profile verifyUid 3.52% -> absent MaxLeaseId 3.51% -> absent CPU utilisation 8.85% -> 8.6% of 64 vCPU Throughput is UNCHANGED (12.53 MB/s both arms). That is expected and worth recording: at ~9% CPU on 64 vCPU this Alpha is not CPU-bound, because processApplyCh applies one Raft entry at a time. This removes real work and real contention on shared membership state, which matters for a CPU-bound deployment, for co-located Alphas, and once cross-transaction concurrency raises utilisation — but it does not move throughput today. Only relevant when mutations carry explicit uids, which is the common case for the high-ingest workload this series targets. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Measured results for the four perf commits on an i4i.16xlarge (32 physical cores / 64 vCPU, local NVMe), 25 [uid] @reverse predicates, 20k triples per transaction, 64 writer threads, 300s runs, using badger_write_bytes_user — the same metric as the Preview1 vs Preview1-Hybrid comparison in dgraph-io#9727. Records the throughput ladder (9.63 -> 12.45 MB/s), the goroutine-parking progression that shows the bottleneck migrating from the global cache lock to the conflict-key lock and then away entirely, and the CPU utilisation that stayed under 10% of 64 vCPU throughout. Also documents the runs that were DISCARDED and why, so they are not re-run by accident: a stale Alpha holding :8080 that made both arms profile the same binary, a load generator whose file striding caused 1.47M transaction aborts, and one throughput outlier. Each of these produced plausible-looking numbers that were wrong. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Full write-up for 5b69ac2, kept because the interesting part is the decision NOT to shard. Sharding Txn.conflicts was the obvious fix and was rejected on arithmetic: a line-level profile showed ~80% of addConflictKeyWithUid's cost was lock acquire/release and only ~15% the map insert, so batching to one acquisition per Process* captured the win while sharding would have chased a ~1% residual — at the cost of changing the container type, nine test sites, FillContext, and NumShards, which is shared with the delta and index maps. Records the explicit revisit trigger so the question stays closed unless the numbers change. Also captures the three implementation traps that produce silent corruption rather than a failure: eager key expansion (call sites reuse one scratch key buffer, so a deferred fingerprint hashes the wrong uid), defer ordering in InsertTokenizerIndexes (registering the flush below cache.Lock compiles and passes every test while keeping the nested lock), and break-not-return on worker error (a shrunken conflict set risks a lost update, since Zero's hasConflict is a pure existential). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Forward-looking companion to the validation record. Three sections worth
flagging for review:
Flag simplification. The pipeline now carries four superflag knobs, two of
which are inert unless mutations-pipeline-goroutines=-1 (not the default),
and the shipped default of 30 sits exactly on the allocateWorkers cliff:
with 25 predicates it yields a {1:20, 2:5} grant. Before 612ae8c that
silently disabled the lock-free path for 20 of 25 predicates. Also notes
that WorkerOptions.String() hand-formats the struct and omits every
MutationsPipeline* field, so the startup log cannot confirm the effective
configuration — this cost real time during live debugging.
Where the CPU actually goes now. After the series the profile is no longer
dominated by posting/: RWMutex read traffic (~11%), worker.(*groupi).Tablet
(~7-9%, a read-mostly membership map behind SafeMutex hit once per
proposal), and allocation pressure (~10%). MaxLeaseId/verifyUid was in this
list until 016a03a removed it.
Harness lessons, recorded so they are not re-learned: Dgraph returns HTTP
200 with an errors body on rejection, explicit uids need a zero lease
first, cross-predicate lock contention requires multiple predicates per
TRANSACTION rather than multiple client threads, and sequential A/B on a
thermally-throttling laptop produced a spurious 43% regression on an
unchanged code path.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Deltas.AddToDeltas was the top blocker left inside the mutation pipeline — 57.6% of blocked pipeline goroutines. It is one Set into a LockedShardedMap[string, []byte] that every predicate worker writes to, and the map was fixed at NumShards = 30. That constant is sized for types.ShardedMap, an unrelated single-threaded map used by the query path, and is far too thin for 64 vCPU. Two coupled changes: 1. Shard count is now per-instance, defaulting to 4x GOMAXPROCS clamped to [64, 1024] and rounded to a power of two. The struct already held shards and locks as slices, so only the constructor was fixed at NumShards — but getShardIndex ALSO indexed off the global constant, which would over-run the slice or strand every shard above index 30 once the count varied. It now masks against the map's own size. types.ShardedMap and the shared NumShards constant are deliberately untouched. 2. String keys hashed with farm.Fingerprint64([]byte(k)), whose conversion escapes to the heap on every Get/Set — previously ~13% of all allocations in a 20k-edge batch, since every delta write hashes its key. maphash.String is allocation-free. Nothing persisted depends on shard assignment, so the hash may change freely. uint64 keys are multiplied by a 64-bit constant before masking, because dense monotonic uids would stripe poorly against a power-of-two mask. Merge now falls back to a key-by-key path when two maps have different shard counts; positional merging would silently drop keys. It has no callers today, but per-instance sizing made that a live hazard. Micro-benchmark, interleaved, median of 5 (Set with N goroutines): workers=1 499 -> 434 ns/op 3 -> 2 allocs workers=32 229 -> 171 ns/op 2 -> 1 allocs (1.34x) workers=64 212 -> 165 ns/op 2 -> 1 allocs (1.28x) End to end on i4i.16xlarge (32 physical cores), 25 [uid] @reverse predicates, 20k triples/txn, 64 threads, budget=30, 2 runs per arm: badger_write_bytes_user 12.44 -> 13.17 MB/s (+5.9%) triples/s 123795 -> 131449 (+6.2%) blocked on mutex 25.0% -> 10.5% runnable 48.8% -> 75.6% pipeline goroutines 160 -> 86 Cumulative over stock: 9.63 -> 13.17 MB/s (+36.8%). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…rule CLAUDE.md is gitignored in this repo, so the same summary goes here to survive in git history and reach anyone reading the branch. Records the six-commit ladder (9.63 -> 13.17 MB/s, +36.8%) and the observed bottleneck migration — global cache lock 86% -> conflict lock 71% -> AddToDeltas 58% -> nothing above ~10% — which is why the commits had to land in that order, each fix exposing the next. The load-bearing part is the rule for future work: a single Alpha sits at ~10% of 64 vCPU because processApplyCh applies one Raft entry at a time, so a change that only reduces CPU or allocations will NOT move throughput. 016a03a removed ~3.5% of CPU for exactly zero gain. Candidates must be ranked by share of BLOCKED goroutines, not CPU. Without that rule the remaining backlog items look attractive and are worthless — which is why they were skipped deliberately rather than left undone by accident. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Records the strongest gate run so far: a mixed-workload corpus (~408k triples, 17 predicate types) ingested into four fresh clusters and fingerprinted with 72 DQL queries. stock@30, new@0, new@30 and new@auto all produce IDENTICAL results with zero mutation errors, covering every apply-path branch the pipeline work touched — ProcessSingle/List/Reverse/ Count, ten index tokenizers, @upsert, @noconflict, @lang, lists, DEL and star-delete — rather than only the [uid] @reverse shape used for benchmarking. Also records, with evidence, that the worker package's -race failures are pre-existing and structural rather than caused by this work: worker.Init overwrites a package-global limiter and starts a bleed() goroutine that is never stopped, so every test's Init races the previous test's goroutine and the detector blames whichever test is running. Confirmed by running the same suite on ab0a49e — both trees fail, with fluctuating sets, and TestLimiterDeadlock fails 5/5 on both. posting passes cleanly on both. Plus the harness trap that invalidated the first attempt: /admin export is asynchronous, so sleeping and reading the export dir compares partial flushes, not databases. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Full t/ runner suite against a docker image verified by symbol inspection to contain the changes. 39 packages ok with no test or package failures, including systest/mutations-and-queries, systest/vector (HNSW), and worker — which passes cleanly here despite the pre-existing worker.Init race that makes its -race unit run flaky. Records the setup a clean box needs (Docker, gotestsum, a git root, make local-image; ack is unavailable on RHEL 9 but the runner does not need it), and two things that look like failures and are not: the four panic: entries come from the deliberate panic_catcher test, and TestRebuildTokIndex only appears to stall if badger's INFO output is truncated by grep | head — it passes in 0.06s. One honest gap: the first full-suite invocation exited 1 with no failing test and a log ending mid-posting; rerunning that package alone gave exit 0. Not explained. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Re-ran the A/B against the ext-pointer-tech-collab three-stage generator, which replicates the NiFi/dgraph4j production pipeline: 30 @reverse and 20 @upsert predicates, 1.67M seeded nodes, Zipf hot-node reverse fan-out, @upsert re-asserts and update waves, 532 transaction-sized files. Both arms restore the same seeded snapshot so only the binary differs. stock 1.475 -> shard 1.735 MB/s (+17.6%), 44,699 -> 52,357 nquads/s (+17.1%), 129.4s -> 110.5s wall clock, reproducible to ~1% across 2 runs per arm. Abort counts are identical (~1,860 both arms), which is the strongest correctness signal yet for the conflict-key batching: conflict semantics are unchanged under real @upsert contention. Records two things the synthetic corpus hid: absolute throughput is ~1.5 MB/s rather than 13 because this workload is abort-dominated (only ~283 of 532 files commit; the ceiling is Zero's conflict arbitration, not the apply path), and the relative gain is half the synthetic figure. 17.6% is the number to quote for production, not 36.8%. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Swept client threads {2,4,8,16} on the production-replica corpus. Two
findings.
Throughput more than TRIPLES as threads drop from 16 to 2 (stock 1.48 ->
5.30 MB/s, 44.7k -> 143k nquads/s); aborts fall 1,860 -> 380 and committed
files rise 283 -> 502 of 532. On this workload the Alpha is not the
constraint — Zero's conflict arbitration over @upsert re-asserts is, and
each extra client thread buys more aborted work than committed work. This
quantifies what the production client-side node-UID lock cache is worth,
and suggests an ingest tuned for high thread counts without it is losing
throughput rather than gaining it.
The apply-path gain holds at every thread count (1.12x at 2 threads through
1.18x at 16) and is therefore not an artifact of abort noise: at 2 threads,
where 502/532 files commit and aborts are minimal, it is still +12%.
Abort counts track each other across all four thread counts (380/379,
945/931, 1465/1479, 1855/1857), confirming conflict semantics are unchanged.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…tion-* Four superflag keys become three, renamed to say what they scope. All of this parallelizes a single mutation; processApplyCh is still serial across transactions, so the names now make that visible rather than leaving an operator to infer it. mutations-pipeline-threshold -> intra-mutation-min-edges mutations-pipeline-goroutines -> intra-mutation-parallelism mutations-pipeline-goroutines-fraction ^ (merged: off|auto|N|Fx) mutations-pipeline-min-edges-per-worker -> intra-mutation-edges-per-worker The two merged keys were the same axis in two notations -- a worker count and a multiple of GOMAXPROCS -- with the first doubling as the tag selecting which was read. The -1 AUTO sentinel is gone; auto is 1x. Two of the four keys used to be inert unless goroutines == -1. That is fixed by deleting the conditional, not the flags: edges-per-worker now caps every sizing mode, because "do not spin N workers for a handful of edges" is as true of a fixed count as a derived one. It stays a flag deliberately -- 256 was adopted by analogy to DivideAndRule and never measured, and it is the binding term on a large box, where a 20k-edge mutation caps at 78 workers however many cores exist. Decouple the lock-free store from the worker grant. allocateWorkers now floors every predicate at 1 instead of returning nil, so a grant of 1 means "no fan-out", not "take the legacy locked store". Tying those together made the disabled setting also give up the lock-free store -- about two thirds of this branch's measured gain -- and made the flag non-monotonic, since a one-worker grant dropped index tokenization from 10 goroutines to 1. numGo is now max(10, grant). intra-mutation-min-edges=0 remains the single kill switch. Also adds the observability that was missing: the three fields now appear in WorkerOptions.String() (the startup log could not previously confirm whether the budget was 30 or 0), a V(2) line reports the resolved grant and which term bound it, and intra_mutation_no_fanout_total counts the silent-degradation case. Measured on an i4i.16xlarge against the reverse-heavy corpus this branch was optimized for: 132,730 vs 133,026 triples/s at a fixed budget (-0.22%) and 132,664 vs 132,513 at auto (+0.11%) -- both within run-to-run noise. TestSchemaMatrixByteIdentical now compares against the REAL legacy path (a serial runMutation loop) rather than the pipeline with its budget off, which the store decoupling removed. That is strictly stronger, and it surfaced three pipeline-vs-legacy divergences that predate this work; they are measured on the pre-change tree and recorded in pipeline-todo.md rather than pinned here. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
pipeline-todo.md §1 is closed, with the reasoning that mattered: the fix for the inert flags was removing the conditional rather than the flags, and edges-per-worker stayed tunable because it is unmeasured and is what actually binds on a large box. Two findings the trace produced that were not in the original write-up: a worker count of 1 was silently identical to 0 while 2 gave every predicate a lock-free grant, and the AUTO fraction saturated -- making the help text's "raise it to oversubscribe to 2-3x cores" advice unfollowable at the batch size it was written for. Both are now pinned by tests. New §1c records three pipeline-vs-legacy divergences surfaced by rebasing the byte-identity matrix onto the real legacy path. All three reproduce on the pre-change tree, so they are unrelated to this work. The @lang case deserves its own investigation: the pipeline drops 600 conflict keys the legacy path emits, which is the direction that risks a lost update, and the pipeline has been the production default all along. ec2-validation-results.md keeps its tables verbatim -- they describe runs that actually happened under the old keys -- with a translation table for reproducing them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…le it The client-thread sweep concluded that "Zero's conflict arbitration over @upsert re-asserts" is the constraint. Profiling says otherwise: Zero uses 9% of ONE core on a 64-vCPU box, its goroutine count is flat, and exactly one sample out of 1,695 lands inside Oracle.commit. It decides those ~1,860 aborts essentially for free. The real mechanism is wasted apply work. A doomed transaction still takes a startTs, still gets proposed through Raft, and still runs a full pre-write through the serial apply loop before being discarded — ~2,100 applies to commit ~283 files at 16 threads, so ~85% of apply work is thrown away while the loop sits 68-88% occupied. The distinction is actionable: a faster Zero buys nothing, fewer conflicts buys ~3x. New §5a adds the control the original sweep lacked. Falling throughput with rising threads cannot by itself separate "aborts are expensive" from "concurrency is expensive", so the same sweep was run on a corpus with no upserts and therefore no aborts. It goes the other way — 92,982 -> 134,173 triples/s from 2 to 64 threads, against a 3.2x fall on the abort-heavy corpus. Concurrency is not what hurts. Lining those up gives what a client-side lock is worth: the apply loop caps this box at ~130-143k/s, abort-heavy at 2 threads already reaches it, and abort-heavy at 16 threads sits 3.2x below. A lock should recover roughly 3x at 16 threads — but it reaches that ceiling rather than beating it, since it removes abort waste without lifting the serial apply loop. Its edge over simply lowering thread count is that it serialises only conflicting writes. Also records that Zero publishes no transaction metrics at all, and notes "triples" in the sweep heading is the verb, not RDF triples. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
pipeline-todo.md, ec2-validation-results.md and conflict-key-batching-plan.md are working notes for this series, not upstream documentation. The authoritative copies now live in the dgraph-plans repo (01-intra-txn-apply/mirror/), verified in sync before this removal. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011uGZrf7FExkfwT5cWowrb5
…ne-reverse-parallel
rahst12
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
The complete per-predicate mutation pipeline, now targeting
main: the pipeline implementation revived from #9467, its correctness-fix series, and a measured intra-mutation parallelism series on top (+36.8% sustained ingest on a 32-core EC2 box, +17.6% on a production-replica predicate mix).The legacy apply path processes all edges of a Raft proposal serially. The pipeline groups a mutation's edges by predicate and applies them in parallel — one runner per predicate, with this PR adding a proportional worker budget across predicates, per-CPU (
auto) sizing, and a parallel forward data-write inProcessListso[uid] @reversepredicates parallelize too (their forward write was serial and measured 1.00× before that change).Why the base changed
This PR previously targeted
harshil-goel/mutation-pipeline(#9467). That branch has been inactive since May 2026 and its author is no longer active on the project, so merging into it would strand the work. This branch already contains every commit from #9467 with authorship preserved — 20 commits including the pipeline itself and two rounds of correctness fixes — so retargeting tomainmakes this PR the single, reviewable path for all of it. #9467 can be closed as superseded by this PR.What's included
The pipeline series (from #9467, authorship preserved):
InsertTokenizerIndexesdeadlock on UIDs ≥ 2⁶³;IterateDisk.IsEmptystubbed tofalse(brokehas()after star-deletes); HNSW vector-insert races (now serialized); star-delete skipping index deletion during background indexing; scalar Del-of-old-value wiping a new Set inProcessCount; nil-deref inProcessSingleon multi-Del-per-uid; six further gaps surfaced by alpha integration tests[uid] @reverse @counttests, a 592-genre reverse-count systestThe intra-mutation parallelism series (this branch, each commit benchmarked on i4i.16xlarge):
LockedShardedMapsizing+36.8% cumulative (reverse-heavy corpus: 25
[uid] @reversepredicates, 20k triples/txn, 64 writer threads, 300 s runs). A production-replica predicate mix measured +17.6%.Review-critical facts, stated up front
intra-mutation-min-edgesdefaults to1, so every mutation ≥1 edge routes through the pipeline. The single kill switch isintra-mutation-min-edges=0, which restores the legacy serial path entirely.mutations-pipeline-thresholdtointra-mutation-*. The names say what they scope: this parallelizes a single mutation. Transactions still apply serially inprocessApplyCh, so these flags will not relieve a many-concurrent-writers bottleneck.How to configure
All keys live under the existing
--feature-flagssuperflag. Every key is live at every setting:intra-mutation-min-edges1>0andlen(edges) >= it.0= legacy path entirely, and is the single kill switch.intra-mutation-parallelismautooff= one per predicate, no fan-out.N= exactly N.Fx= F per CPU available to Go (e.g.1.5x).auto=1x.intra-mutation-edges-per-worker256totalEdges / thisso small mutations don't over-spawn. Applies to every sizing mode.Note which term binds: at the default
edges-per-worker=256, a 20k-edge mutation caps at 78 workers however many cores exist — above ~78 CPUs the cap governs, not the multiplier. Loweringintra-mutation-edges-per-workeris what lifts that ceiling. TheV(2)log line reports which term won (boundBy=).Correctness verification
intra-mutation-parallelism=offis byte-identical to the one-goroutine-per-predicate path; byte-identical committed output and identical conflict-key set verified under-racefor scalar,[uid] @reverse, and per-CPU-vs-fixed sizing.ProcessReversestays serial deliberately (many-to-one, hot-target-prone — parallelizing it measured as a net loss); all workers join before the reverse/index/count passes, preserving MVCC order.main(2026-08-08): zero conflicts; full build and theposting/,worker/,x/unit suites pass on the merged tree.Known open item
On an
@lang-heavy workload the pipeline path emits fewer conflict keys than the legacy path (divergence measured at ~600 keys on a synthetic corpus). This is pre-existing pipeline behavior, not introduced by this series, and is documented as the highest-priority follow-up since the risk direction (lost update under SI) matters once the pipeline is the default. Flagging it here rather than leaving it for reviewers to find.Benchmarks (8-core dev box — a floor; the win scales with cores)
~20k-edge batch, 5 hot reverse targets.
edges/s, speedup vsbudget=0:[uid] @reversedominant@indexdominant (control)BenchmarkReverseDominant/BenchmarkReverseFiftyFiftyare included so this can be re-swept on production-class hardware.🤖 Generated with Claude Code
https://claude.ai/code/session_011uGZrf7FExkfwT5cWowrb5