changefeedccl: delete per-table resolved timestamps code

Now that we periodically persist the entire span frontier, there is
no longer a need for per-table resolved timestamps and thus all the
relevant code can be deleted.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/cdcprogresspb/BUILD.bazel

@@ -7,11 +7,7 @@ proto_library(
     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",
-    ],
+    deps = ["@com_github_gogo_protobuf//gogoproto:gogo_proto"],
 )
 
 go_proto_library(
@@ -20,11 +16,7 @@ go_proto_library(
     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",
-    ],
+    deps = ["@com_github_gogo_protobuf//gogoproto"],
 )
 
 go_library(

pkg/ccl/changefeedccl/cdcprogresspb/progress.proto

@@ -8,19 +8,6 @@ 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";
-import "util/hlc/timestamp.proto";
-
-// ResolvedTables contains per-table resolved timestamp information.
-message ResolvedTables {
-  map<uint32, util.hlc.Timestamp> tables = 1 [
-    (gogoproto.castkey) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.ID",
-    (gogoproto.nullable) = false
-  ];
-
-  // TODO(#148124): Write to this field for aggregator-to-frontier messages.
-  cockroach.sql.jobs.jobspb.ResolvedSpan.BoundaryType boundary_type = 2;
-}
 
 // ProtectedTimestampRecords is a map from table descriptor IDs to protected timestamp record IDs.
 message ProtectedTimestampRecords {

pkg/ccl/changefeedccl/changefeed_job_info.go

@@ -23,8 +23,6 @@ const (
 )
 
 const (
-	// resolvedTablesFilename: ~changefeed/resolved-tables.binpb
-	resolvedTablesFilename = jobInfoFilenamePrefix + "resolved-tables" + jobInfoFilenameExtension
 	// perTableProtectedTimestampsFilename: ~changefeed/per-table-protected-timestamps.binpb
 	perTableProtectedTimestampsFilename = jobInfoFilenamePrefix + "per-table-protected-timestamps" + jobInfoFilenameExtension
 )
@@ -49,7 +47,6 @@ func writeChangefeedJobInfo(
 
 // readChangefeedJobInfo reads a changefeed job info protobuf from the
 // job_info table. A changefeed-specific filename is required.
-// TODO(#148119): Use this function to read.
 func readChangefeedJobInfo(
 	ctx context.Context, filename string, info protoutil.Message, txn isql.Txn, jobID jobspb.JobID,
 ) error {

pkg/ccl/changefeedccl/changefeed_job_info_test.go

@@ -9,55 +9,53 @@ import (
 	"context"
 	"testing"
 
+	"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/jobs"
-	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
-	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/stretchr/testify/require"
 )
 
-func TestChangefeedJobInfoResolvedTables(t *testing.T) {
+func TestChangefeedJobInfoRoundTrip(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
-		ctx := context.Background()
-		sqlDB := sqlutils.MakeSQLRunner(s.DB)
-
-		// Make sure per-table tracking is enabled.
-		changefeedbase.TrackPerTableProgress.Override(ctx, &s.Server.ClusterSettings().SV, true)
-
-		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
-		sqlDB.Exec(t, `CREATE TABLE bar (x INT PRIMARY KEY, y STRING)`)
-		feed := feed(t, f, `CREATE CHANGEFEED FOR foo, bar`)
-		defer closeFeed(t, feed)
-
-		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'one')`)
-		sqlDB.Exec(t, `INSERT INTO bar VALUES (10, 'ten')`)
-		assertPayloads(t, feed, []string{
-			`foo: [1]->{"after": {"a": 1, "b": "one"}}`,
-			`bar: [10]->{"after": {"x": 10, "y": "ten"}}`,
-		})
-
-		// The ResolvedTables message should be persisted to the job_info table
-		// at the same time as the highwater being set.
-		enterpriseFeed := feed.(cdctest.EnterpriseTestFeed)
-		waitForHighwater(t, enterpriseFeed, s.Server.JobRegistry().(*jobs.Registry))
-
-		// Make sure the ResolvedTables message was persisted and can be decoded.
-		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())
-		})
-		require.NoError(t, err)
-		require.Len(t, resolvedTables.Tables, 2)
+	ctx := context.Background()
+	srv := serverutils.StartServerOnly(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+
+	jobID := jobspb.JobID(123456)
+
+	// Create a basic progress record.
+	uuid1 := uuid.MakeV4()
+	uuid2 := uuid.MakeV4()
+	ptsRecords := cdcprogresspb.ProtectedTimestampRecords{
+		ProtectedTimestampRecords: map[descpb.ID]*uuid.UUID{
+			descpb.ID(100): &uuid1,
+			descpb.ID(200): &uuid2,
+		},
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	// Write the progress record.
+	err := srv.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		return writeChangefeedJobInfo(ctx,
+			perTableProtectedTimestampsFilename, &ptsRecords, txn, jobID)
+	})
+	require.NoError(t, err)
+
+	// Read the record back.
+	var readPTSRecords cdcprogresspb.ProtectedTimestampRecords
+	err = srv.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		return readChangefeedJobInfo(ctx,
+			perTableProtectedTimestampsFilename, &readPTSRecords, txn, jobID)
+	})
+	require.NoError(t, err)
+
+	// Verify the read data matches the written data.
+	require.Equal(t, ptsRecords, readPTSRecords)
 }

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1877,20 +1877,6 @@ func (cf *changeFrontier) checkpointJobProgress(
 				progress.StatusMessage = fmt.Sprintf("running: resolved=%s", frontier)
 			}
 
-			// Write per-table progress if enabled.
-			if cf.spec.ProgressConfig != nil && cf.spec.ProgressConfig.PerTableTracking {
-				resolvedTables := &cdcprogresspb.ResolvedTables{
-					Tables: make(map[descpb.ID]hlc.Timestamp),
-				}
-				for tableID, tableFrontier := range cf.frontier.Frontiers() {
-					resolvedTables.Tables[tableID] = tableFrontier.Frontier()
-				}
-
-				if err := writeChangefeedJobInfo(ctx, resolvedTablesFilename, resolvedTables, txn, cf.spec.JobID); err != nil {
-					return errors.Wrap(err, "error writing resolved tables to job info")
-				}
-			}
-
 			ju.UpdateProgress(progress)
 
 			return nil