Merge #141735 #142186 #142489 #142829 #142836

141735: sql: enable tenant testing for zone tests r=rafiss a=shubhamdhama

`` sql: enable tenant testing for `zone_test.go` ``

Updating `TestValidSetShowZones` to use
`createTestServerParamsAllowTenants` instead of `createTestServerParams`
required some investigation. It turned out simpler than expected. Zone
settings for meta ranges aren't supported for secondary tenants, so this
change simply exclude them from setup and validating.

Informs: #140446
Epic: CRDB-48357
Release note: None

`` sql: enable tenant testing for `zone_config_test` ``

Since `GetSpanConfigForKey` is only available in system tenant, it is not
used when running under secondary tenants.

142186: kvserver: mark replica as unavailable if leaderless for a long time r=iskettaneh a=iskettaneh

This PR marks the replica as unavailable in if it has been leaderless for a duration above:
`kv.replica_raft.leaderless_unavailable_threshold`. This helps requests to bail early on unavailable ranges without relying on the replica circuit breaker to trip. This has multiple benefits:

1) Faster reaction time than the replica circuit breaker: If two nodes fail (assuming R=3), many ranges will become unavailable. With the replica circuit breaker, a scan query will need to trip the circuit
breaker for multiple replicas, causing added delays. However, with this approach, the replica will basically become unavailable on the tick path (rather than on the request path). Meaning that a scan query wouldn't need to trip one circuit breaker after the other.

2) Lighter weight: Instead of relying on the replica circuit breaker to test the replication pipeline before it marks the range as
available again, this approach relies on the Raft signal to know when there is a leader again, indicating that the range is
available again.

3) With leader leases, a replica won't propose a lease if it's not the leader. This means that with leader leases, the replica circuit breaker might not trip if the range have lost quorum. However, with this commit, the replica will eventually forget who the leader was, and eventually the leaderlessWatcher would mark it as unavailable.

Fixes: #139638

Release note: None

142489: row, sql: implement mutations on vector indexes r=mw5h a=mw5h

This patch plumbs the output of the vector search operators into rowenc for encoding into vector indexes. The output of these vector search operators is included in the row values for mutation operators after partial index values and are plumbed into a new vector index update helper, which tracks the column values until they're needed by rowenc.

While we're here, we homogonize how
pkg/sql/{delete,insert,update,upsert}.go consume row values, hopefully improving legibility.

Epic: CRDB-42943
Release note: None

142829: sql/catalog: fix object renames for PCR reader catalogs r=fqazi a=fqazi

Previously, the PCR reader catalog would only delete namespace entries for a descriptor if it was not modified. This meant the reader catalog logic could leave behind stale entries in the system.namespace table after a object was renamed. To address this, this patch detects if an object is renamed, and allows the old namespace entry to be deleted.

Fixes: #142828

Release note (bug fix): PCR reader catalogs could have orphan rows in system.namespace after a object is renamed.

142836: kvserver: deflake TestLeasePreferencesDuringOutage r=kvoli a=arulajmani

There was a race here, when heartbeating node liveness epochs, where another node could increment our epoch. This patch retries in such situations.

Fixes #142795

Release note: None

Co-authored-by: Shubham Dhama <[email protected]>
Co-authored-by: Ibrahim Kettaneh <[email protected]>
Co-authored-by: Matt White <[email protected]>
Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Arul Ajmani <[email protected]>

Author: 6 people

docs/generated/settings/settings.html

@@ -116,6 +116,7 @@
 <tr><td><div id="setting-kv-rangefeed-closed-timestamp-refresh-interval" class="anchored"><code>kv.rangefeed.closed_timestamp_refresh_interval</code></div></td><td>duration</td><td><code>3s</code></td><td>the interval at which closed-timestamp updatesare delivered to rangefeeds; set to 0 to use kv.closed_timestamp.side_transport_interval</td><td>Dedicated/Self-hosted (read-write); Serverless (read-only)</td></tr>
 <tr><td><div id="setting-kv-rangefeed-enabled" class="anchored"><code>kv.rangefeed.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>if set, rangefeed registration is enabled</td><td>Dedicated/Self-hosted (read-write); Serverless (read-only)</td></tr>
 <tr><td><div id="setting-kv-replica-circuit-breaker-slow-replication-threshold" class="anchored"><code>kv.replica_circuit_breaker.slow_replication_threshold</code></div></td><td>duration</td><td><code>1m0s</code></td><td>duration after which slow proposals trip the per-Replica circuit breaker (zero duration disables breakers)</td><td>Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-kv-replica-raft-leaderless-unavailable-threshold" class="anchored"><code>kv.replica_raft.leaderless_unavailable_threshold</code></div></td><td>duration</td><td><code>1m0s</code></td><td>duration after which leaderless replicas is considered unavailable. Set to 0 to disable leaderless replica availability checks</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-replica-stats-addsst-request-size-factor" class="anchored"><code>kv.replica_stats.addsst_request_size_factor</code></div></td><td>integer</td><td><code>50000</code></td><td>the divisor that is applied to addsstable request sizes, then recorded in a leaseholders QPS; 0 means all requests are treated as cost 1</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-replication-reports-interval" class="anchored"><code>kv.replication_reports.interval</code></div></td><td>duration</td><td><code>1m0s</code></td><td>the frequency for generating the replication_constraint_stats, replication_stats_report and replication_critical_localities reports (set to 0 to disable)</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-snapshot-rebalance-max-rate" class="anchored"><code>kv.snapshot_rebalance.max_rate</code></div></td><td>byte size</td><td><code>32 MiB</code></td><td>the rate limit (bytes/sec) to use for rebalance and upreplication snapshots</td><td>Dedicated/Self-Hosted</td></tr>

pkg/ccl/partitionccl/zone_test.go

@@ -50,9 +50,7 @@ func TestValidIndexPartitionSetShowZones(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-	})
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
 	defer s.Stopper().Stop(context.Background())
 
 	sqlDB := sqlutils.MakeSQLRunner(db)
@@ -253,9 +251,7 @@ func TestInvalidIndexPartitionSetShowZones(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-	})
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
 	defer s.Stopper().Stop(context.Background())
 
 	for i, tc := range []struct {

pkg/crosscluster/logical/lww_kv_processor.go

@@ -476,13 +476,15 @@ func (p *kvTableWriter) insertRow(ctx context.Context, b *kv.Batch, after cdceve
 
 	var ph row.PartialIndexUpdateHelper
 	// TODO(dt): support partial indexes.
+	var vh row.VectorIndexUpdateHelper
+	// TODO(mw5h, drewk): support vector indexes.
 	oth := &row.OriginTimestampCPutHelper{
 		OriginTimestamp: after.MvccTimestamp,
 		// TODO(ssd): We should choose this based by comparing the cluster IDs of the source
 		// and destination clusters.
 		// ShouldWinTie: true,
 	}
-	return p.ri.InsertRow(ctx, &row.KVBatchAdapter{Batch: b}, p.newVals, ph, oth, row.CPutOp, false /* traceKV */)
+	return p.ri.InsertRow(ctx, &row.KVBatchAdapter{Batch: b}, p.newVals, ph, vh, oth, row.CPutOp, false /* traceKV */)
 }
 
 func (p *kvTableWriter) updateRow(
@@ -497,13 +499,15 @@ func (p *kvTableWriter) updateRow(
 
 	var ph row.PartialIndexUpdateHelper
 	// TODO(dt): support partial indexes.
+	var vh row.VectorIndexUpdateHelper
+	// TODO(mw5h, drewk): support vector indexes.
 	oth := &row.OriginTimestampCPutHelper{
 		OriginTimestamp: after.MvccTimestamp,
 		// TODO(ssd): We should choose this based by comparing the cluster IDs of the source
 		// and destination clusters.
 		// ShouldWinTie: true,
 	}
-	_, err := p.ru.UpdateRow(ctx, b, p.oldVals, p.newVals, ph, oth, false)
+	_, err := p.ru.UpdateRow(ctx, b, p.oldVals, p.newVals, ph, vh, oth, false)
 	return err
 }
 
@@ -516,6 +520,8 @@ func (p *kvTableWriter) deleteRow(
 
 	var ph row.PartialIndexUpdateHelper
 	// TODO(dt): support partial indexes.
+	var vh row.VectorIndexUpdateHelper
+	// TODO(mw5h, drewk): support vector indexes.
 	oth := &row.OriginTimestampCPutHelper{
 		PreviousWasDeleted: before.IsDeleted(),
 		OriginTimestamp:    after.MvccTimestamp,
@@ -524,7 +530,7 @@ func (p *kvTableWriter) deleteRow(
 		// ShouldWinTie: true,
 	}
 
-	return p.rd.DeleteRow(ctx, b, p.oldVals, ph, oth, false)
+	return p.rd.DeleteRow(ctx, b, p.oldVals, ph, vh, oth, false)
 }
 
 func (p *kvTableWriter) fillOld(vals cdcevent.Row) error {

pkg/kv/kvserver/client_lease_test.go

@@ -1178,7 +1178,17 @@ func TestLeasePreferencesDuringOutage(t *testing.T) {
 			// allocator on server 0 may see everyone as temporarily dead due to the
 			// clock move above.
 			for _, i := range []int{0, 3, 4} {
-				require.NoError(t, tc.Servers[i].HeartbeatNodeLiveness())
+				testutils.SucceedsSoon(t, func() error {
+					err := tc.Servers[i].HeartbeatNodeLiveness()
+					if err != nil {
+						if errors.Is(err, liveness.ErrEpochIncremented) {
+							t.Logf("retrying heartbeat after err %s", err)
+							return err
+						}
+						t.Fatalf("unexpected error heartbeating liveness record for server %d: %s", i, err)
+					}
+					return nil
+				})
 				require.NoError(t, tc.GetFirstStoreFromServer(t, i).GossipStore(ctx, true))
 			}
 		}

pkg/kv/kvserver/client_replica_test.go

@@ -39,6 +39,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts/ptutil"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/raftutil"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/stateloader"
+	"github.com/cockroachdb/cockroach/pkg/raft"
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/server"
@@ -2728,6 +2729,142 @@ func TestChangeReplicasGeneration(t *testing.T) {
 	assert.EqualValues(t, repl.Desc().Generation, oldGeneration+3, "\nold: %+v\nnew: %+v", oldDesc, newDesc)
 }
 
+// TestLossQuorumCauseLeaderlessWatcherToSignalUnavailable checks that if a range
+// lost its quorum, the remaining replicas in that range will have their
+// leaderlessWatcher indicate that the range is unavailable. Also, it checks
+// that when the range regains quorum, the leaderlessWatcher will indicate that
+// the range is available.
+func TestLossQuorumCauseLeaderWatcherToSignalUnavailable(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	manualClock := hlc.NewHybridManualClock()
+
+	stickyVFSRegistry := fs.NewStickyRegistry()
+	lisReg := listenerutil.NewListenerRegistry()
+	defer lisReg.Close()
+
+	// Perform a basic test setup with two nodes.
+	const numServers int = 2
+	st := cluster.MakeTestingClusterSettings()
+
+	// Set `kv.replica_raft.leaderless_unavailable_threshold` to 10 seconds.
+	threshold := time.Second * 10
+	kvserver.ReplicaLeaderlessUnavailableThreshold.Override(ctx, &st.SV, threshold)
+
+	stickyServerArgs := make(map[int]base.TestServerArgs)
+	for i := 0; i < numServers; i++ {
+		stickyServerArgs[i] = base.TestServerArgs{
+			Settings: st,
+			StoreSpecs: []base.StoreSpec{
+				{
+					InMemory:    true,
+					StickyVFSID: strconv.FormatInt(int64(i), 10),
+				},
+			},
+			Knobs: base.TestingKnobs{
+				Server: &server.TestingKnobs{
+					StickyVFSRegistry: stickyVFSRegistry,
+				},
+			},
+		}
+	}
+
+	tc := testcluster.StartTestCluster(t, numServers,
+		base.TestClusterArgs{
+			ReplicationMode:     base.ReplicationManual,
+			ReusableListenerReg: lisReg,
+			ServerArgsPerNode:   stickyServerArgs,
+		})
+
+	defer tc.Stopper().Stop(ctx)
+
+	key := tc.ScratchRange(t)
+	tc.AddVotersOrFatal(t, key, tc.Targets(1)...)
+	desc, err := tc.LookupRange(key)
+	require.NoError(t, err)
+
+	// Randomly stop server index 0 or 1.
+	stoppedNodeInx := rand.Intn(2)
+	aliveNodeIdx := 1 - stoppedNodeInx
+	tc.StopServer(stoppedNodeInx)
+	repl := tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+
+	// The range is available initially.
+	require.False(t, repl.LeaderlessWatcher.IsUnavailable())
+
+	// Wait until the remaining replica becomes leaderless.
+	testutils.SucceedsSoon(t, func() error {
+		if repl.RaftStatus().Lead != raft.None {
+			return errors.New("Leader still exists")
+		}
+		return nil
+	})
+
+	// Increment the clock by the leaderlessWatcher unavailable threshold.
+	manualClock.Increment(threshold.Nanoseconds())
+
+	// Wait for the leaderlessWatcher to indicate that the range is unavailable.
+	testutils.SucceedsSoon(t, func() error {
+		tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+		if !repl.LeaderlessWatcher.IsUnavailable() {
+			return errors.New("range is still available")
+		}
+		return nil
+	})
+
+	sendPutRequestWithTimeout := func(repl *kvserver.Replica, timeout time.Duration) (*kvpb.Error, error) {
+		ctx, cancel := context.WithTimeout(ctx, timeout)
+		defer cancel()
+		ba := &kvpb.BatchRequest{}
+		ba.RangeID = desc.RangeID
+		ba.Timestamp = repl.Clock().Now()
+		ba.Add(putArgs(key, []byte("foo")))
+		_, pErr := repl.Send(ctx, ba)
+		return pErr, ctx.Err()
+	}
+
+	// Requests should immediately return an error indicating that the range is
+	// unavailable.
+	pErr, ctxErr := sendPutRequestWithTimeout(repl, 2*time.Second)
+	require.NoError(t, ctxErr)
+	require.Regexp(t, "replica has been leaderless for 10s", pErr)
+	require.True(t, errors.HasType(pErr.GoError(), (*kvpb.ReplicaUnavailableError)(nil)),
+		"expected ReplicaUnavailableError, got %v", err)
+
+	// At this point we know that the replica is considered unavailable. Regain
+	// the quorum and check that the leaderlessWatcher indicates that the range is
+	// available.
+	require.NoError(t, tc.RestartServer(stoppedNodeInx))
+
+	testutils.SucceedsSoon(t, func() error {
+		repl = tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+		tc.GetFirstStoreFromServer(t, aliveNodeIdx).LookupReplica(roachpb.RKey(key))
+		if repl.LeaderlessWatcher.IsUnavailable() {
+			return errors.New("range is still unavailable")
+		}
+		return nil
+	})
+
+	// Requests should now succeed. We need to try both replicas to avoid
+	// NotLeaseHolderErrors.
+	testutils.SucceedsSoon(t, func() error {
+		for i := range numServers {
+			repl := tc.GetFirstStoreFromServer(t, i).LookupReplica(roachpb.RKey(key))
+			pErr, ctxErr = sendPutRequestWithTimeout(repl, 2*time.Second)
+			if ctxErr == nil && pErr == nil {
+				return nil
+			}
+		}
+		// If we reach this point, we know that the request failed, return the
+		// error.
+		if ctxErr != nil {
+			return ctxErr
+		}
+		return pErr.GoError()
+	})
+}
+
 func TestClearRange(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/kv/kvserver/replica.go

@@ -176,6 +176,67 @@ func (c *atomicConnectionClass) set(cc rpc.ConnectionClass) {
 	atomic.StoreUint32((*uint32)(c), uint32(cc))
 }
 
+// leaderlessWatcher is a lightweight implementation of the signaller interface
+// that is used to signal when a replica doesn't know who the leader is for an
+// extended period of time. This is used to signal that the range in
+// unavailable.
+type leaderlessWatcher struct {
+	mu struct {
+		syncutil.RWMutex
+
+		// leaderlessTimestamp records the timestamp captured when the replica
+		// didn't know who the leader was. This is reset on every tick if the
+		// replica knows who the leader is.
+		leaderlessTimestamp time.Time
+
+		// unavailable is set to true if the replica is leaderless for a long time
+		// (longer than ReplicaLeaderlessUnavailableThreshold).
+		unavailable bool
+	}
+
+	// err is the error returned when the replica is leaderless for a long time.
+	err error
+
+	// closedChannel is an already closed channel. Requests will use it to know
+	// that the replica is leaderless, and can be considered unavailable. This
+	// is primarily due to implementation details of the request path, where
+	// the request grabs a signaller in signallerForBatch() and then checks if
+	// the channel is closed to determine if the replica is available.
+	closedChannel chan struct{}
+}
+
+// NewLeaderlessWatcher initializes a new leaderlessWatcher with the default
+// values.
+func NewLeaderlessWatcher(r *Replica) *leaderlessWatcher {
+	closedCh := make(chan struct{})
+	close(closedCh)
+	return &leaderlessWatcher{
+		err: r.replicaUnavailableError(
+			errors.Errorf("replica has been leaderless for %s",
+				ReplicaLeaderlessUnavailableThreshold.Get(&r.store.cfg.Settings.SV))),
+		closedChannel: closedCh,
+	}
+}
+
+func (lw *leaderlessWatcher) Err() error {
+	return lw.err
+}
+
+func (lw *leaderlessWatcher) C() <-chan struct{} {
+	return lw.closedChannel
+}
+
+func (lw *leaderlessWatcher) IsUnavailable() bool {
+	lw.mu.RLock()
+	defer lw.mu.RUnlock()
+	return lw.mu.unavailable
+}
+
+func (lw *leaderlessWatcher) resetLocked() {
+	lw.mu.leaderlessTimestamp = time.Time{}
+	lw.mu.unavailable = false
+}
+
 // ReplicaMutex is an RWMutex. It has its own type to make it easier to look for
 // usages specific to the replica mutex.
 type ReplicaMutex syncutil.RWMutex
@@ -941,6 +1002,10 @@ type Replica struct {
 		lastTickTimestamp hlc.ClockTimestamp
 	}
 
+	// LeaderlessWatcher is used to signal when a replica is leaderless for a long
+	// time.
+	LeaderlessWatcher *leaderlessWatcher
+
 	// The raft log truncations that are pending. Access is protected by its own
 	// mutex. All implementation details should be considered hidden except to
 	// the code in raft_log_truncator.go. External code should only use the

pkg/kv/kvserver/replica_backpressure.go

@@ -118,18 +118,29 @@ func canBackpressureBatch(ba *kvpb.BatchRequest) bool {
 	return false
 }
 
-// signallerForBatch returns the signaller to use for this batch. This is the
-// Replica's breaker's signaller except if any request in the batch uses
-// poison.Policy_Wait, in which case it's a neverTripSignaller. In particular,
-// `(signaller).C() == nil` signals that the request bypasses the circuit
-// breakers.
+// signallerForBatch returns the signaller to use for this batch in the
+// following priorities:
+// 1. If the batch contains a request uses poison.Policy_Wait, we will return
+// neverTripSignaller.
+// 2. If the replica is leaderless for a time longer than the threshold in
+// `kv.replica_raft.leaderless_unavailable_threshold`, use the
+// leaderlessWatcher signal.
+// 3. Otherwise, use the replica's breaker's signaller
 func (r *Replica) signallerForBatch(ba *kvpb.BatchRequest) signaller {
 	for _, ru := range ba.Requests {
 		req := ru.GetInner()
 		if kvpb.BypassesReplicaCircuitBreaker(req) {
 			return neverTripSignaller{}
 		}
 	}
+
+	// If the leaderless watcher indicates that this replica is leaderless for a
+	// long time, use it as the signal.
+	if r.LeaderlessWatcher.IsUnavailable() {
+		return r.LeaderlessWatcher
+	}
+
+	// Otherwise, use the replica's breaker.
 	return r.breaker.Signal()
 }
 

pkg/kv/kvserver/replica_init.go

@@ -245,6 +245,7 @@ func newUninitializedReplicaWithoutRaftGroup(
 	r.breaker = newReplicaCircuitBreaker(
 		store.cfg.Settings, store.stopper, r.AmbientContext, r, onTrip, onReset,
 	)
+	r.LeaderlessWatcher = NewLeaderlessWatcher(r)
 	r.mu.currentRACv2Mode = r.replicationAdmissionControlModeToUse(context.TODO())
 	r.raftMu.flowControlLevel = kvflowcontrol.GetV2EnabledWhenLeaderLevel(
 		r.raftCtx, store.ClusterSettings(), store.TestingKnobs().FlowControlTestingKnobs)