feat(mutations): gate per-predicate pipeline behind a feature flag

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

Author: matthewmcneely

dgraph/cmd/alpha/run.go

@@ -796,6 +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.PrintVersion()
 	glog.Infof("x.Config: %+v", x.Config)

worker/draft.go

@@ -526,8 +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()
 
-	featureFlag := true
-	if featureFlag {
+	if x.WorkerConfig.MutationsUsePipeline {
 		mp := posting.NewMutationPipeline(txn)
 		return mp.Process(ctx, m.Edges)
 	}

worker/server_state.go

@@ -42,7 +42,8 @@ const (
 	GraphQLDefaults    = `introspection=true; debug=false; extensions=true; poll-interval=1s; ` +
 		`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`
+	FeatureFlagsDefaults = `normalize-compatibility-mode=; enable-detailed-metrics=false; ` +
+		`log-slow-query-threshold=0; mutations-use-pipeline=false`
 )
 
 // ServerState holds the state of the Dgraph server.

x/config.go

@@ -138,6 +138,10 @@ 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
 }
 
 // WorkerConfig stores the global instance of the worker package's options.