kvserver: don't apply ExcludeDataFromBackup outside config bounds

pkg/ccl/backupccl/backup_test.go

@@ -9707,7 +9707,7 @@ func TestExcludeDataFromBackupAndRestore(t *testing.T) {
 
 	var exportReqsAtomic int64
 
-	tc, sqlDB, iodir, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, 10,
+	tc, sqlDB, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, 10,
 		InitManualReplication, base.TestClusterArgs{
 			ServerArgs: base.TestServerArgs{
 				// Disabled to run within tenants because the function that sets up the restoring cluster
@@ -9734,16 +9734,6 @@ func TestExcludeDataFromBackupAndRestore(t *testing.T) {
 		})
 	defer cleanupFn()
 
-	_, restoreDB, cleanup := backupRestoreTestSetupEmpty(t, singleNode, iodir, InitManualReplication,
-		base.TestClusterArgs{
-			ServerArgs: base.TestServerArgs{
-				Knobs: base.TestingKnobs{
-					JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(), // speeds up test
-				},
-			},
-		})
-	defer cleanup()
-
 	sqlDB.Exec(t, `SET CLUSTER SETTING kv.rangefeed.enabled = true`)
 	sqlDB.Exec(t, `SET CLUSTER SETTING kv.closed_timestamp.target_duration = '100ms'`)
 	conn := tc.Conns[0]
@@ -9754,44 +9744,53 @@ func TestExcludeDataFromBackupAndRestore(t *testing.T) {
 	sqlDB.Exec(t, `BACKUP DATABASE data INTO $1`, localFoo)
 
 	sqlDB.Exec(t, `INSERT INTO data.foo select * from generate_series(6,10)`)
-	// Create another table.
-	sqlDB.Exec(t, `CREATE TABLE data.bar (id INT, INDEX bar(id))`)
-	sqlDB.Exec(t, `INSERT INTO data.bar select * from generate_series(1,10)`)
 
 	// Set foo to exclude_data_from_backup and back it up. The ExportRequest
 	// should be a noop and backup no data.
 	sqlDB.Exec(t, `ALTER TABLE data.foo SET (exclude_data_from_backup = true)`)
+	fooTableSpan := getTableSpan(t, conn, "foo", "data")
 	waitForReplicaFieldToBeSet(t, tc, conn, "foo", "data", func(r *kvserver.Replica) (bool, error) {
-		if !r.ExcludeDataFromBackup(context.Background()) {
-			return false, errors.New("waiting for the range containing table data.foo to split")
+		exclude, err := r.ExcludeDataFromBackup(context.Background(), fooTableSpan)
+		if err != nil {
+			return false, err
 		}
-		return true, nil
-	})
-	waitForReplicaFieldToBeSet(t, tc, conn, "bar", "data", func(r *kvserver.Replica) (bool, error) {
-		if r.ExcludeDataFromBackup(context.Background()) {
-			return false, errors.New("waiting for the range containing table data.bar to split")
+		if !exclude {
+			return false, errors.New("waiting for the range containing table data.foo to split")
 		}
 		return true, nil
 	})
 
+	// Create another table. Assert we don't see ExcludeDataFromBackup on it
+	sqlDB.Exec(t, `CREATE TABLE data.bar (id INT, INDEX bar(id))`)
+	_, r := getStoreAndReplica(t, tc, conn, "bar", "data")
+	barTableSpan := getTableSpan(t, conn, "bar", "data")
+	exclude, err := r.ExcludeDataFromBackup(context.Background(), barTableSpan)
+	require.NoError(t, err)
+	require.False(t, exclude, "bar should never be excluded")
+	sqlDB.Exec(t, `INSERT INTO data.bar select * from generate_series(1,10)`)
+
 	sqlDB.Exec(t, `BACKUP DATABASE data INTO LATEST IN $1`, localFoo)
 	sqlDB.Exec(t, `CREATE TABLE data.baz (id INT)`)
 	sqlDB.Exec(t, `ALTER TABLE data.baz SET (exclude_data_from_backup = true)`)
 	sqlDB.Exec(t, `INSERT INTO data.baz select * from generate_series(1,10)`)
-
+	bazTableSpan := getTableSpan(t, conn, "baz", "data")
 	waitForReplicaFieldToBeSet(t, tc, conn, "baz", "data", func(r *kvserver.Replica) (bool, error) {
-		if !r.ExcludeDataFromBackup(context.Background()) {
+		exclude, err := r.ExcludeDataFromBackup(context.Background(), bazTableSpan)
+		if err != nil {
+			return false, err
+		}
+		if !exclude {
 			return false, errors.New("waiting for the range containing table data.foo to split")
 		}
 		return true, nil
 	})
 
 	sqlDB.Exec(t, `BACKUP DATABASE data INTO LATEST IN $1`, localFoo)
 
-	restoreDB.Exec(t, `RESTORE DATABASE data FROM LATEST IN $1`, localFoo)
-	require.Len(t, restoreDB.QueryStr(t, `SELECT * FROM data.foo`), 5)
-	require.Len(t, restoreDB.QueryStr(t, `SELECT * FROM data.bar`), 10)
-	require.Len(t, restoreDB.QueryStr(t, `SELECT * FROM data.baz`), 0)
+	sqlDB.Exec(t, `RESTORE DATABASE data FROM LATEST IN $1 WITH new_db_name=data2`, localFoo)
+	require.Len(t, sqlDB.QueryStr(t, `SELECT * FROM data2.foo`), 5)
+	require.Len(t, sqlDB.QueryStr(t, `SELECT * FROM data2.bar`), 10)
+	require.Len(t, sqlDB.QueryStr(t, `SELECT * FROM data2.baz`), 0)
 
 	before := atomic.LoadInt64(&exportReqsAtomic)
 	sqlDB.Exec(t, `BACKUP data.foo TO $1`, localFoo+"/tbl")
@@ -9886,8 +9885,13 @@ func TestExportRequestBelowGCThresholdOnDataExcludedFromBackup(t *testing.T) {
 
 	_, err = conn.Exec(`ALTER TABLE foo SET (exclude_data_from_backup = true)`)
 	require.NoError(t, err)
+	fooTableSpan := getTableSpan(t, conn, "foo", "defaultdb")
 	waitForReplicaFieldToBeSet(t, tc, conn, "foo", "defaultdb", func(r *kvserver.Replica) (bool, error) {
-		if !r.ExcludeDataFromBackup(ctx) {
+		excluded, err := r.ExcludeDataFromBackup(ctx, fooTableSpan)
+		if err != nil {
+			return false, err
+		}
+		if !excluded {
 			return false, errors.New("waiting for exclude_data_from_backup to be applied")
 		}
 		return true, nil

pkg/ccl/backupccl/utils_test.go

@@ -311,11 +311,23 @@ func getTableStartKey(t *testing.T, conn *gosql.DB, tableName, dbName string) ro
 	return startKey
 }
 
+func getTableSpan(t *testing.T, conn *gosql.DB, tableName, dbName string) roachpb.Span {
+	t.Helper()
+	row := conn.QueryRow(
+		fmt.Sprintf(`SELECT crdb_internal.table_span('%s.%s'::regclass::oid::int)[1],
+crdb_internal.table_span('%[1]s.%[2]s'::regclass::oid::int)[2]`,
+			tree.NameString(dbName), tree.NameString(tableName)))
+	var sp roachpb.Span
+	require.NoError(t, row.Scan(&sp.Key, &sp.EndKey))
+	return sp
+}
+
 func getFirstStoreReplica(
 	t *testing.T, s serverutils.TestServerInterface, key roachpb.Key,
 ) (*kvserver.Store, *kvserver.Replica) {
 	t.Helper()
-	store, err := s.StorageLayer().GetStores().(*kvserver.Stores).GetStore(s.GetFirstStoreID())
+	storageLayer := s.StorageLayer()
+	store, err := storageLayer.GetStores().(*kvserver.Stores).GetStore(storageLayer.GetFirstStoreID())
 	require.NoError(t, err)
 	var repl *kvserver.Replica
 	testutils.SucceedsSoon(t, func() error {

pkg/config/system.go

@@ -333,15 +333,15 @@ func (s *SystemConfig) getZoneConfigForKey(
 	return id, s.DefaultZoneConfig, nil
 }
 
-// GetSpanConfigForKey looks of the span config for the given key. It's part of
-// spanconfig.StoreReader interface. Note that it is only usable for the system
-// tenant config.
+// GetSpanConfigForKey looks of the span config for the given key and the bounds
+// that span the configuration applies to. It's part of spanconfig.StoreReader
+// interface. Note that it is only usable for the system tenant config.
 func (s *SystemConfig) GetSpanConfigForKey(
 	ctx context.Context, key roachpb.RKey,
-) (roachpb.SpanConfig, error) {
+) (roachpb.SpanConfig, roachpb.Span, error) {
 	id, zone, err := s.getZoneConfigForKey(keys.SystemSQLCodec, key)
 	if err != nil {
-		return roachpb.SpanConfig{}, err
+		return roachpb.SpanConfig{}, roachpb.Span{}, err
 	}
 	spanConfig := zone.AsSpanConfig()
 	if id <= keys.MaxReservedDescID {
@@ -354,7 +354,8 @@ func (s *SystemConfig) GetSpanConfigForKey(
 		// applicable to user tables.
 		spanConfig.GCPolicy.IgnoreStrictEnforcement = true
 	}
-	return spanConfig, nil
+	prefix := keys.SystemSQLCodec.TablePrefix(uint32(id))
+	return spanConfig, roachpb.Span{Key: prefix, EndKey: prefix.PrefixEnd()}, nil
 }
 
 // DecodeKeyIntoZoneIDAndSuffix figures out the zone that the key belongs to.

pkg/config/system_test.go

@@ -616,7 +616,7 @@ func TestGetZoneConfigForKey(t *testing.T) {
 			objectID = id
 			return cfg.DefaultZoneConfig, nil, false, nil
 		}
-		_, err := cfg.GetSpanConfigForKey(ctx, tc.key)
+		_, _, err := cfg.GetSpanConfigForKey(ctx, tc.key)
 		if err != nil {
 			t.Errorf("#%d: GetSpanConfigForKey(%v) got error: %v", tcNum, tc.key, err)
 		}

pkg/kv/kvserver/asim/state/impl.go

@@ -1237,12 +1237,15 @@ func (s *state) ComputeSplitKey(
 // SpanConfigConformanceReport.
 func (s *state) GetSpanConfigForKey(
 	ctx context.Context, key roachpb.RKey,
-) (roachpb.SpanConfig, error) {
+) (roachpb.SpanConfig, roachpb.Span, error) {
 	rng := s.rangeFor(ToKey(key.AsRawKey()))
 	if rng == nil {
 		panic(fmt.Sprintf("programming error: range for key %s doesn't exist", key))
 	}
-	return *rng.config, nil
+	return *rng.config, roachpb.Span{
+		Key:    rng.startKey.ToRKey().AsRawKey(),
+		EndKey: rng.endKey.ToRKey().AsRawKey(),
+	}, nil
 }
 
 // Scan is added for the rangedesc.Scanner interface, required for

pkg/kv/kvserver/batcheval/cmd_export.go

@@ -108,7 +108,12 @@ func evalExport(
 	// ExportRequest is likely to find its target data has been GC'ed at this
 	// point, and so if the range being exported is part of such a table, we do
 	// not want to send back any row data to be backed up.
-	if cArgs.EvalCtx.ExcludeDataFromBackup(ctx) {
+	excludeFromBackup, err := cArgs.EvalCtx.ExcludeDataFromBackup(ctx,
+		roachpb.Span{Key: args.Key, EndKey: args.EndKey})
+	if err != nil {
+		return result.Result{}, err
+	}
+	if excludeFromBackup {
 		log.Infof(ctx, "[%s, %s) is part of a table excluded from backup, returning empty ExportResponse", args.Key, args.EndKey)
 		return result.Result{}, nil
 	}

pkg/kv/kvserver/batcheval/eval_context.go

@@ -103,7 +103,7 @@ type EvalContext interface {
 	GetMaxSplitCPU(context.Context) (float64, bool)
 
 	GetGCThreshold() hlc.Timestamp
-	ExcludeDataFromBackup(ctx context.Context) bool
+	ExcludeDataFromBackup(context.Context, roachpb.Span) (bool, error)
 	GetLastReplicaGCTimestamp(context.Context) (hlc.Timestamp, error)
 	GetLease() (roachpb.Lease, roachpb.Lease)
 	GetRangeInfo(context.Context) roachpb.RangeInfo
@@ -276,8 +276,8 @@ func (m *mockEvalCtxImpl) MinTxnCommitTS(
 func (m *mockEvalCtxImpl) GetGCThreshold() hlc.Timestamp {
 	return m.GCThreshold
 }
-func (m *mockEvalCtxImpl) ExcludeDataFromBackup(context.Context) bool {
-	return false
+func (m *mockEvalCtxImpl) ExcludeDataFromBackup(context.Context, roachpb.Span) (bool, error) {
+	return false, nil
 }
 func (m *mockEvalCtxImpl) GetLastReplicaGCTimestamp(context.Context) (hlc.Timestamp, error) {
 	panic("unimplemented")

pkg/kv/kvserver/client_merge_test.go

@@ -4283,6 +4283,14 @@ func TestMergeQueue(t *testing.T) {
 	lhsStartKey := roachpb.RKey(tc.ScratchRange(t))
 	rhsStartKey := lhsStartKey.Next().Next()
 	rhsEndKey := rhsStartKey.Next().Next()
+	lhsSp := roachpb.Span{
+		Key:    lhsStartKey.AsRawKey(),
+		EndKey: rhsStartKey.AsRawKey(),
+	}
+	rhsSp := roachpb.Span{
+		Key:    rhsStartKey.AsRawKey(),
+		EndKey: rhsEndKey.AsRawKey(),
+	}
 
 	for _, k := range []roachpb.RKey{lhsStartKey, rhsStartKey, rhsEndKey} {
 		split(t, k.AsRawKey(), hlc.Timestamp{} /* expirationTime */)
@@ -4297,12 +4305,12 @@ func TestMergeQueue(t *testing.T) {
 		if l := lhs(); l == nil {
 			t.Fatal("left-hand side range not found")
 		} else {
-			l.SetSpanConfig(conf)
+			l.SetSpanConfig(conf, lhsSp)
 		}
 		if r := rhs(); r == nil {
 			t.Fatal("right-hand side range not found")
 		} else {
-			r.SetSpanConfig(conf)
+			r.SetSpanConfig(conf, rhsSp)
 		}
 	}
 
@@ -4344,7 +4352,7 @@ func TestMergeQueue(t *testing.T) {
 		reset(t)
 		conf := conf
 		conf.RangeMinBytes *= 2
-		lhs().SetSpanConfig(conf)
+		lhs().SetSpanConfig(conf, lhsSp)
 		verifyMergedSoon(t, store, lhsStartKey, rhsStartKey)
 	})
 

pkg/kv/kvserver/client_spanconfigs_test.go

@@ -181,7 +181,7 @@ func (m *mockSpanConfigSubscriber) ComputeSplitKey(
 
 func (m *mockSpanConfigSubscriber) GetSpanConfigForKey(
 	ctx context.Context, key roachpb.RKey,
-) (roachpb.SpanConfig, error) {
+) (roachpb.SpanConfig, roachpb.Span, error) {
 	return m.Store.GetSpanConfigForKey(ctx, key)
 }
 

pkg/kv/kvserver/lease_queue.go

@@ -97,7 +97,7 @@ func newLeaseQueue(store *Store, allocator allocatorimpl.Allocator) *leaseQueue
 func (lq *leaseQueue) shouldQueue(
 	ctx context.Context, now hlc.ClockTimestamp, repl *Replica, confReader spanconfig.StoreReader,
 ) (shouldQueue bool, priority float64) {
-	conf, err := confReader.GetSpanConfigForKey(ctx, repl.startKey)
+	conf, _, err := confReader.GetSpanConfigForKey(ctx, repl.startKey)
 	if err != nil {
 		return false, 0
 	}
@@ -115,7 +115,7 @@ func (lq *leaseQueue) process(
 	}
 	defer repl.allocatorToken.Release(ctx)
 
-	conf, err := confReader.GetSpanConfigForKey(ctx, repl.startKey)
+	conf, _, err := confReader.GetSpanConfigForKey(ctx, repl.startKey)
 	if err != nil {
 		return false, err
 	}

pkg/kv/kvserver/merge_queue_test.go

@@ -156,7 +156,7 @@ func TestMergeQueueShouldQueue(t *testing.T) {
 			repl.mu.state.Stats = &enginepb.MVCCStats{KeyBytes: tc.bytes}
 			zoneConfig := zonepb.DefaultZoneConfigRef()
 			zoneConfig.RangeMinBytes = proto.Int64(tc.minBytes)
-			repl.SetSpanConfig(zoneConfig.AsSpanConfig())
+			repl.SetSpanConfig(zoneConfig.AsSpanConfig(), roachpb.Span{Key: tc.startKey, EndKey: tc.endKey})
 			shouldQ, priority := mq.shouldQueue(ctx, hlc.ClockTimestamp{}, repl, config.NewSystemConfig(zoneConfig))
 			if tc.expShouldQ != shouldQ {
 				t.Errorf("incorrect shouldQ: expected %v but got %v", tc.expShouldQ, shouldQ)

pkg/kv/kvserver/mvcc_gc_queue_test.go

@@ -872,7 +872,7 @@ func testMVCCGCQueueProcessImpl(t *testing.T, useEfos bool) {
 		}
 		defer snap.Close()
 
-		conf, err := cfg.GetSpanConfigForKey(ctx, desc.StartKey)
+		conf, _, err := cfg.GetSpanConfigForKey(ctx, desc.StartKey)
 		if err != nil {
 			t.Fatalf("could not find zone config for range %s: %+v", tc.repl, err)
 		}
@@ -1468,7 +1468,7 @@ func TestMVCCGCQueueChunkRequests(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	conf, err := confReader.GetSpanConfigForKey(ctx, roachpb.RKey("key"))
+	conf, _, err := confReader.GetSpanConfigForKey(ctx, roachpb.RKey("key"))
 	if err != nil {
 		t.Fatalf("could not find span config for range %s", err)
 	}
@@ -1522,7 +1522,11 @@ func TestMVCCGCQueueGroupsRangeDeletions(t *testing.T) {
 	require.NoError(t, store.AddReplica(r2))
 	r2.RaftStatus()
 	r2.handleGCHintResult(ctx, &roachpb.GCHint{LatestRangeDeleteTimestamp: hlc.Timestamp{WallTime: 1}})
-	r2.SetSpanConfig(roachpb.SpanConfig{GCPolicy: roachpb.GCPolicy{TTLSeconds: 100}})
+	r2.SetSpanConfig(roachpb.SpanConfig{GCPolicy: roachpb.GCPolicy{TTLSeconds: 100}},
+		roachpb.Span{
+			Key:    key("b").AsRawKey(),
+			EndKey: key("c").AsRawKey(),
+		})
 
 	gcQueue := newMVCCGCQueue(store)