conflict: create conflict workload for testing LDR

This creates a conflict workload and roachtest for stress testing LDR
with random schemas. The `conflict` workload accepts connections for two
independent clusters. For each randomly generated row, mutated versions
of the row are inserted into the peer clusters at the same time. The
test validates that the two clusters eventually converge and there are
no entries in the DLQ.

Release note: none
Part of: #148386

Author: jeffswenson

pkg/BUILD.bazel

@@ -177,6 +177,7 @@ ALL_TESTS = [
     "//pkg/config/zonepb:zonepb_test",
     "//pkg/config:config_disallowed_imports_test",
     "//pkg/config:config_test",
+    "//pkg/crosscluster/ldrrandgen:ldrrandgen_test",
     "//pkg/crosscluster/logical:logical_disallowed_imports_test",
     "//pkg/crosscluster/logical:logical_test",
     "//pkg/crosscluster/physical:physical_test",
@@ -848,6 +849,7 @@ ALL_TESTS = [
     "//pkg/util:util_test",
     "//pkg/workload/bank:bank_test",
     "//pkg/workload/cli:cli_test",
+    "//pkg/workload/conflict:conflict_test",
     "//pkg/workload/faker:faker_test",
     "//pkg/workload/histogram/exporter:exporter_test",
     "//pkg/workload/histogram:histogram_test",
@@ -1354,6 +1356,8 @@ GO_TARGETS = [
     "//pkg/config/zonepb:zonepb_test",
     "//pkg/config:config",
     "//pkg/config:config_test",
+    "//pkg/crosscluster/ldrrandgen:ldrrandgen",
+    "//pkg/crosscluster/ldrrandgen:ldrrandgen_test",
     "//pkg/crosscluster/logical:logical",
     "//pkg/crosscluster/logical:logical_test",
     "//pkg/crosscluster/physical:physical",
@@ -2863,6 +2867,8 @@ GO_TARGETS = [
     "//pkg/workload/bulkingest:bulkingest",
     "//pkg/workload/cli:cli",
     "//pkg/workload/cli:cli_test",
+    "//pkg/workload/conflict:conflict",
+    "//pkg/workload/conflict:conflict_test",
     "//pkg/workload/connectionlatency:connectionlatency",
     "//pkg/workload/debug:debug",
     "//pkg/workload/examples:examples",

pkg/ccl/workloadccl/allccl/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//pkg/ccl/workloadccl/roachmartccl",
         "//pkg/workload/bank",
         "//pkg/workload/bulkingest",
+        "//pkg/workload/conflict",
         "//pkg/workload/connectionlatency",
         "//pkg/workload/debug",
         "//pkg/workload/examples",

pkg/ccl/workloadccl/allccl/all.go

@@ -13,6 +13,7 @@ import (
 	_ "github.com/cockroachdb/cockroach/pkg/ccl/workloadccl/roachmartccl"
 	_ "github.com/cockroachdb/cockroach/pkg/workload/bank"
 	_ "github.com/cockroachdb/cockroach/pkg/workload/bulkingest"
+	_ "github.com/cockroachdb/cockroach/pkg/workload/conflict"
 	_ "github.com/cockroachdb/cockroach/pkg/workload/connectionlatency"
 	_ "github.com/cockroachdb/cockroach/pkg/workload/debug"
 	_ "github.com/cockroachdb/cockroach/pkg/workload/examples"

pkg/cmd/roachtest/tests/logical_data_replication.go

@@ -216,18 +216,35 @@ func registerLogicalDataReplicationTests(r registry.Registry) {
 			},
 			run: TestLDRCreateTablesTPCC,
 		},
+		{
+			name: "ldr/conflict",
+			clusterSpec: multiClusterSpec{
+				leftNodes:  3,
+				rightNodes: 3,
+				clusterOpts: []spec.Option{
+					spec.CPU(4),
+					spec.WorkloadNode(),
+					spec.WorkloadNodeCPU(4),
+					spec.VolumeSize(100),
+				},
+			},
+			ldrConfig: ldrConfig{
+				createTables: true,
+			},
+			run: TestLDRConflict,
+		},
 	}
 
 	for _, sp := range specs {
-
 		r.Add(registry.TestSpec{
-			Name:             sp.name,
-			Owner:            registry.OwnerDisasterRecovery,
-			Timeout:          60 * time.Minute,
-			CompatibleClouds: registry.OnlyGCE,
-			Suites:           registry.Suites(registry.Nightly),
-			Cluster:          sp.clusterSpec.ToSpec(r),
-			Leases:           registry.MetamorphicLeases,
+			Name:                       sp.name,
+			Owner:                      registry.OwnerDisasterRecovery,
+			Timeout:                    60 * time.Minute,
+			CompatibleClouds:           registry.OnlyGCE,
+			Suites:                     registry.Suites(registry.Nightly),
+			Cluster:                    sp.clusterSpec.ToSpec(r),
+			Leases:                     registry.MetamorphicLeases,
+			RequiresDeprecatedWorkload: true, // TODO(jeffswenson): require this only for conflict test.
 			Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 				rng, seed := randutil.NewPseudoRand()
 				t.L().Printf("random seed is %d", seed)
@@ -388,6 +405,73 @@ func TestLDRTPCC(
 	VerifyCorrectness(ctx, c, t, setup, leftJobID, rightJobID, 2*time.Minute, workload)
 }
 
+func TestLDRConflict(
+	ctx context.Context, t test.Test, c cluster.Cluster, setup multiClusterSetup, ldrConfig ldrConfig,
+) {
+	setup.left.sysSQL.Exec(t, "CREATE DATABASE conflict")
+	setup.right.sysSQL.Exec(t, "CREATE DATABASE conflict")
+
+	var leftJobID, rightJobID int
+	setup.right.sysSQL.Exec(t, fmt.Sprintf("CREATE EXTERNAL CONNECTION IF NOT EXISTS left AS '%s'", setup.left.PgURLForDatabase("conflict")))
+	setup.left.sysSQL.Exec(t, fmt.Sprintf("CREATE EXTERNAL CONNECTION IF NOT EXISTS right AS '%s'", setup.right.PgURLForDatabase("conflict")))
+
+	leftURLs, err := c.InternalPGUrl(ctx, t.L(), setup.left.gatewayNodes, roachprod.PGURLOptions{
+		Database: "conflict",
+	})
+	require.NoError(t, err)
+	rightURLs, err := c.InternalPGUrl(ctx, t.L(), setup.right.gatewayNodes, roachprod.PGURLOptions{
+		Database: "conflict",
+	})
+	require.NoError(t, err)
+
+	leftURL := fmt.Sprintf("\"%s\"", leftURLs[0])
+	rightURL := fmt.Sprintf("\"%s\"", rightURLs[0])
+
+	c.Run(ctx, option.WithNodes(setup.workloadNode), "./workload", "init", "conflict", leftURL)
+
+	t.Status("creating bidirectional replication job")
+	setup.right.sysSQL.QueryRow(t, `
+	CREATE LOGICALLY REPLICATED TABLE conflict.conflict FROM TABLE conflict.conflict
+		ON 'external://left'
+		WITH BIDIRECTIONAL ON 'external://right'
+	`).Scan(&rightJobID)
+
+	t.Status("waiting for right job to start up")
+	waitForReplicatedTime(t, rightJobID, setup.right.db, getLogicalDataReplicationJobInfo, 2*time.Minute)
+
+	t.Status("waiting for left job to be created")
+	testutils.SucceedsWithin(t, func() error {
+		return setup.left.db.QueryRow("SELECT job_id FROM [SHOW JOBS] WHERE job_type = 'LOGICAL REPLICATION'").Scan(&leftJobID)
+	}, 2*time.Minute)
+
+	t.Status("waiting for left job to start up")
+	waitForReplicatedTime(t, leftJobID, setup.left.db, getLogicalDataReplicationJobInfo, 2*time.Minute)
+
+	// TODO(jeffswenson): mix in random schema changes. The high level plan is:
+	// 1. Pause the workload.
+	// 2. Wait for LDR replication to catch up.
+	// 3. Stop the LDR jobs.
+	// 4. Make random schema changes.
+	// 5. Start the LDR jobs again using now() as the cursor.
+	// 6. Resume the workload.
+	t.Status("running workload")
+	c.Run(ctx, option.WithNodes(setup.workloadNode),
+		"./workload", "run", "conflict", "--duration=15m",
+		// Tolerate errors because there are some we can't easily avoid in random schemas that
+		// contain computed columns. For example, the computed column a+b may cause an insert error
+		// if a+b overflows the type.
+		"--tolerate-errors",
+		"--peer_url", rightURL,
+		leftURL)
+
+	t.Status("verifying results")
+	VerifyCorrectness(ctx, c, t, setup, leftJobID, rightJobID, 2*time.Minute, LDRWorkload{
+		dbName:            "conflict",
+		tableNames:        []string{"conflict"},
+		manualSchemaSetup: true,
+	})
+}
+
 // TestLDRCreateTablesTPCC inits the left cluster with 1000 warehouse tpcc,
 // begins unidirectional fast initial scan LDR, starts a tpcc 1000 wh workload
 // on the left, and observes initial scan, catchup scan, and steady state
@@ -990,15 +1074,15 @@ func VerifyCorrectness(
 	ldrWorkload LDRWorkload,
 ) {
 	now := timeutil.Now()
-	t.L().Printf("Waiting for replicated times to catchup before verifying left and right clusters")
+	t.Status("waiting for replicated times to catchup before verifying left and right clusters")
 	if leftJobID != 0 {
 		waitForReplicatedTimeToReachTimestamp(t, leftJobID, setup.left.db, getLogicalDataReplicationJobInfo, waitTime, now)
 		require.NoError(t, replicationtestutils.CheckEmptyDLQs(ctx, setup.left.db, ldrWorkload.dbName))
 	}
 	waitForReplicatedTimeToReachTimestamp(t, rightJobID, setup.right.db, getLogicalDataReplicationJobInfo, waitTime, now)
 	require.NoError(t, replicationtestutils.CheckEmptyDLQs(ctx, setup.right.db, ldrWorkload.dbName))
 
-	t.L().Printf("Verifying equality of left and right clusters")
+	t.Status("verifying equality of left and right clusters")
 
 	type fingerprint struct {
 		table string

pkg/crosscluster/ldrrandgen/BUILD.bazel

@@ -0,0 +1,39 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "ldrrandgen",
+    srcs = ["logical.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/crosscluster/ldrrandgen",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/sql/catalog/colinfo",
+        "//pkg/sql/randgen",
+        "//pkg/sql/sem/tree",
+        "//pkg/sql/types",
+    ],
+)
+
+go_test(
+    name = "ldrrandgen_test",
+    srcs = [
+        "logical_test.go",
+        "main_test.go",
+    ],
+    embed = [":ldrrandgen"],
+    deps = [
+        "//pkg/base",
+        "//pkg/ccl",
+        "//pkg/ccl/storageccl",
+        "//pkg/security/securityassets",
+        "//pkg/security/securitytest",
+        "//pkg/server",
+        "//pkg/sql/sem/tree",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/randutil",
+        "@com_github_stretchr_testify//require",
+    ],
+)

pkg/crosscluster/ldrrandgen/logical.go

@@ -0,0 +1,100 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package ldrrandgen
+
+import (
+	"context"
+	"math/rand"
+
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
+	"github.com/cockroachdb/cockroach/pkg/sql/randgen"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+)
+
+func GenerateLDRTable(
+	ctx context.Context, rng *rand.Rand, tableName string, supportKVWriter bool,
+) *tree.CreateTable {
+	columnByName := func(name tree.Name, columnDefs []*tree.ColumnTableDef) *tree.ColumnTableDef {
+		for _, col := range columnDefs {
+			if col.Name == name {
+				return col
+			}
+		}
+		return nil
+	}
+
+	tableDef := randgen.RandCreateTableWithName(ctx, rng, tableName, 0, []randgen.TableOption{
+		randgen.WithPrimaryIndexRequired(),
+		randgen.WithSkipColumnFamilyMutations(),
+		randgen.WithColumnFilter(func(columnDef *tree.ColumnTableDef) bool {
+			// We don't allow Arrays of bits because rand.LoadTable doesn't correctly identify their bit width.
+			if columnDef.Type.(*types.T).Family() == types.ArrayFamily && columnDef.Type.(*types.T).ArrayContents().Family() == types.BitFamily {
+				return false
+			}
+			// We don't allow the special '"char"' column because pgwire truncates the value to 1 byte.
+			// TODO(jeffswenson): remove this once #149427 is fixed.
+			if columnDef.Type.(*types.T).Family() == types.StringFamily && columnDef.Type.(*types.T).Width() == 1 {
+				return false
+			}
+			return true
+		}),
+		randgen.WithPrimaryIndexFilter(func(indexDef *tree.IndexTableDef, columnDefs []*tree.ColumnTableDef) bool {
+			for _, col := range indexDef.Columns {
+				columnDef := columnByName(col.Column, columnDefs)
+				// TODO(127315): types with composite encoding are not supported in the
+				// primary key by LDR.
+				if colinfo.CanHaveCompositeKeyEncoding(columnDef.Type.(*types.T)) {
+					return false
+				}
+				// Do not allow computed columns in the primary key. Non-virtual computed columns in the primary key is
+				// allowed by LDR in general, but its not compatible with the conflict workload.
+				// TODO(jeffswenson): support computed columns in the primary key in the conflict workload.
+				if columnDef.IsComputed() {
+					return false
+				}
+			}
+			if supportKVWriter && indexDef.Sharded != nil {
+				// The KV writer does not support hash sharded indexes.
+				return false
+			}
+			return true
+		}),
+		randgen.WithIndexFilter(func(indexDef tree.TableDef, columnDefs []*tree.ColumnTableDef) bool {
+			switch indexDef := indexDef.(type) {
+			case *tree.UniqueConstraintTableDef:
+				// Do not allow unique indexes. The random data may cause
+				// spurious unique constraint violations.
+				// TODO(jeffswenson): extend the conflict workload to support unique indexes on fields
+				// that can randomly generate exclusively unique values for each row. E.g. UUIDs could be unique, but
+				// BOOLs are too limiting.
+				return false
+			case *tree.IndexTableDef:
+				for _, col := range indexDef.Columns {
+					if supportKVWriter && col.Expr != nil {
+						// Do not allow expression indexes. These cause SQL to generate a hidden computed column, which is not
+						// supported by the kv writer.
+						if col.Expr != nil {
+							return false
+						}
+					}
+					columnDef := columnByName(col.Column, columnDefs)
+					if columnDef.IsVirtual() {
+						// Virtual computed columns are not supported in indexes by the classic sql writer or the kv writer.
+						// TODO(jeffswenson): remove this restriction once the crud writer is the only writer.
+						return false
+					}
+				}
+				if supportKVWriter && indexDef.Sharded != nil {
+					// The KV writer does not support hash sharded indexes.
+					return false
+				}
+			}
+			return true
+		}),
+	})
+	return tableDef
+}

pkg/crosscluster/replicationtestutils/logical_test.go renamed to pkg/crosscluster/ldrrandgen/logical_test.go

@@ -3,14 +3,16 @@
 // Use of this software is governed by the CockroachDB Software License
 // included in the /LICENSE file.
 
-package replicationtestutils
+package ldrrandgen
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -43,13 +45,14 @@ func TestGenerateLDRTable(t *testing.T) {
 
 	stmt := GenerateLDRTable(ctx, rndSrc, "test_writer", true)
 	t.Logf("creating table: %s", stmt)
-	dbA.Exec(t, stmt)
-
-	dbAURL := GetExternalConnectionURI(t, server, server, serverutils.DBName("a"))
-	dbBURL := GetExternalConnectionURI(t, server, server, serverutils.DBName("b"))
-	dbB.Exec(t,
-		"CREATE LOGICALLY REPLICATED TABLE b.test_writer FROM TABLE a.test_writer ON $1 WITH BIDIRECTIONAL ON $2",
-		dbAURL.String(),
-		dbBURL.String(),
-	)
+	dbA.Exec(t, tree.AsStringWithFlags(stmt, tree.FmtParsable))
+
+	urlA, cleanupA := server.PGUrl(t, serverutils.DBName("a"))
+	defer cleanupA()
+	urlB, cleanupB := server.PGUrl(t, serverutils.DBName("b"))
+	defer cleanupB()
+	sqlDB.Exec(t, fmt.Sprintf("CREATE EXTERNAL CONNECTION a AS '%s'", urlA.String()))
+	sqlDB.Exec(t, fmt.Sprintf("CREATE EXTERNAL CONNECTION b AS '%s'", urlB.String()))
+
+	dbB.Exec(t, "CREATE LOGICALLY REPLICATED TABLE b.test_writer FROM TABLE a.test_writer ON 'external://a' WITH BIDIRECTIONAL ON 'external://b'")
 }