Merge #150859 #153721 #153750 #153776 #153783 #153785

150859: cmd/roachtest: add changefeed roachtests exercising execution_locality r=aerfrei a=log-head

in multi-region clusters

Previously, we were not testing to verify that a changefeed with execution_locality enabled creates an acceptable plan and is able to progress. Prior to 24.1, the changefeed would only put aggregators on nodes that were leaseholders for the data. If execution locality is enabled to restrict the changefeed to nodes without leaseholders for that data, then all the ranges would be planned on the gateway node. This roachtest verifies that aggregators are spread across nodes within a region (no one node has a majority of changefeed aggregators).

Epic: CRDB-38755
Fixes: #153736
Release note: None

153721: bench/rttanalysis: shard TestBenchmarkExpectation to avoid timeouts r=spilchen a=spilchen

The TestBenchmarkExpectation benchmark has been frequently timing out after 15 minutes.  This appears to be caused by slow CI machines rather than issues with the test logic itself.

To address this, the test is now split into four shards. Each shard is executed separately and receives the full 15-minute timeout budget. This should reduce the likelihood of timeout test failures.

Fixes #148384

Release note: none
Epic: none

153750: changefeedccl: explicitly set per-table PTS cluster setting in tests r=andyyang890 a=aerfrei

Previously, the per-table protected timestamp cluster setting was only explicitly set to true in tests, which matched the default value. As a result, when tests attempted to disable the setting, it remained enabled, leading to unexpected behavior and assertion failures.

This change ensures the cluster setting is always explicitly set to the value provided by the test, avoiding reliance on defaults and making test behavior consistent and predictable.

Fixes: #153088
Epic: CRDB-1421
Release note: None

153776: asim: fix lower_bound r=wenyihu6 a=tbg

It wasn't being removed from the args, so using it necessarily tripped the test.
Epic: CRDB-49117

153783: sql: fix a couple of edge case bugs with CHECK EXTERNAL CONNECTION r=yuzefovich a=yuzefovich

This commit fixes a couple of edge case bugs with CHECK EXTERNAL CONNECTION implementation. Namely:
- previously, the execution goroutine (that is created in `startExec`) would use the tracing span that is finished, which is not allowed. This is now fixed by deriving a new tracing span.
- also previously we would call `MoveToDraining` from an auxiliary goroutine of the cloud check processor when the context cancellation is observed. This would race with `MoveToDraining` call from the main goroutine in `Next`. The implicit contract of this helper method is that it's only called from the main goroutine, and otherwise it can lead to "MoveToDraining called in state × with err" errors. This is now fixed by removing the call from the auxiliary goroutine since it's actually not needed.

Additionally, this commit simplifies dealing with the `rows` channel a bit.

I decided to omit the release note given we recently merged a fix that contained one.

Fixes: #153378.
Release note: None

153785: remove stray cpu.out file r=rickystewart a=stevendanna

Epic: none
Release note: None

Co-authored-by: Matthew Lougheed <[email protected]>
Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: Aerin Freilich <[email protected]>
Co-authored-by: Tobias Grieger <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: Steven Danna <[email protected]>

Author: 7 people

cpu.out

@@ -13,6 +13,8 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/base",
+        "//pkg/jobs",
+        "//pkg/jobs/jobspb",
         "//pkg/kv/kvclient/kvcoord",
         "//pkg/sql",
         "//pkg/sql/parser",
@@ -56,9 +58,9 @@ go_test(
     data = glob(["testdata/**"]),
     embed = [":rttanalysis"],
     exec_properties = {"test.Pool": "large"},
+    shard_count = 4,
     deps = [
         "//pkg/base",
-        "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
@@ -69,6 +71,7 @@ go_test(
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",
         "//pkg/testutils/testcluster",
+        "//pkg/util/envutil",
         "//pkg/util/protoutil",
         "//pkg/util/randutil",
     ],

pkg/bench/rttanalysis/BUILD.bazel

@@ -9,6 +9,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
@@ -51,15 +53,66 @@ func (r *Registry) Run(b *testing.B) {
 // benchmarks can be filtered by passing the usual test filters underneath
 // this test's name.
 //
-// It takes a long time and thus is skipped under stress, race
-// and short.
+// It takes a long time and thus is skipped under duress and short.
 func (r *Registry) RunExpectations(t *testing.T) {
-	skip.UnderStress(t)
-	skip.UnderRace(t)
+	r.RunExpectationsSharded(t, 1, 1)
+}
+
+// RunExpectationsSharded runs all the benchmarks for one iteration
+// and validates that the number of RPCs meets the expectation. If run
+// with the --rewrite flag, it will rewrite the run benchmarks. The
+// benchmarks can be filtered by passing the usual test filters underneath
+// this test's name.
+//
+// It takes a long time and thus is skipped under duress and short.
+//
+// When shard and totalShards are provided (> 1), only a subset of benchmarks
+// assigned to the specific shard will be run, enabling parallel execution.
+// Test groups are distributed across shards using round-robin assignment.
+func (r *Registry) RunExpectationsSharded(t *testing.T, shard, totalShards int) {
+	defer jobs.TestingSetIDsToIgnore(map[jobspb.JobID]struct{}{3001: {}, 3002: {}})()
+	skip.UnderDuress(t)
 	skip.UnderShort(t)
-	skip.UnderDeadlock(t)
 
-	runBenchmarkExpectationTests(t, r)
+	// If totalShards is 1, run all tests; otherwise shard them
+	var registryToUse *Registry
+	if totalShards <= 1 {
+		// Run all test groups
+		registryToUse = r
+	} else {
+		// Create a registry with only the test groups assigned to this shard
+		shardRegistry := &Registry{
+			numNodes: r.numNodes,
+			cc:       r.cc,
+			r:        make(map[string][]RoundTripBenchTestCase),
+		}
+
+		// Distribute test groups across shards using round-robin assignment
+		// First, get all group names and sort them for consistent ordering
+		groupNames := make([]string, 0, len(r.r))
+		for groupName := range r.r {
+			groupNames = append(groupNames, groupName)
+		}
+		// Sort for deterministic assignment across runs
+		for i := 0; i < len(groupNames); i++ {
+			for j := i + 1; j < len(groupNames); j++ {
+				if groupNames[i] > groupNames[j] {
+					groupNames[i], groupNames[j] = groupNames[j], groupNames[i]
+				}
+			}
+		}
+
+		// Assign groups to shards using round-robin
+		for i, groupName := range groupNames {
+			assignedShard := (i % totalShards) + 1
+			if assignedShard == shard {
+				shardRegistry.r[groupName] = r.r[groupName]
+			}
+		}
+		registryToUse = shardRegistry
+	}
+
+	runBenchmarkExpectationTests(t, registryToUse)
 }
 
 // Register registers a set of test cases to a given benchmark name. It is

pkg/bench/rttanalysis/registry.go

@@ -6,13 +6,44 @@
 package rttanalysis
 
 import (
+	"strconv"
 	"testing"
 
-	"github.com/cockroachdb/cockroach/pkg/jobs"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 )
 
-func TestBenchmarkExpectation(t *testing.T) {
-	defer jobs.TestingSetIDsToIgnore(map[jobspb.JobID]struct{}{3001: {}, 3002: {}})()
-	reg.RunExpectations(t)
+// NOTE: If you change the number of shards, you must also update the
+// shard_count in BUILD.bazel to match.
+const shardCount = 4
+
+// Validate that shardCount matches TEST_TOTAL_SHARDS environment variable at init time
+var _ = func() int {
+	totalShardsStr, found := envutil.ExternalEnvString("TEST_TOTAL_SHARDS", 1)
+	if totalShardsStr == "" || !found {
+		return 0
+	}
+	totalShards, err := strconv.Atoi(totalShardsStr)
+	if err != nil {
+		return 0
+	}
+	if totalShards != shardCount {
+		panic("shardCount mismatch: update shard_count in pkg/bench/rttanalysis/BUILD.bazel to match shardCount constant")
+	}
+	return 0
+}()
+
+func TestBenchmarkExpectationShard1(t *testing.T) {
+	reg.RunExpectationsSharded(t, 1, shardCount)
+}
+
+func TestBenchmarkExpectationShard2(t *testing.T) {
+	reg.RunExpectationsSharded(t, 2, shardCount)
+}
+
+func TestBenchmarkExpectationShard3(t *testing.T) {
+	reg.RunExpectationsSharded(t, 3, shardCount)
+}
+
+func TestBenchmarkExpectationShard4(t *testing.T) {
+	reg.RunExpectationsSharded(t, 4, shardCount)
 }

pkg/bench/rttanalysis/validate_benchmark_data_test.go

@@ -276,6 +276,7 @@ go_library(
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog/catpb",
+        "//pkg/sql/execinfrapb",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/sem/tree",

pkg/cmd/roachtest/tests/BUILD.bazel

@@ -59,6 +59,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachprod/prometheus"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/vm"
 	roachprodaws "github.com/cockroachdb/cockroach/pkg/roachprod/vm/aws"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
@@ -1819,10 +1820,8 @@ func runCDCMultiTablePTSBenchmark(
 		numRanges = params.numRanges
 	}
 
-	if params.perTablePTS {
-		if _, err := db.Exec("SET CLUSTER SETTING changefeed.protected_timestamp.per_table.enabled = true"); err != nil {
-			t.Fatalf("failed to set per-table protected timestamps: %v", err)
-		}
+	if _, err := db.Exec("SET CLUSTER SETTING changefeed.protected_timestamp.per_table.enabled = $1", params.perTablePTS); err != nil {
+		t.Fatalf("failed to set per-table protected timestamps: %v", err)
 	}
 
 	initCmd := fmt.Sprintf("./cockroach workload init bank --rows=%d --ranges=%d --tables=%d {pgurl%s}",
@@ -1912,7 +1911,182 @@ func configureDBForMultiTablePTSBenchmark(db *gosql.DB) error {
 	return nil
 }
 
+func getDiagramProcessors(ctx context.Context, db *gosql.DB) ([]any, error) {
+	var diagramURL string
+	diagramQuery := `SELECT value 
+	FROM system.job_info ji
+	INNER JOIN system.jobs j ON ji.job_id = j.id
+	WHERE j.job_type = 'CHANGEFEED' AND ji.info_key LIKE '~dsp-diag-url-%'`
+	if err := db.QueryRowContext(ctx, diagramQuery).Scan(&diagramURL); err != nil {
+		return nil, err
+	}
+	diagram, err := execinfrapb.FromURL(diagramURL)
+	if err != nil {
+		return nil, err
+	}
+	diagramJSON, err := json.Marshal(diagram)
+	if err != nil {
+		return nil, err
+	}
+	var flow map[string]any
+	if err := json.Unmarshal(diagramJSON, &flow); err != nil {
+		return nil, err
+	}
+	processors, ok := flow["processors"].([]any)
+	if !ok {
+		return nil, fmt.Errorf("processors not found in flow")
+	}
+	return processors, nil
+}
+
+type ChangefeedDistribution struct {
+	NodeToSpansWatched map[int]int
+	ZoneToSpansWatched map[string]int
+	TotalSpansWatched  int
+	TotalAggregators   int
+	TotalLeaseHolders  int
+	TotalRanges        int
+	NodeToZone         map[int]string
+}
+
+func getChangefeedDistribution(
+	processors []any, nodeToZone map[int]string, t test.Test,
+) ChangefeedDistribution {
+	changefeedDistribution := ChangefeedDistribution{
+		NodeToSpansWatched: make(map[int]int),
+		ZoneToSpansWatched: make(map[string]int),
+		TotalSpansWatched:  0,
+		TotalAggregators:   0,
+		TotalLeaseHolders:  0,
+		TotalRanges:        0,
+		NodeToZone:         nodeToZone,
+	}
+	for _, p := range processors {
+		procMap, ok := p.(map[string]any)
+		if !ok {
+			t.Fatalf("processor not a map")
+		}
+		nodeIdx, ok := procMap["nodeIdx"].(float64)
+		require.True(t, ok, "node idx not found in processor")
+		core, ok := procMap["core"].(map[string]any)
+		require.True(t, ok, "core not found in processor")
+		title, ok := core["title"].(string)
+		require.True(t, ok, "title not found in core")
+		if strings.HasPrefix(title, "ChangeAggregator") {
+			changefeedDistribution.TotalAggregators++
+			details := core["details"].([]any)
+			for _, detail := range details {
+				if strings.HasPrefix(detail.(string), "Watches") {
+					re := regexp.MustCompile(`Watches \[(\d+)\]:`)
+					matches := re.FindStringSubmatch(detail.(string))
+					if len(matches) > 1 {
+						numWatches, err := strconv.Atoi(matches[1])
+						require.NoError(t, err)
+						changefeedDistribution.NodeToSpansWatched[int(nodeIdx)] += numWatches
+						changefeedDistribution.TotalSpansWatched += numWatches
+						changefeedDistribution.ZoneToSpansWatched[changefeedDistribution.NodeToZone[int(nodeIdx)]] += numWatches
+
+					}
+				}
+			}
+		}
+	}
+	return changefeedDistribution
+}
+
+func veryifyLeaseHolderDistribution(
+	db *gosql.DB, t test.Test, nodeToZone map[int]string,
+) map[string]int {
+	var rows *gosql.Rows
+	// Get lease holders for all ranges in tpcc database.
+	leaseHolderQuery := `SELECT r.start_pretty, r.replicas, r.replica_localities, r.lease_holder 
+	FROM crdb_internal.ranges r 
+	JOIN crdb_internal.tables t ON r.start_pretty like concat('/Table/', t.table_id::STRING,'%')
+	WHERE t.database_name = 'tpcc'`
+	rows, err := db.Query(leaseHolderQuery)
+	zoneToLeaseHolderCount := make(map[string]int)
+	require.NoError(t, err)
+	defer rows.Close()
+	for rows.Next() {
+		var startKeyPretty string
+		var replicas []uint8
+		var replicaLocalities []uint8
+		var leaseHolder int
+		require.NoError(t, rows.Scan(&startKeyPretty, &replicas, &replicaLocalities, &leaseHolder))
+		for indx := range replicas {
+			require.NotEqual(t, replicas[indx], 0)
+			replicas[indx]--
+		}
+		leaseHolder--
+		zoneToLeaseHolderCount[nodeToZone[leaseHolder]]++
+	}
+	return zoneToLeaseHolderCount
+}
+
 func registerCDC(r registry.Registry) {
+	r.Add(registry.TestSpec{
+		// This test
+		// 1. Creates a cluster with 3 nodes each in us-east and us-west
+		// 2. Runs a tpcc workload, then sets tpcc database to primary region us-west
+		// 3. Creates a changefeed with execution locality set to us-east
+		// 4. Gets the changefeed diagram and creates mappings
+
+		// This test is used to verify that ranges are evenly distributed across
+		// change aggregators in the execution_locality region while targeting tables
+		// whose primary region is different. In issue #2955, in that scenario,
+		// a single change aggregator (on the gateway node) would watch all the ranges.
+		// The above scenario occured with the older bin-packing oracle rather than
+		// the bulk oracle.
+		Name:             "cdc/multi-region-execution-locality-tpcc",
+		Owner:            registry.OwnerCDC,
+		Cluster:          r.MakeClusterSpec(7, spec.Geo(), spec.GatherCores(), spec.GCEZones("us-east1-b,us-west1-b")),
+		CompatibleClouds: registry.OnlyGCE,
+		Suites:           registry.Suites(),
+		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+			nodeToZone := map[int]string{
+				0: "us-east1-b",
+				1: "us-east1-b",
+				2: "us-east1-b",
+				3: "us-west1-b",
+				4: "us-west1-b",
+				5: "us-west1-b",
+			}
+			ct := newCDCTester(ctx, t, c)
+			defer ct.Close()
+
+			ct.runTPCCWorkload(tpccArgs{warehouses: 100})
+
+			var err error
+			_, err = ct.DB().Exec("ALTER DATABASE tpcc SET PRIMARY REGION 'us-west1'")
+			require.NoError(t, err)
+
+			feed := ct.newChangefeed(feedArgs{
+				sinkType: cloudStorageSink,
+				targets:  allTpccTargets,
+				opts: map[string]string{
+					"execution_locality": "'region=us-east1'",
+				},
+			})
+			ct.waitForWorkload()
+			feed.waitForCompletion()
+
+			processors, err := getDiagramProcessors(ctx, ct.DB())
+			require.NoError(t, err)
+
+			changefeedDistribution := getChangefeedDistribution(processors, nodeToZone, t)
+			require.Greater(t, changefeedDistribution.TotalAggregators, 1)
+			for nodeIdx, spansWatched := range changefeedDistribution.NodeToSpansWatched {
+				require.LessOrEqual(t, spansWatched, changefeedDistribution.TotalSpansWatched/2, "nodeIdx %d watched %d spans, total spans watched %d", nodeIdx, spansWatched, changefeedDistribution.TotalSpansWatched)
+			}
+			require.Equal(t, 1, len(changefeedDistribution.ZoneToSpansWatched))
+			require.Equal(t, changefeedDistribution.ZoneToSpansWatched["us-east1-b"], changefeedDistribution.TotalSpansWatched)
+			zoneToLeaseHolderCount := veryifyLeaseHolderDistribution(ct.DB(), t, nodeToZone)
+			// Majority of lease holders should be in us-west1-b. Some may not, but most should.
+			if zoneToLeaseHolderCount["us-east1-b"] != 0 {
+				require.Greater(t, zoneToLeaseHolderCount["us-west1-b"]/zoneToLeaseHolderCount["us-east1-b"], 10)
+			}
+		},
+	})
 	r.Add(registry.TestSpec{
 		Name:      "cdc/initial-scan-only",
 		Owner:     registry.OwnerCDC,

pkg/cmd/roachtest/tests/cdc.go

@@ -105,7 +105,7 @@ func scanThreshold(t *testing.T, d *datadriven.TestData) (th assertion.Threshold
 		th.ThresholdType = assertion.UpperBound
 		return th
 	}
-	scanArg(t, d, "lower_bound", &th.Value)
+	scanMustExist(t, d, "lower_bound", &th.Value)
 	th.ThresholdType = assertion.LowerBound
 	return th
 }

pkg/kv/kvserver/asim/tests/helpers_test.go

@@ -655,6 +655,7 @@ go_test(
         "backfill_test.go",
         "builtin_mem_usage_test.go",
         "builtin_test.go",
+        "check_external_connection_test.go",
         "check_test.go",
         "closed_session_cache_test.go",
         "comment_on_column_test.go",