spanconfig: introduce the spanconfig.SQLWatcher

This patch introduces the SQLWatcher, which is intended to incrementally
watch for updates to system.zones and system.descriptors. It does so by
establishing rangefeeds at a given timestamp.

The SQLWatcher periodically invokes a callback with a list of updates
that have been observed in the window
(previous checkpointTS, checkpointTS]. The checkpointTS is also
provided to the callback.

Internally, the SQLWatcher uses a buffer to keep track of events
generated by the SQLWatcher's rangefeeds. It also tracks the individual
frontier timestamps of both the rangefeeds. This helps to maintain the
notion of the combined frontier timestamp, which is computed as the
minimum of the two. This combined frontier timestamp serves as the
checkpoint to the SQLWatcher's callback function.

This interface isn't hooked up to anything yet. It'll be used by the
sponconfig.Reconciler soon to perform partial reconciliation once
full reconciliation is done. It is intended that the IDs from the
updates produced by the SQLWatcher will be fed into the SQLTranslator.

References #67679
Carved from #69661

Release note: None

pkg/BUILD.bazel

@@ -183,6 +183,7 @@ ALL_TESTS = [
     "//pkg/spanconfig/spanconfigkvaccessor:spanconfigkvaccessor_test",
     "//pkg/spanconfig/spanconfigmanager:spanconfigmanager_test",
     "//pkg/spanconfig/spanconfigsqltranslator:spanconfigsqltranslator_test",
+    "//pkg/spanconfig/spanconfigsqlwatcher:spanconfigsqlwatcher_test",
     "//pkg/spanconfig/spanconfigstore:spanconfigstore_test",
     "//pkg/spanconfig/spanconfigtestutils:spanconfigtestutils_test",
     "//pkg/sql/catalog/catalogkeys:catalogkeys_test",

pkg/ccl/changefeedccl/helpers_tenant_shim_test.go

@@ -71,6 +71,7 @@ func (t *testServerShim) DistSenderI() interface{}              { panic(unsuppor
 func (t *testServerShim) MigrationServer() interface{}          { panic(unsupportedShimMethod) }
 func (t *testServerShim) SpanConfigAccessor() interface{}       { panic(unsupportedShimMethod) }
 func (t *testServerShim) SpanConfigSQLTranslator() interface{}  { panic(unsupportedShimMethod) }
+func (t *testServerShim) SpanConfigSQLWatcher() interface{}     { panic(unsupportedShimMethod) }
 func (t *testServerShim) SQLServer() interface{}                { panic(unsupportedShimMethod) }
 func (t *testServerShim) SQLLivenessProvider() interface{}      { panic(unsupportedShimMethod) }
 func (t *testServerShim) StartupMigrationsManager() interface{} { panic(unsupportedShimMethod) }

pkg/kv/kvclient/rangefeed/rangefeedbuffer/buffer.go

@@ -50,7 +50,7 @@ func New(limit int) *Buffer {
 }
 
 // Add adds the given entry to the buffer.
-func (b *Buffer) Add(ctx context.Context, ev Event) error {
+func (b *Buffer) Add(ev Event) error {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 

pkg/kv/kvclient/rangefeed/rangefeedbuffer/buffer_test.go

@@ -48,11 +48,11 @@ func TestBuffer(t *testing.T) {
 	}
 
 	{ // Flushing at a timestamp lower than buffered events should return nothing.
-		require.NoError(t, buffer.Add(ctx, makeEvent("a", ts(13)))) // a@13
-		require.NoError(t, buffer.Add(ctx, makeEvent("b", ts(11)))) // b@11
-		require.NoError(t, buffer.Add(ctx, makeEvent("c", ts(15)))) // c@15
-		require.NoError(t, buffer.Add(ctx, makeEvent("d", ts(12)))) // d@12
-		require.NoError(t, buffer.Add(ctx, makeEvent("e", ts(18)))) // e@18
+		require.NoError(t, buffer.Add(makeEvent("a", ts(13)))) // a@13
+		require.NoError(t, buffer.Add(makeEvent("b", ts(11)))) // b@11
+		require.NoError(t, buffer.Add(makeEvent("c", ts(15)))) // c@15
+		require.NoError(t, buffer.Add(makeEvent("d", ts(12)))) // d@12
+		require.NoError(t, buffer.Add(makeEvent("e", ts(18)))) // e@18
 
 		events := buffer.Flush(ctx, ts(10))
 		require.True(t, len(events) == 0)
@@ -75,18 +75,18 @@ func TestBuffer(t *testing.T) {
 	}
 
 	{ // Adding events with timestamps <= the last flush are discarded.
-		require.NoError(t, buffer.Add(ctx, makeEvent("a", ts(13)))) // a@13
-		require.NoError(t, buffer.Add(ctx, makeEvent("b", ts(11)))) // b@11
-		require.NoError(t, buffer.Add(ctx, makeEvent("c", ts(15)))) // c@15
-		require.NoError(t, buffer.Add(ctx, makeEvent("d", ts(12)))) // d@12
+		require.NoError(t, buffer.Add(makeEvent("a", ts(13)))) // a@13
+		require.NoError(t, buffer.Add(makeEvent("b", ts(11)))) // b@11
+		require.NoError(t, buffer.Add(makeEvent("c", ts(15)))) // c@15
+		require.NoError(t, buffer.Add(makeEvent("d", ts(12)))) // d@12
 
 		events := buffer.Flush(ctx, ts(15))
 		require.True(t, len(events) == 0)
 	}
 
 	{ // Additional events are flushed out at appropriate points.
-		require.NoError(t, buffer.Add(ctx, makeEvent("f", ts(19)))) // f@19
-		require.NoError(t, buffer.Add(ctx, makeEvent("g", ts(21)))) // g@21
+		require.NoError(t, buffer.Add(makeEvent("f", ts(19)))) // f@19
+		require.NoError(t, buffer.Add(makeEvent("g", ts(21)))) // g@21
 
 		events := buffer.Flush(ctx, ts(20))
 		require.True(t, len(events) == 2)
@@ -107,10 +107,10 @@ func TestBuffer(t *testing.T) {
 
 	{ // Ensure that buffer limits are respected.
 		for i := 0; i < limit; i++ {
-			require.NoError(t, buffer.Add(ctx, makeEvent("x", ts(101)))) // x@101
+			require.NoError(t, buffer.Add(makeEvent("x", ts(101)))) // x@101
 		}
 
-		err := buffer.Add(ctx, makeEvent("x", ts(101)))
+		err := buffer.Add(makeEvent("x", ts(101)))
 		require.ErrorIs(t, err, rangefeedbuffer.ErrBufferLimitExceeded)
 	}
 }

pkg/server/BUILD.bazel

@@ -115,6 +115,7 @@ go_library(
         "//pkg/spanconfig/spanconfigkvaccessor",
         "//pkg/spanconfig/spanconfigmanager",
         "//pkg/spanconfig/spanconfigsqltranslator",
+        "//pkg/spanconfig/spanconfigsqlwatcher",
         "//pkg/sql",
         "//pkg/sql/catalog/bootstrap",
         "//pkg/sql/catalog/catalogkeys",

pkg/server/server_sql.go

@@ -53,6 +53,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/spanconfig"
 	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigmanager"
 	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigsqltranslator"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigsqlwatcher"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
@@ -841,14 +842,25 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		// Instantiate a span config manager. If we're the host tenant we'll
 		// only do it if COCKROACH_EXPERIMENTAL_SPAN_CONFIGS is set.
 		spanConfigKnobs, _ := cfg.TestingKnobs.SpanConfig.(*spanconfig.TestingKnobs)
+		if spanConfigKnobs == nil {
+			spanConfigKnobs = &spanconfig.TestingKnobs{}
+		}
 		sqlTranslator := spanconfigsqltranslator.New(execCfg, codec)
+		sqlWatcher := spanconfigsqlwatcher.New(
+			codec,
+			cfg.Settings,
+			cfg.rangeFeedFactory,
+			cfg.stopper,
+			spanConfigKnobs,
+		)
 		spanConfigMgr = spanconfigmanager.New(
 			cfg.db,
 			jobRegistry,
 			cfg.circularInternalExecutor,
 			cfg.stopper,
 			cfg.Settings,
 			cfg.spanConfigAccessor,
+			sqlWatcher,
 			sqlTranslator,
 			spanConfigKnobs,
 		)

pkg/server/testserver.go

@@ -961,6 +961,16 @@ func (ts *TestServer) SpanConfigSQLTranslator() interface{} {
 	return ts.sqlServer.spanconfigMgr.SQLTranslator
 }
 
+// SpanConfigSQLWatcher is part of TestServerInterface.
+func (ts *TestServer) SpanConfigSQLWatcher() interface{} {
+	if ts.sqlServer.spanconfigMgr == nil {
+		panic(
+			"span config manager uninitialized; see EnableSpanConfigs testing knob to use span configs",
+		)
+	}
+	return ts.sqlServer.spanconfigMgr.SQLWatcher
+}
+
 // SQLServer is part of TestServerInterface.
 func (ts *TestServer) SQLServer() interface{} {
 	return ts.PGServer().SQLServer

pkg/spanconfig/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "//pkg/base",
         "//pkg/keys",
         "//pkg/roachpb:with-mocks",
+        "//pkg/sql/catalog",
         "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
     ],

pkg/spanconfig/spanconfig.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 )
@@ -77,6 +78,47 @@ func FullTranslate(
 	return s.Translate(ctx, descpb.IDs{keys.RootNamespaceID})
 }
 
+// SQLWatcher watches for events on system.zones and system.descriptors.
+type SQLWatcher interface {
+	// WatchForSQLUpdates watches for changes to zones and descriptors starting at
+	// the given timestamp (exclusive) by establishing rangefeeds over
+	// system.zones and system.descriptors.
+	//
+	// It periodically calls the handle function with a list of SQLWatcherUpdates
+	// and a checkpointTS. Invocations of handle are guaranteed to provide the
+	// following semantics:
+	// 1. Calls to handle are serial.
+	// 2. The timestamp supplied to handle is monotonically increasing.
+	// 3. The list of SQLWatcherUpdates supplied to handle includes all events
+	// in the window (prevInvocationCheckpointTS, checkpointTS].
+	// 4. No further calls to handle are made if an invocation returns an error.
+	//
+	// These guarantees allow users of this interface to safely persist the
+	// checkpointTS and use it as the startTS for WatchForSQLUpdates later without
+	// missing any updates.
+	//
+	// All rangefeeds established by the SQLWatcher will be torn down before the
+	// call to WatchForSQLUpdates returns.
+	WatchForSQLUpdates(ctx context.Context, startTS hlc.Timestamp, handle SQLWatcherHandleFunc) error
+}
+
+// SQLWatcherHandleFunc is the type of the handler function periodically invoked
+// by the SQLWatcher.
+type SQLWatcherHandleFunc func(
+	ctx context.Context, updates []SQLWatcherUpdate, checkpointTS hlc.Timestamp,
+) error
+
+// SQLWatcherUpdate captures the ID and type of a descriptor or zone that the
+// SQLWatcher has observed change.
+type SQLWatcherUpdate struct {
+	// ID of the descriptor/zone that has changed.
+	ID descpb.ID
+
+	// DescriptorType of the descriptor/zone that has changed. Could be either the
+	// specific type or catalog.Any.
+	DescriptorType catalog.DescriptorType
+}
+
 // ReconciliationDependencies captures what's needed by the span config
 // reconciliation job to perform its task. The job is responsible for
 // reconciling a tenant's zone configurations with the clusters span
@@ -86,11 +128,7 @@ type ReconciliationDependencies interface {
 
 	SQLTranslator
 
-	// TODO(arul): We'll also want access to a "SQLWatcher", something that
-	// watches for changes to system.{descriptors, zones} to feed IDs to the
-	// SQLTranslator. These interfaces will be used by the "Reconciler to perform
-	// full/partial reconciliation, checkpoint the span config job, and update KV
-	// with the tenants span config state.
+	SQLWatcher
 }
 
 // Store is a data structure used to store spans and their corresponding

pkg/spanconfig/spanconfigmanager/manager.go

@@ -60,6 +60,7 @@ type Manager struct {
 	knobs    *spanconfig.TestingKnobs
 
 	spanconfig.KVAccessor
+	spanconfig.SQLWatcher
 	spanconfig.SQLTranslator
 }
 
@@ -73,19 +74,18 @@ func New(
 	stopper *stop.Stopper,
 	settings *cluster.Settings,
 	kvAccessor spanconfig.KVAccessor,
+	sqlWatcher spanconfig.SQLWatcher,
 	sqlTranslator spanconfig.SQLTranslator,
 	knobs *spanconfig.TestingKnobs,
 ) *Manager {
-	if knobs == nil {
-		knobs = &spanconfig.TestingKnobs{}
-	}
 	return &Manager{
 		db:            db,
 		jr:            jr,
 		ie:            ie,
 		stopper:       stopper,
 		settings:      settings,
 		KVAccessor:    kvAccessor,
+		SQLWatcher:    sqlWatcher,
 		SQLTranslator: sqlTranslator,
 		knobs:         knobs,
 	}

pkg/spanconfig/spanconfigmanager/manager_test.go

@@ -74,6 +74,7 @@ func TestManagerConcurrentJobCreation(t *testing.T) {
 		ts.Stopper(),
 		ts.ClusterSettings(),
 		ts.SpanConfigAccessor().(spanconfig.KVAccessor),
+		ts.SpanConfigSQLWatcher().(spanconfig.SQLWatcher),
 		ts.SpanConfigSQLTranslator().(spanconfig.SQLTranslator),
 		&spanconfig.TestingKnobs{
 			ManagerCreatedJobInterceptor: func(jobI interface{}) {
@@ -162,6 +163,7 @@ func TestManagerStartsJobIfFailed(t *testing.T) {
 		ts.Stopper(),
 		ts.ClusterSettings(),
 		ts.SpanConfigAccessor().(spanconfig.KVAccessor),
+		ts.SpanConfigSQLWatcher().(spanconfig.SQLWatcher),
 		ts.SpanConfigSQLTranslator().(spanconfig.SQLTranslator),
 		&spanconfig.TestingKnobs{
 			ManagerAfterCheckedReconciliationJobExistsInterceptor: func(exists bool) {
@@ -237,6 +239,7 @@ func TestManagerCheckJobConditions(t *testing.T) {
 		ts.Stopper(),
 		ts.ClusterSettings(),
 		ts.SpanConfigAccessor().(spanconfig.KVAccessor),
+		ts.SpanConfigSQLWatcher().(spanconfig.SQLWatcher),
 		ts.SpanConfigSQLTranslator().(spanconfig.SQLTranslator),
 		&spanconfig.TestingKnobs{
 			ManagerDisableJobCreation: true,

pkg/spanconfig/spanconfigsqlwatcher/BUILD.bazel

@@ -0,0 +1,67 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "spanconfigsqlwatcher",
+    srcs = [
+        "buffer.go",
+        "sqlwatcher.go",
+        "zonesdecoder.go",
+    ],
+    importpath = "github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigsqlwatcher",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/keys",
+        "//pkg/kv/kvclient/rangefeed:with-mocks",
+        "//pkg/kv/kvclient/rangefeed/rangefeedbuffer",
+        "//pkg/roachpb:with-mocks",
+        "//pkg/settings/cluster",
+        "//pkg/spanconfig",
+        "//pkg/sql/catalog",
+        "//pkg/sql/catalog/descpb",
+        "//pkg/sql/catalog/systemschema",
+        "//pkg/sql/rowenc",
+        "//pkg/sql/sem/tree",
+        "//pkg/sql/types",
+        "//pkg/util/hlc",
+        "//pkg/util/log",
+        "//pkg/util/log/logcrash",
+        "//pkg/util/stop",
+        "//pkg/util/syncutil",
+        "@com_github_cockroachdb_errors//:errors",
+    ],
+)
+
+go_test(
+    name = "spanconfigsqlwatcher_test",
+    srcs = [
+        "buffer_test.go",
+        "main_test.go",
+        "sqlwatcher_test.go",
+        "zonesdecoder_test.go",
+    ],
+    embed = [":spanconfigsqlwatcher"],
+    deps = [
+        "//pkg/base",
+        "//pkg/config/zonepb",
+        "//pkg/jobs",
+        "//pkg/keys",
+        "//pkg/kv/kvclient/rangefeed:with-mocks",
+        "//pkg/security",
+        "//pkg/security/securitytest",
+        "//pkg/server",
+        "//pkg/spanconfig",
+        "//pkg/sql/catalog",
+        "//pkg/sql/catalog/descpb",
+        "//pkg/testutils",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/hlc",
+        "//pkg/util/leaktest",
+        "//pkg/util/protoutil",
+        "//pkg/util/syncutil",
+        "@com_github_cockroachdb_errors//:errors",
+        "@com_github_gogo_protobuf//proto",
+        "@com_github_stretchr_testify//require",
+    ],
+)