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"
@@ -842,13 +843,20 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		// only do it if COCKROACH_EXPERIMENTAL_SPAN_CONFIGS is set.
 		spanConfigKnobs, _ := cfg.TestingKnobs.SpanConfig.(*spanconfig.TestingKnobs)
 		sqlTranslator := spanconfigsqltranslator.New(execCfg, codec)
+		sqlWatcher := spanconfigsqlwatcher.New(
+			codec,
+			cfg.Settings,
+			cfg.rangeFeedFactory,
+			cfg.stopper,
+		)
 		spanConfigMgr = spanconfigmanager.New(
 			cfg.db,
 			jobRegistry,
 			cfg.circularInternalExecutor,
 			cfg.stopper,
 			cfg.Settings,
 			cfg.spanConfigAccessor,
+			sqlWatcher,
 			sqlTranslator,
 			spanConfigKnobs,
 		)

pkg/server/testserver.go

@@ -962,6 +962,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,46 @@ func FullTranslate(
 	return s.Translate(ctx, descpb.IDs{keys.RootNamespaceID})
 }
 
+// SQLWatcher can be used to watch 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 monotonically increasing checkpointTS. The list of SQLWatcherUpdates
+	// is guaranteed to include all events on system.zones and system.descriptors
+	// in the window (previous checkpointTS, checkpointTS]. All calls to handle
+	// are executed serially.
+	//
+	// The checkpointTS can be stored and used as the startTS to subsequent calls
+	// to WatchForSQLUpdates. The very first call to handle guarantees the
+	// invariant above with startTS == previous checkpointTS.
+	WatchForSQLUpdates(ctx context.Context, startTS hlc.Timestamp, handle SQLWatcherHandleFunc) error
+
+	// Close stops the rangefeeds created by the SQLWatcher and waits for them
+	// to shut down before returning. Close is idempotent.
+	Close()
+}
+
+// SQLWatcherHandleFunc is the type of the handler function expected 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 +127,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,6 +74,7 @@ func New(
 	stopper *stop.Stopper,
 	settings *cluster.Settings,
 	kvAccessor spanconfig.KVAccessor,
+	sqlWatcher spanconfig.SQLWatcher,
 	sqlTranslator spanconfig.SQLTranslator,
 	knobs *spanconfig.TestingKnobs,
 ) *Manager {
@@ -86,6 +88,7 @@ func New(
 		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,65 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "spanconfigsqlwatcher",
+    srcs = [
+        "buffer.go",
+        "sql_watcher.go",
+        "zones_decoder.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",
+        "sql_watcher_test.go",
+        "zones_decoder_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/testcluster",
+        "//pkg/util/hlc",
+        "//pkg/util/leaktest",
+        "//pkg/util/protoutil",
+        "//pkg/util/syncutil",
+        "@com_github_gogo_protobuf//proto",
+        "@com_github_stretchr_testify//require",
+    ],
+)