VReplication: Implement Experimental Parallel Applier

.golangci.yml

@@ -236,3 +236,4 @@ formatters:
     paths:
       - examples$
       - ^go/vt/proto/
+      - ^test/antithesis/

CLAUDE.md

@@ -197,6 +197,11 @@ return user.NeedsMigration() && migrate(user) || user
 - **Copyright header** - New Go files must include the project copyright header with the current year
 - **Always run `gofumpt -w`** on changed Go files before committing - this is mandatory
 - **Always run `goimports -local "vitess.io/vitess" -w`** on changed Go files before committing
+- **Always run `golangci-lint run --path-mode=abs --timeout 10m`** (from the `go/` directory, scoped to the changed package(s)) before reporting work complete. CI runs it and will surface modernize/style issues that `go vet`, `gofumpt`, and `goimports` do not — for example:
+  - `waitgroup`: prefer `WaitGroup.Go(func() { ... })` over `wg.Add(1); go func() { defer wg.Done(); ... }()`
+  - `rangeint`: prefer `for range N` over `for i := 0; i < N; i++` when the index is unused
+  - `bloop`: prefer `b.Loop()` over `for i := 0; i < b.N; i++` in benchmarks
+  - `unusedparams`, `unusedwrite`, `unusedfunc`: clean these in code you touch
 - **Use format verbs precisely** - Use `%s` for strings and `%d` for integers, not `%v` for everything
 - **Structured logging** - New log messages should use structured logging with `slog`-style fields (e.g., `log.Warn("message", slog.Any("error", err))`) rather than printf-style logging with format strings
 - **Reuse existing helpers** - Before writing new parsing/validation code, check for existing utilities (e.g., `sqlerror` package for MySQL error codes, `mysqlctl.ParseVersionString()`, `strings.Split()`, `topoproto.TabletAliasString()` for formatting tablet aliases)

changelog/25.0/25.0.0/summary.md

@@ -6,6 +6,7 @@
 
 - **[Major Changes](#major-changes)**
     - **[New Support](#new-support)**
+        - [Experimental parallel VReplication applier](#vreplication-parallel-applier)
     - **[Breaking Changes](#breaking-changes)**
         - [`--watch-replication-stream` flag removed](#vttablet-watch-replication-stream-removed)
         - [Snapshot Topology feature removed](#vtorc-snapshot-topology-removed)
@@ -15,6 +16,8 @@
 - **[Minor Changes](#minor-changes)**
     - **[VReplication](#minor-changes-vreplication)**
         - [Default data protection for `_reverse` workflow cancel/complete](#vreplication-reverse-workflow-data-protection)
+        - [Unknown VStream event types are now hard errors in the applier](#vreplication-unknown-event-error)
+        - [Workflow config overrides sent to source tablets are now allowlisted](#vreplication-source-overrides-allowlist)
     - **[VTGate](#minor-changes-vtgate)**
         - [New controls for cross-keyspace reads](#vtgate-cross-keyspace-reads)
     - **[VTTablet](#minor-changes-vttablet)**
@@ -26,6 +29,15 @@
 
 ### <a id="new-support"/>New Support</a>
 
+#### <a id="vreplication-parallel-applier"/>Experimental parallel VReplication applier</a>
+
+> [!WARNING]
+> This feature is experimental.
+
+VReplication can now apply binlog events using multiple concurrent MySQL connections instead of a single serial connection. Set `--vreplication-parallel-replication-workers=N` (default `1` = serial, maximum `64`) on `vttablet`, or the `vreplication-parallel-replication-workers` per-workflow config override, to dispatch non-conflicting transactions to `N` worker goroutines during the replication (running) phase. Conflicts are detected with target-side writeset hashing (primary key, unique key, and foreign key values — similar to MySQL's own `WRITESET` dependency tracking), so it works regardless of the source's `binlog_transaction_dependency_tracking` setting. Commits remain strictly ordered, so the workflow position, lag metrics, and `WaitForPos` semantics are unchanged. Transactions the conflict detector cannot reason about (DDL, statement-based events, partial row images, prefix/expression unique indexes, and similar) fall back to serial application.
+
+Note that each worker holds two MySQL connections, so a workflow with `N` workers uses `2N+2` target-side connections.
+
 ### <a id="breaking-changes"/>Breaking Changes</a>
 
 #### <a id="vttablet-watch-replication-stream-removed"/>`--watch-replication-stream` flag removed</a>
@@ -84,6 +96,14 @@ When calling `cancel` or `complete` on an auto-generated `_reverse` workflow wit
 
 The `--keep-data` flag help text has been updated to note this default explicitly. This change applies to MoveTables, Reshard, and other VReplication workflow types that use the shared cancel/complete paths.
 
+#### <a id="vreplication-unknown-event-error"/>Unknown VStream event types are now hard errors in the applier</a>
+
+The VReplication applier previously ignored VStream event types it did not recognize. It now fails the workflow with an error for unknown event types (and unknown `on-ddl` actions), failing closed instead of silently skipping events. All event types produced by supported Vitess versions are handled; this only affects streams from sources emitting event types unknown to the target's version.
+
+#### <a id="vreplication-source-overrides-allowlist"/>Workflow config overrides sent to source tablets are now allowlisted</a>
+
+When a workflow has per-workflow config overrides, the target now sends only the source-relevant subset (packet size, timeouts, experimental flags, and similar) to the source tablet's VStreamer instead of the full override map. This keeps newer target-only override keys from failing workflows whose source tablets run an older version that rejects unknown keys.
+
 See [#19906](https://github.com/vitessio/vitess/pull/19906) for details.
 
 ### <a id="minor-changes-vtgate"/>VTGate</a>

examples/benchmark/bench_compare.sh

@@ -0,0 +1,134 @@
+#!/bin/bash
+
+# A/B comparison: serial (workers=1) vs parallel (workers=4) VReplication applier
+# with mixed write workload (INSERT/UPDATE/DELETE/bulk operations).
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+cd "$SCRIPT_DIR" || exit 1
+
+ROW_COUNT=${ROW_COUNT:-200000}
+SEED_ROWS=${SEED_ROWS:-10000}
+RUN_ORDER=${RUN_ORDER:-random}
+export ROW_COUNT SEED_ROWS
+
+echo "============================================"
+echo "  VReplication Parallel Applier Benchmark"
+echo "  ROW_COUNT=$ROW_COUNT  SEED_ROWS=$SEED_ROWS"
+echo "============================================"
+echo ""
+
+run_bench() {
+	local workers=$1
+	local label=$2
+
+	echo ">>> Run: $label (PARALLEL_WORKERS=$workers) <<<"
+	echo ""
+
+	# Teardown any previous state
+	(cd "$SCRIPT_DIR/../local" && ./501_teardown.sh) 2>/dev/null
+
+	# Setup cluster with specified worker count
+	PARALLEL_WORKERS=$workers ./bench_setup.sh || { echo "FAILED: setup for $label"; return 1; }
+
+	# Run benchmark. Use pipefail so a bench_run.sh validation failure is not
+	# masked by tee's zero exit status.
+	(
+		set -o pipefail
+		./bench_run.sh 2>&1 | tee "/tmp/bench_${workers}_workers.log"
+	) || { echo "FAILED: bench_run for $label (validation or drain failure)"; return 1; }
+
+	echo ""
+	echo ">>> $label complete <<<"
+	echo ""
+}
+
+case "$RUN_ORDER" in
+	serial-first)
+		first_workers=1
+		first_label="Serial (1 worker)"
+		second_workers=4
+		second_label="Parallel (4 workers)"
+		;;
+	parallel-first)
+		first_workers=4
+		first_label="Parallel (4 workers)"
+		second_workers=1
+		second_label="Serial (1 worker)"
+		;;
+	random)
+		if (( RANDOM % 2 == 0 )); then
+			first_workers=1
+			first_label="Serial (1 worker)"
+			second_workers=4
+			second_label="Parallel (4 workers)"
+			RUN_ORDER=serial-first
+		else
+			first_workers=4
+			first_label="Parallel (4 workers)"
+			second_workers=1
+			second_label="Serial (1 worker)"
+			RUN_ORDER=parallel-first
+		fi
+		;;
+	*)
+		echo "Invalid RUN_ORDER: $RUN_ORDER"
+		exit 1
+		;;
+esac
+
+echo "Run order: $RUN_ORDER"
+
+# Run 1
+run_bench "$first_workers" "$first_label" || exit 1
+
+# Teardown between runs
+echo "Tearing down between runs..."
+(cd "$SCRIPT_DIR/../local" && ./501_teardown.sh) 2>/dev/null
+sleep 3
+
+# Run 2
+run_bench "$second_workers" "$second_label" || exit 1
+
+# Teardown after
+echo "Tearing down after benchmark..."
+(cd "$SCRIPT_DIR/../local" && ./501_teardown.sh) 2>/dev/null
+
+# Compare results
+echo ""
+echo "============================================"
+echo "  COMPARISON"
+echo "============================================"
+
+for workers in 1 4; do
+	logfile="/tmp/bench_${workers}_workers.log"
+	if [[ -f "$logfile" ]]; then
+		echo ""
+		echo "--- Workers=$workers ---"
+		grep -E "(Drain time|Throughput|Backlog ops|Seed rows)" "$logfile"
+	fi
+done
+
+# Calculate speedup if both logs exist
+serial_log="/tmp/bench_1_workers.log"
+parallel_log="/tmp/bench_4_workers.log"
+if [[ -f "$serial_log" ]] && [[ -f "$parallel_log" ]]; then
+	serial_time=$(grep "Drain time" "$serial_log" | grep -o '[0-9]*')
+	parallel_time=$(grep "Drain time" "$parallel_log" | grep -o '[0-9]*')
+	if [[ -n "$serial_time" ]] && [[ -n "$parallel_time" ]] && [[ "$parallel_time" -gt 0 ]]; then
+		# Integer math: multiply by 100 for 2 decimal places
+		speedup_x100=$((serial_time * 100 / parallel_time))
+		speedup_whole=$((speedup_x100 / 100))
+		speedup_frac=$((speedup_x100 % 100))
+		printf -v speedup_str '%d.%02d' "$speedup_whole" "$speedup_frac"
+		echo ""
+		echo "--- Speedup ---"
+		echo "  Serial:   ${serial_time}s"
+		echo "  Parallel: ${parallel_time}s"
+		echo "  Speedup:  ${speedup_str}x"
+	fi
+fi
+
+echo ""
+echo "============================================"
+echo "Full logs: /tmp/bench_1_workers.log and /tmp/bench_4_workers.log"
+echo "============================================"