Merge #152883

152883: cdcprogresspb: move ResolvedTables proto into new package r=aerfrei,asg0451 a=andyyang890

This commit moves the `ResolvedTables` protobuf from `changefeedpb` to a new
`cdcprogresspb` package so that it can be added to the processor specs.
This was previously not possible because `pkg/sql/execinfrapb` (where the
processor specs live) disallows imports of `pkg/sql/parser`, which is a
dependency of a different protobuf in `changefeedpb`.

The `ProtectedTimestampRecords` protobuf has also been moved for consistency.

Informs #148119

Release note: None

Co-authored-by: Andy Yang <[email protected]>

Author: craig[bot]

pkg/BUILD.bazel

@@ -931,6 +931,7 @@ GO_TARGETS = [
     "//pkg/ccl/changefeedccl/cdceval:cdceval_test",
     "//pkg/ccl/changefeedccl/cdcevent:cdcevent",
     "//pkg/ccl/changefeedccl/cdcevent:cdcevent_test",
+    "//pkg/ccl/changefeedccl/cdcprogresspb:cdcprogresspb",
     "//pkg/ccl/changefeedccl/cdctest:cdctest",
     "//pkg/ccl/changefeedccl/cdctest:cdctest_test",
     "//pkg/ccl/changefeedccl/cdcutils:cdcutils",

pkg/ccl/changefeedccl/BUILD.bazel

@@ -52,6 +52,7 @@ go_library(
         "//pkg/ccl/changefeedccl/avro",
         "//pkg/ccl/changefeedccl/cdceval",
         "//pkg/ccl/changefeedccl/cdcevent",
+        "//pkg/ccl/changefeedccl/cdcprogresspb",
         "//pkg/ccl/changefeedccl/cdcutils",
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/changefeedccl/changefeedpb",
@@ -238,6 +239,7 @@ go_test(
         "//pkg/ccl",
         "//pkg/ccl/changefeedccl/cdceval",
         "//pkg/ccl/changefeedccl/cdcevent",
+        "//pkg/ccl/changefeedccl/cdcprogresspb",
         "//pkg/ccl/changefeedccl/cdctest",
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/changefeedccl/changefeedpb",

pkg/ccl/changefeedccl/cdcprogresspb/BUILD.bazel

@@ -0,0 +1,40 @@
+load("@rules_proto//proto:defs.bzl", "proto_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
+
+proto_library(
+    name = "cdcprogresspb_proto",
+    srcs = ["progress.proto"],
+    strip_import_prefix = "/pkg",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/jobs/jobspb:jobspb_proto",
+        "//pkg/util/hlc:hlc_proto",
+        "@com_github_gogo_protobuf//gogoproto:gogo_proto",
+    ],
+)
+
+go_proto_library(
+    name = "cdcprogresspb_go_proto",
+    compilers = ["//pkg/cmd/protoc-gen-gogoroach:protoc-gen-gogoroach_compiler"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb",
+    proto = ":cdcprogresspb_proto",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/jobs/jobspb",
+        "//pkg/util/hlc",
+        "@com_github_gogo_protobuf//gogoproto",
+    ],
+)
+
+go_library(
+    name = "cdcprogresspb",
+    srcs = ["progress.go"],
+    embed = [":cdcprogresspb_go_proto"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/sql/catalog/descpb",
+        "//pkg/util/uuid",
+    ],
+)

pkg/ccl/changefeedccl/cdcprogresspb/progress.go

@@ -0,0 +1,11 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package cdcprogresspb
+
+import (
+	_ "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb" // Needed for progress.proto.
+	_ "github.com/cockroachdb/cockroach/pkg/util/uuid"          // Needed for progress.proto.
+)

pkg/ccl/changefeedccl/changefeedpb/changefeed.proto renamed to pkg/ccl/changefeedccl/cdcprogresspb/progress.proto

@@ -4,8 +4,8 @@
 // included in the /LICENSE file.
 
 syntax = "proto3";
-package cockroach.ccl.changefeedccl;
-option go_package = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb";
+package cockroach.ccl.changefeedccl.cdcprogresspb;
+option go_package = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb";
 
 import "gogoproto/gogo.proto";
 import "jobs/jobspb/jobs.proto";

pkg/ccl/changefeedccl/changefeed_job_info_test.go

@@ -9,9 +9,9 @@ import (
 	"context"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
@@ -50,7 +50,7 @@ func TestChangefeedJobInfoResolvedTables(t *testing.T) {
 		waitForHighwater(t, enterpriseFeed, s.Server.JobRegistry().(*jobs.Registry))
 
 		// Make sure the ResolvedTables message was persisted and can be decoded.
-		var resolvedTables changefeedpb.ResolvedTables
+		var resolvedTables cdcprogresspb.ResolvedTables
 		execCfg := s.Server.ExecutorConfig().(sql.ExecutorConfig)
 		err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 			return readChangefeedJobInfo(ctx, resolvedTablesFilename, &resolvedTables, txn, enterpriseFeed.JobID())

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -14,9 +14,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcutils"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/checkpoint"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvevent"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvfeed"
@@ -1868,7 +1868,7 @@ func (cf *changeFrontier) checkpointJobProgress(
 
 			// Write per-table progress if enabled.
 			if cf.spec.ProgressConfig != nil && cf.spec.ProgressConfig.PerTableTracking {
-				resolvedTables := &changefeedpb.ResolvedTables{
+				resolvedTables := &cdcprogresspb.ResolvedTables{
 					Tables: make(map[descpb.ID]hlc.Timestamp),
 				}
 				for tableID, tableFrontier := range cf.frontier.Frontiers() {
@@ -1930,7 +1930,7 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 		}
 	}()
 
-	var ptsEntries changefeedpb.ProtectedTimestampRecords
+	var ptsEntries cdcprogresspb.ProtectedTimestampRecords
 	if err := readChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, cf.spec.JobID); err != nil {
 		return false, err
 	}
@@ -1961,7 +1961,7 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 func (cf *changeFrontier) managePerTableProtectedTimestamps(
 	ctx context.Context,
 	txn isql.Txn,
-	ptsEntries *changefeedpb.ProtectedTimestampRecords,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	highwater hlc.Timestamp,
 ) (newPTS hlc.Timestamp, updatedPerTablePTS bool, err error) {
 	var leastLaggingTimestamp hlc.Timestamp
@@ -2036,7 +2036,7 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 func (cf *changeFrontier) releasePerTableProtectedTimestampRecords(
 	ctx context.Context,
 	txn isql.Txn,
-	ptsEntries *changefeedpb.ProtectedTimestampRecords,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	tableIDs []descpb.ID,
 	pts protectedts.Storage,
 ) error {
@@ -2051,7 +2051,7 @@ func (cf *changeFrontier) releasePerTableProtectedTimestampRecords(
 
 func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 	ctx context.Context,
-	ptsEntries *changefeedpb.ProtectedTimestampRecords,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	tableID descpb.ID,
 	tableHighWater hlc.Timestamp,
 	pts protectedts.Storage,
@@ -2075,7 +2075,7 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 func (cf *changeFrontier) createPerTableProtectedTimestampRecord(
 	ctx context.Context,
 	txn isql.Txn,
-	ptsEntries *changefeedpb.ProtectedTimestampRecords,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	tableID descpb.ID,
 	tableHighWater hlc.Timestamp,
 	pts protectedts.Storage,

pkg/ccl/changefeedccl/changefeedpb/BUILD.bazel

@@ -4,17 +4,9 @@ load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
 
 proto_library(
     name = "changefeedpb_proto",
-    srcs = [
-        "changefeed.proto",
-        "scheduled_changefeed.proto",
-    ],
+    srcs = ["scheduled_changefeed.proto"],
     strip_import_prefix = "/pkg",
     visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/jobs/jobspb:jobspb_proto",
-        "//pkg/util/hlc:hlc_proto",
-        "@com_github_gogo_protobuf//gogoproto:gogo_proto",
-    ],
 )
 
 go_proto_library(
@@ -23,26 +15,16 @@ go_proto_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb",
     proto = ":changefeedpb_proto",
     visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/jobs/jobspb",
-        "//pkg/util/hlc",
-        "//pkg/util/uuid",  # keep
-        "@com_github_gogo_protobuf//gogoproto",
-    ],
 )
 
 go_library(
     name = "changefeedpb",
-    srcs = [
-        "changefeed.go",
-        "marshal.go",
-    ],
+    srcs = ["marshal.go"],
     embed = [":changefeedpb_go_proto"],
     importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/cloud",
-        "//pkg/sql/catalog/descpb",
         "//pkg/sql/parser",
         "//pkg/sql/sem/tree",
         "@com_github_cockroachdb_errors//:errors",

pkg/ccl/changefeedccl/changefeedpb/changefeed.go

@@ -14,9 +14,9 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -1033,7 +1033,7 @@ func TestChangefeedProtectedTimestampUpdateForMultipleTables(t *testing.T) {
 
 		execCfg := s.Server.ExecutorConfig().(sql.ExecutorConfig)
 		err = execCfg.InternalDB.Txn(context.Background(), func(ctx context.Context, txn isql.Txn) error {
-			var ptsEntries changefeedpb.ProtectedTimestampRecords
+			var ptsEntries cdcprogresspb.ProtectedTimestampRecords
 			if err := readChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, eFeed.JobID()); err != nil {
 				return err
 			}
@@ -1131,7 +1131,7 @@ func TestChangefeedPerTableProtectedTimestampProgression(t *testing.T) {
 		assertTablePTSRecords := func(expectedTables map[descpb.ID]struct{}) {
 			testutils.SucceedsSoon(t, func() error {
 				return execCfg.InternalDB.Txn(context.Background(), func(ctx context.Context, txn isql.Txn) error {
-					var ptsEntries changefeedpb.ProtectedTimestampRecords
+					var ptsEntries cdcprogresspb.ProtectedTimestampRecords
 					if err := readChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, eFeed.JobID()); err != nil {
 						return err
 					}