kv: add TestStoreRangeSplitAndMergeWithGlobalReads

Made possible by #60567.

This commit adds a new test called TestStoreRangeSplitAndMergeWithGlobalReads
that tests that a range configured to serve global reads can be split and
merged. In essence, this tests whether the split and merge transactions can
handle having their timestamp bumped by the closed timestamp on the ranges
they're operating on.

The test revealed that range merges did have issues in these cases, because
SubsumeRequests assumed that the intent on the RHS's local descriptor was below
the node's HLC clock. This is no longer always true, so we now perform the
inconsistent scan at hlc.MaxTimestamp, just like QueryIntent requests do.

pkg/kv/kvserver/batcheval/cmd_subsume.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/errors"
 )
 
@@ -90,16 +91,18 @@ func Subsume(
 	}
 
 	// Sanity check the caller has initiated a merge transaction by checking for
-	// a deletion intent on the local range descriptor.
+	// a deletion intent on the local range descriptor. Read inconsistently at
+	// the maximum timestamp to ensure that we see an intent if one exists,
+	// regardless of what timestamp it is written at.
 	descKey := keys.RangeDescriptorKey(desc.StartKey)
-	_, intent, err := storage.MVCCGet(ctx, readWriter, descKey, cArgs.Header.Timestamp,
+	_, intent, err := storage.MVCCGet(ctx, readWriter, descKey, hlc.MaxTimestamp,
 		storage.MVCCGetOptions{Inconsistent: true})
 	if err != nil {
 		return result.Result{}, errors.Errorf("fetching local range descriptor: %s", err)
 	} else if intent == nil {
 		return result.Result{}, errors.AssertionFailedf("range missing intent on its local descriptor")
 	}
-	val, _, err := storage.MVCCGetAsTxn(ctx, readWriter, descKey, cArgs.Header.Timestamp, intent.Txn)
+	val, _, err := storage.MVCCGetAsTxn(ctx, readWriter, descKey, intent.Txn.WriteTimestamp, intent.Txn)
 	if err != nil {
 		return result.Result{}, errors.Errorf("fetching local range descriptor as txn: %s", err)
 	} else if val != nil {

pkg/kv/kvserver/client_split_test.go

@@ -3545,3 +3545,60 @@ func TestSplitBlocksReadsToRHS(t *testing.T) {
 	}
 	require.Nil(t, g.Wait())
 }
+
+// TestStoreRangeSplitAndMergeWithGlobalReads tests that a range configured to
+// serve global reads can be split and merged. In essence, this tests whether
+// the split and merge transactions can handle having their timestamp bumped by
+// the closed timestamp on the ranges they're operating on.
+func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	serv, _, _ := serverutils.StartServer(t, base.TestServerArgs{
+		Knobs: base.TestingKnobs{
+			Store: &kvserver.StoreTestingKnobs{
+				DisableMergeQueue: true,
+			},
+		},
+	})
+	s := serv.(*server.TestServer)
+	defer s.Stopper().Stop(ctx)
+	store, err := s.Stores().GetStore(s.GetFirstStoreID())
+	require.NoError(t, err)
+	config.TestingSetupZoneConfigHook(s.Stopper())
+
+	// Set global reads.
+	descID := uint32(keys.MinUserDescID)
+	descKey := keys.SystemSQLCodec.TablePrefix(descID)
+	zoneConfig := zonepb.DefaultZoneConfig()
+	zoneConfig.GlobalReads = proto.Bool(true)
+	config.TestingSetZoneConfig(config.SystemTenantObjectID(descID), zoneConfig)
+
+	// Trigger gossip callback and wait for propagation
+	require.NoError(t, store.Gossip().AddInfoProto(gossip.KeySystemConfig, &config.SystemConfigEntries{}, 0))
+	testutils.SucceedsSoon(t, func() error {
+		repl := store.LookupReplica(roachpb.RKey(descKey))
+		if repl.ClosedTimestampPolicy() != roachpb.LEAD_FOR_GLOBAL_READS {
+			return errors.Errorf("expected LEAD_FOR_GLOBAL_READS policy")
+		}
+		return nil
+	})
+
+	// Split the range. Should succeed.
+	splitKey := append(descKey, []byte("split")...)
+	splitArgs := adminSplitArgs(splitKey)
+	_, pErr := kv.SendWrapped(ctx, store.TestSender(), splitArgs)
+	require.Nil(t, pErr)
+
+	repl := store.LookupReplica(roachpb.RKey(splitKey))
+	require.Equal(t, splitKey, repl.Desc().StartKey.AsRawKey())
+
+	// Merge the range. Should succeed.
+	mergeArgs := adminMergeArgs(descKey)
+	_, pErr = kv.SendWrapped(ctx, store.TestSender(), mergeArgs)
+	require.Nil(t, pErr)
+
+	repl = store.LookupReplica(roachpb.RKey(splitKey))
+	require.Equal(t, descKey, repl.Desc().StartKey.AsRawKey())
+}

pkg/kv/kvserver/helpers_test.go

@@ -514,6 +514,14 @@ func (r *Replica) ReadProtectedTimestamps(ctx context.Context) {
 	ts = r.readProtectedTimestampsRLocked(ctx, nil /* f */)
 }
 
+// ClosedTimestampPolicy returns the closed timestamp policy of the range, which
+// is updated asynchronously through gossip of zone configurations.
+func (r *Replica) ClosedTimestampPolicy() roachpb.RangeClosedTimestampPolicy {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.closedTimestampPolicyRLocked()
+}
+
 // GetCircuitBreaker returns the circuit breaker controlling
 // connection attempts to the specified node.
 func (t *RaftTransport) GetCircuitBreaker(

pkg/kv/kvserver/replica.go

@@ -722,6 +722,10 @@ func (r *Replica) descRLocked() *roachpb.RangeDescriptor {
 	return r.mu.state.Desc
 }
 
+// closedTimestampPolicyRLocked returns the closed timestamp policy of the
+// range, which is updated asynchronously through gossip of zone configurations.
+// NOTE: an exported version of this method which does not require the replica
+// lock exists in helpers_test.go. Move here if needed.
 func (r *Replica) closedTimestampPolicyRLocked() roachpb.RangeClosedTimestampPolicy {
 	if r.mu.zone.GlobalReads != nil && *r.mu.zone.GlobalReads {
 		return roachpb.LEAD_FOR_GLOBAL_READS