changefeedccl/resolvedspan: integrate span.MultiFrontier into frontiers

This patch integrates `span.MultiFrontier`s into the resolved span frontiers
to give changefeed processors easy access to per-table resolved timestamps.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/BUILD.bazel

@@ -196,7 +196,6 @@ go_test(
     srcs = [
         "alter_changefeed_test.go",
         "changefeed_dist_test.go",
-        "changefeed_processors_test.go",
         "changefeed_stmt_test.go",
         "changefeed_test.go",
         "csv_test.go",

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -615,7 +615,8 @@ func (ca *changeAggregator) setupSpansAndFrontier() (spans []roachpb.Span, err e
 	if err != nil {
 		return nil, err
 	}
-	ca.frontier, err = resolvedspan.NewAggregatorFrontier(ca.spec.Feed.StatementTime, initialHighWater, spans...)
+	ca.frontier, err = resolvedspan.NewAggregatorFrontier(
+		ca.spec.Feed.StatementTime, initialHighWater, ca.FlowCtx.Codec(), spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -1419,7 +1420,8 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 	}
 
 	// Set up the resolved span frontier.
-	cf.frontier, err = resolvedspan.NewCoordinatorFrontier(cf.spec.Feed.StatementTime, initialHighwater, cf.spec.TrackedSpans...)
+	cf.frontier, err = resolvedspan.NewCoordinatorFrontier(
+		cf.spec.Feed.StatementTime, initialHighwater, cf.FlowCtx.Codec(), cf.spec.TrackedSpans...)
 	if err != nil {
 		log.Infof(cf.Ctx(), "change frontier moving to draining due to error setting up frontier: %v", err)
 		cf.MoveToDraining(err)

pkg/ccl/changefeedccl/changefeed_processors_test.go

@@ -8936,7 +8936,12 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 
 		// Verify that the resumed job has restored the progress from the checkpoint
 		// to the change frontier.
-		expectedFrontier, err := span.MakeFrontier(tableSpan)
+		expectedFrontier, err := resolvedspan.NewCoordinatorFrontier(
+			hlc.Timestamp{},
+			hlc.Timestamp{},
+			s.Codec,
+			tableSpan,
+		)
 		if err != nil {
 			t.Fatal(err)
 		}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -14,6 +14,7 @@ go_library(
         "//pkg/jobs/jobspb",
         "//pkg/roachpb",
         "//pkg/settings",
+        "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
         "//pkg/util/span",
         "@com_github_cockroachdb_errors//:errors",
@@ -27,10 +28,14 @@ go_test(
     deps = [
         ":resolvedspan",
         "//pkg/jobs/jobspb",
+        "//pkg/keys",
         "//pkg/roachpb",
+        "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
+        "//pkg/util/randutil",
+        "//pkg/util/span",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],

pkg/ccl/changefeedccl/resolvedspan/BUILD.bazel

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/errors"
@@ -28,9 +29,12 @@ type AggregatorFrontier struct {
 
 // NewAggregatorFrontier returns a new AggregatorFrontier.
 func NewAggregatorFrontier(
-	statementTime hlc.Timestamp, initialHighWater hlc.Timestamp, spans ...roachpb.Span,
+	statementTime hlc.Timestamp,
+	initialHighWater hlc.Timestamp,
+	decoder TablePrefixDecoder,
+	spans ...roachpb.Span,
 ) (*AggregatorFrontier, error) {
-	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, spans...)
+	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -73,9 +77,12 @@ type CoordinatorFrontier struct {
 
 // NewCoordinatorFrontier returns a new CoordinatorFrontier.
 func NewCoordinatorFrontier(
-	statementTime hlc.Timestamp, initialHighWater hlc.Timestamp, spans ...roachpb.Span,
+	statementTime hlc.Timestamp,
+	initialHighWater hlc.Timestamp,
+	decoder TablePrefixDecoder,
+	spans ...roachpb.Span,
 ) (*CoordinatorFrontier, error) {
-	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, spans...)
+	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -184,15 +191,11 @@ func (f *CoordinatorFrontier) MakeCheckpoint(
 	return checkpoint.Make(f.Frontier(), f.Entries(), maxBytes, metrics)
 }
 
-// spanFrontier is a type alias to make it possible to embed and forward calls
-// (e.g. Frontier()) to the underlying span.Frontier.
-type spanFrontier = span.Frontier
-
 // resolvedSpanFrontier wraps a spanFrontier with additional bookkeeping fields
 // used to track resolved spans for a changefeed and methods for computing
 // lagging and checkpoint spans.
 type resolvedSpanFrontier struct {
-	spanFrontier
+	*span.MultiFrontier[descpb.ID]
 
 	// statementTime is the statement time of the changefeed.
 	statementTime hlc.Timestamp
@@ -211,15 +214,17 @@ type resolvedSpanFrontier struct {
 
 // newResolvedSpanFrontier returns a new resolvedSpanFrontier.
 func newResolvedSpanFrontier(
-	statementTime hlc.Timestamp, initialHighWater hlc.Timestamp, spans ...roachpb.Span,
+	statementTime hlc.Timestamp,
+	initialHighWater hlc.Timestamp,
+	decoder TablePrefixDecoder,
+	spans ...roachpb.Span,
 ) (*resolvedSpanFrontier, error) {
-	sf, err := span.MakeFrontierAt(initialHighWater, spans...)
+	sf, err := span.NewMultiFrontierAt(newTableIDPartitioner(decoder), initialHighWater, spans...)
 	if err != nil {
 		return nil, err
 	}
-	sf = span.MakeConcurrentFrontier(sf)
 	return &resolvedSpanFrontier{
-		spanFrontier:     sf,
+		MultiFrontier:    sf,
 		statementTime:    statementTime,
 		initialHighWater: initialHighWater,
 	}, nil
@@ -340,7 +345,7 @@ func (f *resolvedSpanFrontier) HasLaggingSpans(sv *settings.Values) bool {
 // All returns an iterator over the resolved spans in the frontier.
 func (f *resolvedSpanFrontier) All() iter.Seq[jobspb.ResolvedSpan] {
 	return func(yield func(jobspb.ResolvedSpan) bool) {
-		for sp, ts := range f.spanFrontier.Entries() {
+		for sp, ts := range f.Entries() {
 			var boundaryType jobspb.ResolvedSpan_BoundaryType
 			if ok, bt := f.boundary.At(ts); ok {
 				boundaryType = bt
@@ -411,3 +416,26 @@ func (b *resolvedSpanBoundary) Forward(newBoundary resolvedSpanBoundary) bool {
 func (b *resolvedSpanBoundary) SafeFormat(s redact.SafePrinter, _ rune) {
 	s.Printf("%v boundary (%v)", b.typ, b.ts)
 }
+
+// A TablePrefixDecoder decodes table prefixes from keys.
+// The production implementation is keys.SQLCodec.
+type TablePrefixDecoder interface {
+	DecodeTablePrefix(key roachpb.Key) ([]byte, uint32, error)
+}
+
+func newTableIDPartitioner(decoder TablePrefixDecoder) span.PartitionerFunc[descpb.ID] {
+	return func(sp roachpb.Span) (descpb.ID, error) {
+		_, startKeyTableID, err := decoder.DecodeTablePrefix(sp.Key)
+		if err != nil {
+			return 0, err
+		}
+		_, endKeyTableID, err := decoder.DecodeTablePrefix(sp.EndKey)
+		if err != nil {
+			return 0, err
+		}
+		if startKeyTableID != endKeyTableID {
+			return 0, errors.AssertionFailedf("span encompassing multiple tables: %s", sp)
+		}
+		return descpb.ID(startKeyTableID), nil
+	}
+}