Merge #151615 #151675 #151679

151615: execinfrapb: add ChangefeedProgressConfig protobuf r=aerfrei,asg0451 a=andyyang890

This patch adds a new `ChangefeedProgressConfig` protobuf message,
which is being added to changefeed processor specs so that all of
a changefeed's processors use the same consistent configuration
for progress tracking and recording.

Fixes #151613

Release note: None

151675: tree: fix formatting of ON COMMIT PRESERVE ROWS r=rafiss a=rafiss

Formatting for this clause was added in
d80409c, but that had a mistake since the locality clause should always be formatted after the ON COMMIT clause.

informs #149949
Release note: None

151679: keyside: temporarily don't return assertion error for LTREE r=yuzefovich a=yuzefovich

We merged new LTREE datum type only with value encoding, yet this type can be key-encodable (but that is not implemented). In such case, in test builds we return an assertion error for missing key encoding, and many of our tests fail if they see an assertion error. To silence this expected failure mode we add an exception for LTREE to result in a regular error for now, which I think should cover most test failures we've seen.

Fixes: #151614.
Fixes: #151659.

Release note: None

Co-authored-by: Andy Yang <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/kvccl/kvfollowerreadsccl"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -453,6 +454,14 @@ func makePlan(
 			maybeCfKnobs.SpanPartitionsCallback(spanPartitions)
 		}
 
+		// Create progress config based on current settings.
+		var progressConfig *execinfrapb.ChangefeedProgressConfig
+		if execCtx.ExecCfg().Settings.Version.IsActive(ctx, clusterversion.V25_4) {
+			progressConfig = &execinfrapb.ChangefeedProgressConfig{
+				PerTableTracking: changefeedbase.TrackPerTableProgress.Get(sv),
+			}
+		}
+
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {
 			if log.ExpensiveLogEnabled(ctx, 2) {
@@ -475,6 +484,7 @@ func makePlan(
 				JobID:               jobID,
 				Select:              execinfrapb.Expression{Expr: details.Select},
 				Description:         description,
+				ProgressConfig:      progressConfig,
 			}
 		}
 
@@ -489,6 +499,7 @@ func makePlan(
 			JobID:               jobID,
 			UserProto:           execCtx.User().EncodeProto(),
 			Description:         description,
+			ProgressConfig:      progressConfig,
 		}
 
 		if haveKnobs && maybeCfKnobs.OnDistflowSpec != nil {

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -615,9 +615,13 @@ func (ca *changeAggregator) setupSpansAndFrontier() (spans []roachpb.Span, err e
 	if err != nil {
 		return nil, err
 	}
+	var perTableTracking bool
+	if ca.spec.ProgressConfig != nil {
+		perTableTracking = ca.spec.ProgressConfig.PerTableTracking
+	}
 	ca.frontier, err = resolvedspan.NewAggregatorFrontier(
 		ca.spec.Feed.StatementTime, initialHighWater, ca.FlowCtx.Codec(),
-		&ca.FlowCtx.Cfg.Settings.SV,
+		perTableTracking,
 		spans...)
 	if err != nil {
 		return nil, err
@@ -1422,9 +1426,13 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 	}
 
 	// Set up the resolved span frontier.
+	var perTableTracking bool
+	if cf.spec.ProgressConfig != nil {
+		perTableTracking = cf.spec.ProgressConfig.PerTableTracking
+	}
 	cf.frontier, err = resolvedspan.NewCoordinatorFrontier(
 		cf.spec.Feed.StatementTime, initialHighwater, cf.FlowCtx.Codec(),
-		&cf.FlowCtx.Cfg.Settings.SV,
+		perTableTracking,
 		cf.spec.TrackedSpans...)
 	if err != nil {
 		log.Infof(cf.Ctx(), "change frontier moving to draining due to error setting up frontier: %v", err)

pkg/ccl/changefeedccl/changefeed_test.go

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

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -388,3 +388,16 @@ var UseBareTableNames = settings.RegisterBoolSetting(
 	"changefeed.bare_table_names.enabled",
 	"set to true to use bare table names in changefeed topics, false to use quoted table names; default is true",
 	true)
+
+// TrackPerTableProgress controls whether a changefeed's in-memory frontiers
+// should track span progress on a per-table basis (via partitioning into
+// one sub-frontier per table). Enabling this is necessary for any other
+// per-table progress features (e.g. per-table PTS) to work.
+var TrackPerTableProgress = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"changefeed.progress.per_table_tracking.enabled",
+	"track progress on a per-table basis in-memory; enabling this will enable more "+
+		"granular saving/restoring of progress, which will reduce duplicates during restarts, "+
+		"but doing so may incur additional overhead during ordinary changefeed execution",
+	metamorphic.ConstantWithTestBool("changefeed.progress.per_table_tracking.enabled", true),
+)

pkg/ccl/changefeedccl/resolvedspan/BUILD.bazel

@@ -16,7 +16,6 @@ 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",
@@ -31,7 +30,6 @@ 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,22 +16,11 @@ 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 {
@@ -43,10 +32,10 @@ func NewAggregatorFrontier(
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
 	decoder TablePrefixDecoder,
-	sv *settings.Values,
+	perTableTracking bool,
 	spans ...roachpb.Span,
 ) (*AggregatorFrontier, error) {
-	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, sv, spans...)
+	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, perTableTracking, spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -92,10 +81,10 @@ func NewCoordinatorFrontier(
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
 	decoder TablePrefixDecoder,
-	sv *settings.Values,
+	perTableTracking bool,
 	spans ...roachpb.Span,
 ) (*CoordinatorFrontier, error) {
-	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, sv, spans...)
+	rsf, err := newResolvedSpanFrontier(statementTime, initialHighWater, decoder, perTableTracking, spans...)
 	if err != nil {
 		return nil, err
 	}
@@ -230,11 +219,11 @@ func newResolvedSpanFrontier(
 	statementTime hlc.Timestamp,
 	initialHighWater hlc.Timestamp,
 	decoder TablePrefixDecoder,
-	sv *settings.Values,
+	perTableTracking bool,
 	spans ...roachpb.Span,
 ) (*resolvedSpanFrontier, error) {
 	sf, err := func() (maybeTablePartitionedFrontier, error) {
-		if FrontierPerTableTracking.Get(sv) {
+		if perTableTracking {
 			return span.NewMultiFrontierAt(newTableIDPartitioner(decoder), initialHighWater, spans...)
 		}
 		f, err := span.MakeFrontierAt(initialHighWater, spans...)

pkg/ccl/changefeedccl/resolvedspan/frontier_test.go

@@ -6,15 +6,13 @@
 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"
@@ -29,16 +27,14 @@ 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,
+		false, /* perTableTracking */
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -84,7 +80,7 @@ func TestAggregatorFrontier(t *testing.T) {
 		statementTime,
 		initialHighwater,
 		mockDecoder{},
-		&st.SV,
+		false, /* perTableTracking */
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -103,16 +99,14 @@ 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,
+		false, /* perTableTracking */
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -161,7 +155,7 @@ func TestCoordinatorFrontier(t *testing.T) {
 		statementTime,
 		initialHighwater,
 		mockDecoder{},
-		&st.SV,
+		false, /* perTableTracking */
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -266,14 +260,12 @@ 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,
+		false, /* perTableTracking */
 		makeSpan("a", "f"),
 	)
 	require.NoError(t, err)
@@ -328,12 +320,6 @@ 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()
@@ -373,15 +359,15 @@ func TestFrontierPerTableResolvedTimestamps(t *testing.T) {
 						statementTime,
 						initialHighWater,
 						codec,
-						&st.SV,
+						true, /* perTableTracking */
 						tableSpans...,
 					)
 				case "coordinator":
 					return resolvedspan.NewCoordinatorFrontier(
 						statementTime,
 						initialHighWater,
 						codec,
-						&st.SV,
+						true, /* perTableTracking */
 						tableSpans...,
 					)
 				default:

pkg/sql/execinfrapb/processors_changefeeds.proto

@@ -61,6 +61,9 @@ message ChangeAggregatorSpec {
   // have been resolved to the given timestamp, so it is safe to forward these
   // spans to its corresponding timestamps upon resuming.
   optional cockroach.sql.jobs.jobspb.TimestampSpansMap span_level_checkpoint = 9;
+
+  // ProgressConfig is the configuration for the changefeed's progress tracking.
+  optional ChangefeedProgressConfig progress_config = 10;
 }
 
 // ChangeFrontierSpec is the specification for a processor that receives
@@ -92,4 +95,15 @@ message ChangeFrontierSpec {
   // SpanLevelCheckpoint is a map from timestamps to lists of spans that captures
   // the changefeed progress and is used to initialize the frontier on resume.
   optional cockroach.sql.jobs.jobspb.TimestampSpansMap span_level_checkpoint = 6;
+
+  // ProgressConfig is the configuration for the changefeed's progress tracking.
+  optional ChangefeedProgressConfig progress_config = 7;
+}
+
+// ChangefeedProgressConfig is the configuration for the changefeed's progress tracking.
+message ChangefeedProgressConfig {
+  // PerTableTracking configures whether the changefeed should track
+  // span progress on a per-table basis. If configured, any span frontiers
+  // created will track spans belonging to different tables separately.
+  optional bool per_table_tracking = 1 [(gogoproto.nullable) = false];
 }

pkg/sql/parser/testdata/create_table

@@ -2687,3 +2687,19 @@ CREATE TABLE t (l LTREE) -- normalized!
 CREATE TABLE t (l LTREE) -- fully parenthesized
 CREATE TABLE t (l LTREE) -- literals removed
 CREATE TABLE _ (_ LTREE) -- identifiers removed
+
+parse
+CREATE TABLE tbl (a INT PRIMARY KEY) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION
+----
+CREATE TABLE tbl (a INT8 PRIMARY KEY) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION -- normalized!
+CREATE TABLE tbl (a INT8 PRIMARY KEY) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION -- fully parenthesized
+CREATE TABLE tbl (a INT8 PRIMARY KEY) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION -- literals removed
+CREATE TABLE _ (_ INT8 PRIMARY KEY) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION -- identifiers removed
+
+error
+CREATE TABLE tbl AS (SELECT * FROM t) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION
+----
+at or near "locality": syntax error
+DETAIL: source SQL:
+CREATE TABLE tbl AS (SELECT * FROM t) ON COMMIT PRESERVE ROWS LOCALITY REGIONAL BY TABLE IN PRIMARY REGION
+                                                              ^

pkg/sql/rowenc/keyside/encode.go

@@ -183,6 +183,11 @@ func Encode(b []byte, val tree.Datum, dir encoding.Direction) ([]byte, error) {
 		return encodeJSONKey(b, t, dir)
 	}
 	if buildutil.CrdbTestBuild {
+		if _, isLTree := val.(*tree.DLTree); isLTree {
+			// TODO(paulniziolek): remove this exception once key encoding is
+			// added.
+			return nil, errors.Newf("LTREE key encoding is not implemented yet")
+		}
 		return nil, errors.AssertionFailedf("unable to encode table key: %T", val)
 	}
 	return nil, errors.Errorf("unable to encode table key: %T", val)