feat(mutations): replace pipeline on/off flag with edge-count threshold

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) <[email protected]>

Author: matthewmcneely

dgraph/cmd/alpha/run.go

@@ -796,7 +796,7 @@ func run() {
 	x.Config.NormalizeCompatibilityMode = featureFlagsConf.GetString("normalize-compatibility-mode")
 	enableDetailedMetrics := featureFlagsConf.GetBool("enable-detailed-metrics")
 	x.WorkerConfig.SlowQueryLogThreshold = featureFlagsConf.GetDuration("log-slow-query-threshold")
-	x.WorkerConfig.MutationsUsePipeline = featureFlagsConf.GetBool("mutations-use-pipeline")
+	x.WorkerConfig.MutationsPipelineThreshold = int(featureFlagsConf.GetInt64("mutations-pipeline-threshold"))
 
 	x.PrintVersion()
 	glog.Infof("x.Config: %+v", x.Config)

worker/draft.go

@@ -526,7 +526,7 @@ func (n *node) applyMutations(ctx context.Context, proposal *pb.Proposal) (rerr
 	// Discard the posting lists from cache to release memory at the end.
 	defer txn.Update()
 
-	if x.WorkerConfig.MutationsUsePipeline {
+	if t := x.WorkerConfig.MutationsPipelineThreshold; t > 0 && len(m.Edges) >= t {
 		mp := posting.NewMutationPipeline(txn)
 		return mp.Process(ctx, m.Edges)
 	}

worker/server_state.go

@@ -43,7 +43,7 @@ const (
 		`lambda-url=;`
 	CacheDefaults        = `size-mb=4096; percentage=40,40,20; remove-on-update=false`
 	FeatureFlagsDefaults = `normalize-compatibility-mode=; enable-detailed-metrics=false; ` +
-		`log-slow-query-threshold=0; mutations-use-pipeline=false`
+		`log-slow-query-threshold=0; mutations-pipeline-threshold=0`
 )
 
 // ServerState holds the state of the Dgraph server.

x/config.go

@@ -138,10 +138,18 @@ type WorkerOptions struct {
 	HardSync bool
 	// Audit contains the audit flags that enables the audit.
 	Audit bool
-	// MutationsUsePipeline enables the per-predicate mutation pipeline in
-	// applyMutations. When false (default), mutations follow the legacy
-	// serial path. The flag is plumbed via the "feature-flags" superflag.
-	MutationsUsePipeline bool
+	// MutationsPipelineThreshold gates the per-predicate mutation pipeline
+	// in applyMutations. A mutation runs through the pipeline only when
+	// MutationsPipelineThreshold > 0 and len(m.Edges) >= the threshold;
+	// otherwise it falls back to the legacy serial path. Set to 0 (default)
+	// to disable the pipeline entirely. Set to 1 to always use the pipeline.
+	// The pipeline pays goroutine spin-up cost per predicate, so small
+	// mutations are slower on it; bulk multi-predicate mutations are
+	// faster — pick a value above the per-mutation edge count where the
+	// crossover happens for your workload (~100 in benchmarks here).
+	// Plumbed via the "feature-flags" superflag as
+	// "mutations-pipeline-threshold".
+	MutationsPipelineThreshold int
 }
 
 // WorkerConfig stores the global instance of the worker package's options.