changefeedccl: database-level changefeeds: tableset watcher

This is a new package that implements a watcher
for database-level changefeeds. It is used to
watch for changes to the set of watched tables, so
that the feed coordinator may react to them.

Fixes: #148836
Fixes: #148829

Release note: None

Author: asg0451

pkg/BUILD.bazel

@@ -38,6 +38,7 @@ ALL_TESTS = [
     "//pkg/ccl/changefeedccl/kvfeed:kvfeed_test",
     "//pkg/ccl/changefeedccl/resolvedspan:resolvedspan_test",
     "//pkg/ccl/changefeedccl/schemafeed:schemafeed_test",
+    "//pkg/ccl/changefeedccl/tableset:tableset_test",
     "//pkg/ccl/changefeedccl:changefeedccl_test",
     "//pkg/ccl/cliccl:cliccl_test",
     "//pkg/ccl/cloudccl/amazon:amazon_test",
@@ -933,6 +934,8 @@ GO_TARGETS = [
     "//pkg/ccl/changefeedccl/schemafeed/schematestutils:schematestutils",
     "//pkg/ccl/changefeedccl/schemafeed:schemafeed",
     "//pkg/ccl/changefeedccl/schemafeed:schemafeed_test",
+    "//pkg/ccl/changefeedccl/tableset:tableset",
+    "//pkg/ccl/changefeedccl/tableset:tableset_test",
     "//pkg/ccl/changefeedccl/timers:timers",
     "//pkg/ccl/changefeedccl:changefeedccl",
     "//pkg/ccl/changefeedccl:changefeedccl_test",

pkg/ccl/changefeedccl/tableset/BUILD.bazel

@@ -0,0 +1,64 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "tableset",
+    srcs = ["watcher.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/tableset",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/ccl/changefeedccl/cdcevent",
+        "//pkg/ccl/changefeedccl/changefeedbase",
+        "//pkg/jobs/jobspb",
+        "//pkg/kv/kvclient/rangefeed",
+        "//pkg/kv/kvpb",
+        "//pkg/roachpb",
+        "//pkg/sql",
+        "//pkg/sql/catalog/catalogkeys",
+        "//pkg/sql/catalog/descpb",
+        "//pkg/sql/catalog/systemschema",
+        "//pkg/sql/sem/tree",
+        "//pkg/util/buildutil",
+        "//pkg/util/hlc",
+        "//pkg/util/log",
+        "//pkg/util/mon",
+        "//pkg/util/span",
+        "//pkg/util/syncutil",
+        "//pkg/util/timeutil",
+        "//pkg/util/tracing",
+        "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_logtags//:logtags",
+        "@org_golang_x_sync//errgroup",
+    ],
+)
+
+go_test(
+    name = "tableset_test",
+    srcs = [
+        "main_test.go",
+        "watcher_test.go",
+    ],
+    embed = [":tableset"],
+    deps = [
+        "//pkg/base",
+        "//pkg/security/securityassets",
+        "//pkg/security/securitytest",
+        "//pkg/server",
+        "//pkg/settings/cluster",
+        "//pkg/sql",
+        "//pkg/sql/catalog",
+        "//pkg/sql/catalog/descpb",
+        "//pkg/sql/catalog/descs",
+        "//pkg/testutils",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/hlc",
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/mon",
+        "//pkg/util/timeutil",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+        "@org_golang_x_sync//errgroup",
+    ],
+)

pkg/ccl/changefeedccl/tableset/main_test.go

@@ -0,0 +1,35 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package tableset
+
+import (
+	"os"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/security/securityassets"
+	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
+	"github.com/cockroachdb/cockroach/pkg/server"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+)
+
+func init() {
+	securityassets.SetLoader(securitytest.EmbeddedAssets)
+}
+
+func TestMain(m *testing.M) {
+	serverutils.InitTestServerFactory(server.TestServerFactory)
+	serverutils.InitTestClusterFactory(testcluster.TestClusterFactory)
+
+	defer serverutils.TestingSetDefaultTenantSelectionOverride(
+		base.TestIsForStuffThatShouldWorkWithSecondaryTenantsButDoesntYet(76378),
+	)()
+
+	os.Exit(m.Run())
+}
+
+//go:generate ../util/leaktest/add-leaktest.sh *_test.go