Merge #149261 #151745 #151749 #151751

149261: crosscluster/logical: auto-update config on external conn changes r=edwardguo-crl a=edwardguo-crl

 Release note (sql change): Automatically reflect external connection changes in logical clusters.
 
Epic: none

Reference: #98610

151745: mixedversion: don't mark shared process tenant availability in monitor r=Dev-Kyle,herkolategan a=DarrylWong

The monitor does not support monitoring shared process tenants in isolation, i.e. it can only monitor both system and the secondary shared process.

However, the mixed version framework was reporting availability of shared process tenants to the monitor. This broke an assumption that the monitor makes that any registered process is either the system tenant or a separate process tenant. This would cause the monitor to mistakenly believe there was a running separate process tenant.

Fixes: #151450

151749: roachtest: update rust_postgres test for LTREE r=rafiss a=rafiss

Two tests are passing now that support for LTREE was added.

fixes #151637
Release note: None

151751: bench/rttanalysis: skip BenchmarkShowGrants under short config r=rafiss a=rafiss

Previously, this benchmark was included in CI runs and would take more than 20 minutes to complete a single iteration (--bench-time=1s --count=1).

This is because the benchmark aims for a measured runtime of 1 second, but the unmeasured operations within each iteration take over 20 minutes combined to reach that target.

This change skips the benchmark in CI runs.

Fixes: #151653

Release note: None

Co-authored-by: Edward Guo <[email protected]>
Co-authored-by: DarrylWong <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 4 people

pkg/bench/rttanalysis/grant_revoke_role_bench_test.go

@@ -5,7 +5,11 @@
 
 package rttanalysis
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
+)
 
 func BenchmarkGrantRole(b *testing.B) { reg.Run(b) }
 func init() {
@@ -28,7 +32,11 @@ CREATE ROLE c;`,
 	})
 }
 
-func BenchmarkShowGrants(b *testing.B) { reg.Run(b) }
+func BenchmarkShowGrants(b *testing.B) {
+	skip.UnderShort(b, "skipping long benchmark")
+	reg.Run(b)
+}
+
 func init() {
 	reg.Register("ShowGrants", []RoundTripBenchTestCase{
 		{

pkg/cmd/roachtest/roachtestutil/mixedversion/helper.go

@@ -97,13 +97,19 @@ func (s *Service) RandomDB(rng *rand.Rand) (int, *gosql.DB) {
 	return node, s.Connect(node)
 }
 
-// prepareQuery returns a connection to one of the `nodes` provided
-// and logs the query and gateway node in the step's log file. Called
+// prepareQuery returns a connection to one of the available nodes in `nodes`
+// provided and logs the query and gateway node in the step's log file. Called
 // before the query is actually performed.
 func (s *Service) prepareQuery(
 	rng *rand.Rand, nodes option.NodeListOption, query string, args ...any,
 ) (*gosql.DB, error) {
 	availableNodes := s.AvailableNodes().Intersect(nodes)
+	if len(availableNodes) == 0 {
+		return nil, errors.Newf(
+			"no available nodes in the intersection of %s and %s",
+			s.AvailableNodes(), nodes,
+		)
+	}
 	node := availableNodes.SeededRandNode(rng)[0]
 	db := s.Connect(node)
 
@@ -137,8 +143,7 @@ func (s *Service) Exec(rng *rand.Rand, query string, args ...interface{}) error
 func (s *Service) ExecWithGateway(
 	rng *rand.Rand, nodes option.NodeListOption, query string, args ...interface{},
 ) error {
-	availableNodes := s.AvailableNodes().Intersect(nodes)
-	db, err := s.prepareQuery(rng, availableNodes, query, args...)
+	db, err := s.prepareQuery(rng, nodes, query, args...)
 	if err != nil {
 		return err
 	}

pkg/cmd/roachtest/roachtestutil/mixedversion/steps.go

@@ -903,7 +903,7 @@ func (s networkPartitionInjectStep) Run(
 	ctx context.Context, l *logger.Logger, _ *rand.Rand, h *Helper,
 ) error {
 	h.runner.monitor.ExpectProcessDead(s.targetNode)
-	if h.Tenant != nil {
+	if h.DeploymentMode() == SeparateProcessDeployment {
 		opt := option.VirtualClusterName(h.Tenant.Descriptor.Name)
 		h.runner.monitor.ExpectProcessDead(s.targetNode, opt)
 	}
@@ -958,7 +958,7 @@ func (s networkPartitionRecoveryStep) Run(
 	}
 
 	h.runner.monitor.ExpectProcessAlive(s.targetNode)
-	if h.Tenant != nil {
+	if h.DeploymentMode() == SeparateProcessDeployment {
 		opt := option.VirtualClusterName(h.Tenant.Descriptor.Name)
 		h.runner.monitor.ExpectProcessAlive(s.targetNode, opt)
 	}

pkg/cmd/roachtest/tests/rust_postgres_blocklist.go

@@ -56,8 +56,6 @@ var rustPostgresBlocklist = blocklist{
 	"types.enum_":                           "experimental feature - https://github.com/cockroachdb/cockroach/issues/46260",
 	"types.lquery":                          "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
 	"types.lquery_any":                      "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
-	"types.ltree":                           "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
-	"types.ltree_any":                       "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
 	"types.ltxtquery":                       "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
 	"types.ltxtquery_any":                   "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
 	"types.test_array_vec_params":           "default int size (int4 vs int8) mismatch",

pkg/crosscluster/logical/BUILD.bazel

@@ -34,6 +34,7 @@ go_library(
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/utilccl",
         "//pkg/cloud",
+        "//pkg/cloud/externalconn",
         "//pkg/clusterversion",
         "//pkg/crosscluster",
         "//pkg/crosscluster/physical",

pkg/crosscluster/logical/logical_replication_job.go

@@ -8,9 +8,11 @@ package logical
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/physical"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationutils"
@@ -279,7 +281,9 @@ func (r *logicalReplicationResumer) ingest(
 		stopReplanner()
 	}()
 
+	confPoller := make(chan struct{})
 	execPlan := func(ctx context.Context) error {
+		defer close(confPoller)
 		rh := rowHandler{
 			replicatedTimeAtStart: replicatedTimeAtStart,
 			frontier:              frontier,
@@ -322,6 +326,36 @@ func (r *logicalReplicationResumer) ingest(
 		return err
 	}
 
+	refreshConn := func(ctx context.Context) error {
+		ingestionJob := r.job
+		details := ingestionJob.Details().(jobspb.LogicalReplicationDetails)
+		resolvedDest, err := resolveDest(ctx, jobExecCtx.ExecCfg(), details.SourceClusterConnUri)
+		if err != nil {
+			return err
+		}
+		pollingInterval := 2 * time.Minute
+		if knobs := jobExecCtx.ExecCfg().StreamingTestingKnobs; knobs != nil && knobs.ExternalConnectionPollingInterval != nil {
+			pollingInterval = *knobs.ExternalConnectionPollingInterval
+		}
+		t := time.NewTicker(pollingInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-confPoller:
+				return nil
+			case <-t.C:
+				newDest, err := reloadDest(ctx, ingestionJob.ID(), jobExecCtx.ExecCfg())
+				if err != nil {
+					log.Warningf(ctx, "failed to check for updated configuration: %v", err)
+				} else if newDest != resolvedDest {
+					return errors.Mark(errors.Newf("replan due to detail change: old=%s, new=%s", resolvedDest, newDest), sql.ErrPlanChanged)
+				}
+			}
+		}
+	}
+
 	defer func() {
 		if l := payload.MetricsLabel; l != "" {
 			metrics.LabeledScanningRanges.Update(map[string]string{"label": l}, 0)
@@ -331,7 +365,7 @@ func (r *logicalReplicationResumer) ingest(
 		metrics.CatchupRanges.Update(0)
 	}()
 
-	err = ctxgroup.GoAndWait(ctx, execPlan, replanner, startHeartbeat)
+	err = ctxgroup.GoAndWait(ctx, execPlan, replanner, startHeartbeat, refreshConn)
 	if errors.Is(err, sql.ErrPlanChanged) {
 		metrics.ReplanCount.Inc(1)
 	}
@@ -890,6 +924,7 @@ func (r *logicalReplicationResumer) ingestWithRetries(
 	ro := getRetryPolicy(execCtx.ExecCfg().StreamingTestingKnobs)
 	var err error
 	var lastReplicatedTime hlc.Timestamp
+
 	for retrier := retry.Start(ro); retrier.Next(); {
 		err = r.ingest(ctx, execCtx)
 		if err == nil {
@@ -1054,6 +1089,35 @@ func getRetryPolicy(knobs *sql.StreamingTestingKnobs) retry.Options {
 	}
 }
 
+func resolveDest(
+	ctx context.Context, execCfg *sql.ExecutorConfig, sourceURI string,
+) (string, error) {
+	u, err := url.Parse(sourceURI)
+	if err != nil {
+		return "", err
+	}
+
+	resolved := ""
+	err = execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		conn, err := externalconn.LoadExternalConnection(ctx, u.Host, txn)
+		if err != nil {
+			return err
+		}
+		resolved = conn.UnredactedConnectionStatement()
+		return nil
+	})
+	return resolved, err
+}
+
+func reloadDest(ctx context.Context, id jobspb.JobID, execCfg *sql.ExecutorConfig) (string, error) {
+	reloadedJob, err := execCfg.JobRegistry.LoadJob(ctx, id)
+	if err != nil {
+		return "", err
+	}
+	newDetails := reloadedJob.Details().(jobspb.LogicalReplicationDetails)
+	return resolveDest(ctx, execCfg, newDetails.SourceClusterConnUri)
+}
+
 func init() {
 	m := MakeMetrics(base.DefaultHistogramWindowInterval())
 	jobs.RegisterConstructor(

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -2443,6 +2443,102 @@ func TestLogicalReplicationGatewayRoute(t *testing.T) {
 	require.Empty(t, progress.Details.(*jobspb.Progress_LogicalReplication).LogicalReplication.PartitionConnUris)
 }
 
+// TestAlterExternalConnection tests that logical replication streams can
+// dynamically switch between different source nodes when the external
+// connection URI is updated. It verifies that data continues to replicate
+// correctly after the connection change.
+func TestAlterExternalConnection(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t)
+	skip.UnderRace(t)
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	pollingInterval := 100 * time.Millisecond
+
+	clusterArgs := base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			DefaultTestTenant: base.TestControlsTenantsExplicitly,
+			Knobs: base.TestingKnobs{
+				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
+				Streaming: &sql.StreamingTestingKnobs{
+					ExternalConnectionPollingInterval: &pollingInterval,
+				},
+			},
+		},
+	}
+
+	activeLogicalSessionSQL := "SELECT count(*) > 0 FROM crdb_internal.node_sessions WHERE application_name like '$ internal repstream job id=%' AND status='ACTIVE'"
+	countLogicalSessionSQL := "SELECT count(*) FROM crdb_internal.node_sessions WHERE application_name like '$ internal repstream job id=%' AND status='ACTIVE'"
+	server, node0, runners, dbNames := setupServerWithNumDBs(t, ctx, clusterArgs, 3, 2)
+	defer server.Stopper().Stop(ctx)
+
+	dbA := runners[0]
+	dbB := runners[1]
+
+	dbANode0URL, cleanup := node0.PGUrl(t, serverutils.DBName(dbNames[0]))
+	defer cleanup()
+	dbANode0 := sqlutils.MakeSQLRunner(node0.SQLConn(t, serverutils.DBName(dbNames[0])))
+	node1 := server.Server(1).ApplicationLayer()
+	dbANode1URL, cleanup := node1.PGUrl(t, serverutils.DBName(dbNames[0]))
+	defer cleanup()
+	dbANode1 := sqlutils.MakeSQLRunner(node1.SQLConn(t, serverutils.DBName(dbNames[0])))
+
+	q0 := dbANode0URL.Query()
+	q0.Set(streamclient.RoutingModeKey, string(streamclient.RoutingModeGateway))
+	dbANode0URL.RawQuery = q0.Encode()
+
+	q1 := dbANode1URL.Query()
+	q1.Set(streamclient.RoutingModeKey, string(streamclient.RoutingModeGateway))
+	dbANode1URL.RawQuery = q1.Encode()
+
+	// We want to make sure operations for cluster B is on seperate node from cluster A.
+	node2 := server.Server(2).ApplicationLayer()
+	dbBNode2 := sqlutils.MakeSQLRunner(node2.SQLConn(t, serverutils.DBName(dbNames[1])))
+
+	require.NotEqual(t, dbANode0URL.String(), dbANode1URL.String())
+
+	externalConnName := "test_conn"
+	dbBNode2.Exec(t, fmt.Sprintf("CREATE EXTERNAL CONNECTION '%s' AS '%s'", externalConnName, dbANode0URL.String()))
+
+	var jobID jobspb.JobID
+	dbBNode2.QueryRow(t, fmt.Sprintf(
+		"CREATE LOGICAL REPLICATION STREAM FROM TABLE tab ON 'external://%s' INTO TABLE tab",
+		externalConnName)).Scan(&jobID)
+
+	dbANode0.Exec(t, "INSERT INTO tab VALUES (1, 'via_node_0')")
+
+	now := node0.Clock().Now()
+	WaitUntilReplicatedTime(t, now, dbB, jobID)
+
+	dbANode0.CheckQueryResults(t,
+		activeLogicalSessionSQL,
+		[][]string{{"true"}})
+	dbANode1.CheckQueryResults(t,
+		countLogicalSessionSQL,
+		[][]string{{"0"}})
+
+	dbBNode2.CheckQueryResults(t, "SELECT * FROM tab WHERE pk = 1", [][]string{
+		{"1", "via_node_0"},
+	})
+
+	dbBNode2.Exec(t, fmt.Sprintf("ALTER EXTERNAL CONNECTION '%s' AS '%s'", externalConnName, dbANode1URL.String()))
+	dbANode1.CheckQueryResultsRetry(t,
+		activeLogicalSessionSQL,
+		[][]string{{"true"}})
+	dbANode0.CheckQueryResultsRetry(t,
+		countLogicalSessionSQL,
+		[][]string{{"0"}})
+
+	dbA.Exec(t, "INSERT INTO tab VALUES (2, 'via_node_1')")
+	now = node0.Clock().Now()
+	WaitUntilReplicatedTime(t, now, dbB, jobID)
+
+	dbBNode2.CheckQueryResults(t, "SELECT * FROM tab WHERE pk = 2", [][]string{
+		{"2", "via_node_1"},
+	})
+}
+
 func TestMismatchColIDs(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	skip.UnderDeadlock(t)

pkg/sql/exec_util.go

@@ -2048,6 +2048,10 @@ type StreamingTestingKnobs struct {
 	// for whether the job record is updated on a progress update.
 	CutoverProgressShouldUpdate func() bool
 
+	//ExternalConnectionPollingInterval override the LDR alter
+	// connection polling frequency.
+	ExternalConnectionPollingInterval *time.Duration
+
 	DistSQLRetryPolicy *retry.Options
 
 	AfterRetryIteration func(err error)