changefeedccl/resolvedspan: add feature flag for per-table tracking

This patch adds a new cluster setting named
`changefeed.frontier.per_table_tracking.enabled`
to feature flag enabling per-table tracking in
the resolved span frontiers.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -616,7 +616,9 @@ func (ca *changeAggregator) setupSpansAndFrontier() (spans []roachpb.Span, err e
 		return nil, err
 	}
 	ca.frontier, err = resolvedspan.NewAggregatorFrontier(
-		ca.spec.Feed.StatementTime, initialHighWater, ca.FlowCtx.Codec(), spans...)
+		ca.spec.Feed.StatementTime, initialHighWater, ca.FlowCtx.Codec(),
+		&ca.FlowCtx.Cfg.Settings.SV,
+		spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -1421,7 +1423,9 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 
 	// Set up the resolved span frontier.
 	cf.frontier, err = resolvedspan.NewCoordinatorFrontier(
-		cf.spec.Feed.StatementTime, initialHighwater, cf.FlowCtx.Codec(), cf.spec.TrackedSpans...)
+		cf.spec.Feed.StatementTime, initialHighwater, cf.FlowCtx.Codec(),
+		&cf.FlowCtx.Cfg.Settings.SV,
+		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_test.go

@@ -8940,6 +8940,7 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 			hlc.Timestamp{},
 			hlc.Timestamp{},
 			s.Codec,
+			&s.Server.ClusterSettings().SV,
 			tableSpan,
 		)
 		if err != nil {

pkg/ccl/changefeedccl/resolvedspan/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "//pkg/settings",
         "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
+        "//pkg/util/metamorphic",
         "//pkg/util/span",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
@@ -30,6 +31,7 @@ go_test(
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/roachpb",
+        "//pkg/settings/cluster",
         "//pkg/sql/catalog/descpb",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",

pkg/ccl/changefeedccl/resolvedspan/frontier.go

@@ -16,11 +16,22 @@ import (
 	"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/metamorphic"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
 
+// FrontierPerTableTracking controls whether the in-memory frontiers changefeeds
+// use for tracking span progress should do per-table tracking (via partitioning
+// into one sub-frontier per table).
+var FrontierPerTableTracking = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"changefeed.frontier.per_table_tracking.enabled",
+	"track progress on a per-table basis in-memory",
+	metamorphic.ConstantWithTestBool("changefeed.frontier.per_table_tracking.enabled", true),
+)
+
 // AggregatorFrontier wraps a resolvedSpanFrontier with additional
 // checks specific to how change aggregators process boundaries.
 type AggregatorFrontier struct {
@@ -32,9 +43,10 @@ func NewAggregatorFrontier(
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
 	decoder TablePrefixDecoder,
+	sv *settings.Values,
 	spans ...roachpb.Span,
 ) (*AggregatorFrontier, error) {
-	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, spans...)
+	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, sv, spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -80,9 +92,10 @@ func NewCoordinatorFrontier(
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
 	decoder TablePrefixDecoder,
+	sv *settings.Values,
 	spans ...roachpb.Span,
 ) (*CoordinatorFrontier, error) {
-	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, spans...)
+	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, sv, spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -195,7 +208,7 @@ func (f *CoordinatorFrontier) MakeCheckpoint(
 // used to track resolved spans for a changefeed and methods for computing
 // lagging and checkpoint spans.
 type resolvedSpanFrontier struct {
-	*span.MultiFrontier[descpb.ID]
+	maybeTablePartitionedFrontier
 
 	// statementTime is the statement time of the changefeed.
 	statementTime hlc.Timestamp
@@ -217,16 +230,26 @@ func newResolvedSpanFrontier(
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
 	decoder TablePrefixDecoder,
+	sv *settings.Values,
 	spans ...roachpb.Span,
 ) (*resolvedSpanFrontier, error) {
-	sf, err := span.NewMultiFrontierAt(newTableIDPartitioner(decoder), initialHighWater, spans...)
+	sf, err := func() (maybeTablePartitionedFrontier, error) {
+		if FrontierPerTableTracking.Get(sv) {
+			return span.NewMultiFrontierAt(newTableIDPartitioner(decoder), initialHighWater, spans...)
+		}
+		f, err := span.MakeFrontierAt(initialHighWater, spans...)
+		if err != nil {
+			return nil, err
+		}
+		return notTablePartitionedFrontier{spanFrontier: span.MakeConcurrentFrontier(f)}, nil
+	}()
 	if err != nil {
 		return nil, err
 	}
 	return &resolvedSpanFrontier{
-		MultiFrontier:    sf,
-		statementTime:    statementTime,
-		initialHighWater: initialHighWater,
+		maybeTablePartitionedFrontier: sf,
+		statementTime:                 statementTime,
+		initialHighWater:              initialHighWater,
 	}, nil
 }
 
@@ -417,6 +440,39 @@ func (b *resolvedSpanBoundary) SafeFormat(s redact.SafePrinter, _ rune) {
 	s.Printf("%v boundary (%v)", b.typ, b.ts)
 }
 
+// A maybeTablePartitionedFrontier is a frontier that might be tracking
+// spans in per-table sub-frontiers.
+type maybeTablePartitionedFrontier interface {
+	span.Frontier
+
+	// Frontiers returns an iterator over the table ID and sub-frontiers
+	// being tracked by the frontier. If the frontier is not tracking
+	// on a per-table basis, the iterator will return a single frontier
+	// with an ID of 0.
+	Frontiers() iter.Seq2[descpb.ID, span.Frontier]
+}
+
+var _ maybeTablePartitionedFrontier = (*span.MultiFrontier[descpb.ID])(nil)
+
+// 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
+
+// notTablePartitionedFrontier is a frontier that does not track spans on
+// a per-table basis.
+type notTablePartitionedFrontier struct {
+	spanFrontier
+}
+
+var _ maybeTablePartitionedFrontier = notTablePartitionedFrontier{}
+
+// Frontiers implements maybeTablePartitionedFrontier.
+func (f notTablePartitionedFrontier) Frontiers() iter.Seq2[descpb.ID, span.Frontier] {
+	return func(yield func(descpb.ID, span.Frontier) bool) {
+		yield(0, f.spanFrontier)
+	}
+}
+
 // A TablePrefixDecoder decodes table prefixes from keys.
 // The production implementation is keys.SQLCodec.
 type TablePrefixDecoder interface {

pkg/ccl/changefeedccl/resolvedspan/frontier_test.go

@@ -6,13 +6,15 @@
 package resolvedspan_test
 
 import (
+	"context"
 	"iter"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/resolvedspan"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -27,13 +29,16 @@ func TestAggregatorFrontier(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	st := cluster.MakeTestingClusterSettings()
+
 	// Create a fresh frontier with no progress.
 	statementTime := makeTS(10)
 	var initialHighwater hlc.Timestamp
 	f, err := resolvedspan.NewAggregatorFrontier(
 		statementTime,
 		initialHighwater,
 		mockDecoder{},
+		&st.SV,
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -79,6 +84,7 @@ func TestAggregatorFrontier(t *testing.T) {
 		statementTime,
 		initialHighwater,
 		mockDecoder{},
+		&st.SV,
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -97,13 +103,16 @@ func TestCoordinatorFrontier(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	st := cluster.MakeTestingClusterSettings()
+
 	// Create a fresh frontier with no progress.
 	statementTime := makeTS(10)
 	var initialHighwater hlc.Timestamp
 	f, err := resolvedspan.NewCoordinatorFrontier(
 		statementTime,
 		initialHighwater,
 		mockDecoder{},
+		&st.SV,
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -152,6 +161,7 @@ func TestCoordinatorFrontier(t *testing.T) {
 		statementTime,
 		initialHighwater,
 		mockDecoder{},
+		&st.SV,
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -256,11 +266,14 @@ func TestAggregatorFrontier_ForwardResolvedSpan(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	st := cluster.MakeTestingClusterSettings()
+
 	// Create a fresh frontier with no progress.
 	f, err := resolvedspan.NewAggregatorFrontier(
 		hlc.Timestamp{},
 		hlc.Timestamp{},
 		mockDecoder{},
+		&st.SV,
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -315,6 +328,12 @@ func TestFrontierPerTableResolvedTimestamps(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	ctx := context.Background()
+	st := cluster.MakeTestingClusterSettings()
+
+	// Explicitly enable per-table tracking for this test.
+	resolvedspan.FrontierPerTableTracking.Override(ctx, &st.SV, true)
+
 	for _, frontierType := range []string{"aggregator", "coordinator"} {
 		t.Run(frontierType, func(t *testing.T) {
 			rnd, _ := randutil.NewPseudoRand()
@@ -354,13 +373,15 @@ func TestFrontierPerTableResolvedTimestamps(t *testing.T) {
 						statementTime,
 						initialHighWater,
 						codec,
+						&st.SV,
 						tableSpans...,
 					)
 				case "coordinator":
 					return resolvedspan.NewCoordinatorFrontier(
 						statementTime,
 						initialHighWater,
 						codec,
+						&st.SV,
 						tableSpans...,
 					)
 				default: