Merge #152448 #155701 #156190 #156416

152448: changefeedccl: protect system tables with their own pts record  r=andyyang890 a=aerfrei

Previously, we were relying on each pts record that protects
a user table to also protect system tables. This would mean
that we'd protect the system tables once for each table.

Now, we have a dedicated pts record for system tables that
protects back to the highwater and do not include the system
table protections in the table specific records. We only
protect the system tables once.

Fixes: #152446
Epic: CRDB-1421

Release note: None

155701: jobsprotectedtsccl: move to jobsprotectedts r=msbutler a=stevendanna

These tests are now allowed to live in the jobs package so the jobsccl package is no longer require.

Epic: none

Release note: None

156190: changefeedccl: update timestamp on event descriptor cache hits r=andyyang890 a=aerfrei

Before, in some cases we could replan a CDC query with an
old timestamp from the event descriptor cache, which would
try to read the database descriptor at an old potentially
garbage collected timestamp.

Now, we make sure the event descriptor cache always gives
events descriptors with an up to date timestamp.

Epic: none
Fixes: #156091

Release note (bug fix): A bug where changefeeds using CDC
queries could sometimes unexpectedly fail after a schema
change with a descriptor retrieval error has been fixed.

156416: sql/jsonpath: support index acceleration with AnyKey (`*`) at the chain end r=ZhouXing19 a=ZhouXing19

fixes: #156340

We now support index accelerating `jsonb_path_exists` filters with json path expression that ends with an AnyKey (`*`).

Note that the AnyKey is allowed only at the end of the expression. I.e. the following are not allowed:

``` 
$.a.*.b 
$.a.b.*.* 
```

Release note (sql change): We now support index accelerating `jsonb_path_exists` filters with json path expression that ends with an AnyKey (`*`).

Co-authored-by: Aerin Freilich <aerin.freilich@cockroachlabs.com>
Co-authored-by: Steven Danna <danna@cockroachlabs.com>
Co-authored-by: ZhouXing19 <zhouxing@uchicago.edu>

Author: 4 people

pkg/BUILD.bazel

@@ -48,7 +48,6 @@ ALL_TESTS = [
     "//pkg/ccl/cloudccl/externalconn:externalconn_test",
     "//pkg/ccl/cloudccl/gcp:gcp_test",
     "//pkg/ccl/importerccl:importerccl_test",
-    "//pkg/ccl/jobsccl/jobsprotectedtsccl:jobsprotectedtsccl_test",
     "//pkg/ccl/jwtauthccl:jwtauthccl_test",
     "//pkg/ccl/kvccl/kvfollowerreadsccl:kvfollowerreadsccl_test",
     "//pkg/ccl/kvccl/kvtenantccl/upgradeccl:upgradeccl_test",
@@ -212,6 +211,7 @@ ALL_TESTS = [
     "//pkg/jobs/jobsauth:jobsauth_test",
     "//pkg/jobs/jobspb:jobspb_test",
     "//pkg/jobs/jobsprofiler:jobsprofiler_test",
+    "//pkg/jobs/jobsprotectedts:jobsprotectedts_test",
     "//pkg/jobs:jobs_test",
     "//pkg/keys:keys_test",
     "//pkg/keyvisualizer/spanstatsconsumer:spanstatsconsumer_test",
@@ -976,7 +976,6 @@ GO_TARGETS = [
     "//pkg/ccl/cmdccl/stub-schema-registry:stub-schema-registry_lib",
     "//pkg/ccl/gssapiccl:gssapiccl",
     "//pkg/ccl/importerccl:importerccl_test",
-    "//pkg/ccl/jobsccl/jobsprotectedtsccl:jobsprotectedtsccl_test",
     "//pkg/ccl/jwtauthccl:jwtauthccl",
     "//pkg/ccl/jwtauthccl:jwtauthccl_test",
     "//pkg/ccl/kvccl/kvfollowerreadsccl:kvfollowerreadsccl",
@@ -1432,6 +1431,7 @@ GO_TARGETS = [
     "//pkg/jobs/jobsprofiler:jobsprofiler",
     "//pkg/jobs/jobsprofiler:jobsprofiler_test",
     "//pkg/jobs/jobsprotectedts:jobsprotectedts",
+    "//pkg/jobs/jobsprotectedts:jobsprotectedts_test",
     "//pkg/jobs/jobstest:jobstest",
     "//pkg/jobs/metricspoller:metricspoller",
     "//pkg/jobs:jobs",

pkg/ccl/changefeedccl/BUILD.bazel

@@ -282,6 +282,7 @@ go_test(
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/kv/kvserver/protectedts",
         "//pkg/kv/kvserver/protectedts/ptpb",
+        "//pkg/kv/kvserver/protectedts/ptutil",
         "//pkg/multitenant/tenantcapabilitiespb",
         "//pkg/roachpb",
         "//pkg/scheduledjobs",

pkg/ccl/changefeedccl/cdcevent/event.go

@@ -521,9 +521,14 @@ func getEventDescriptorCached(
 	idVer := CacheKey{ID: desc.GetID(), Version: desc.GetVersion(), FamilyID: family.ID}
 
 	if v, ok := cache.Get(idVer); ok {
-		ed := v.(*EventDescriptor)
-		if catalog.UserDefinedTypeColsHaveSameVersion(ed.td, desc) {
-			return ed, nil
+		cached := v.(*EventDescriptor)
+		if catalog.UserDefinedTypeColsHaveSameVersion(cached.td, desc) {
+			// Make a shallow copy to avoid modifying the cached value. The cached
+			// EventDescriptor is shared across changefeed operations and may be
+			// referenced concurrently.
+			ed := *cached
+			ed.SchemaTS = schemaTS
+			return &ed, nil
 		}
 	}
 

pkg/ccl/changefeedccl/cdcprogresspb/progress.proto

@@ -9,11 +9,20 @@ option go_package = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcp
 
 import "gogoproto/gogo.proto";
 
-// ProtectedTimestampRecords is a map from table descriptor IDs to protected timestamp record IDs.
+// ProtectedTimestampRecords contains the IDs of protected timestamps records
+// for a changefeed.
 message ProtectedTimestampRecords {
-  map<uint32, bytes> protected_timestamp_records = 1 [
+  // UserTables is a map from the ID of a user table to the protected timestamp
+  // record that protects it.
+  map<uint32, bytes> user_tables = 1 [
     (gogoproto.castkey) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.ID",
     (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID",
     (gogoproto.nullable) = false
   ];
+  // SystemTables is the ID of PTS record that protects the system tables
+  // back to the changefeed's highwater.
+  bytes system_tables = 2 [
+    (gogoproto.nullable) = false,
+    (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"
+  ];
 }

pkg/ccl/changefeedccl/changefeed_job_info_test.go

@@ -35,7 +35,7 @@ func TestChangefeedJobInfoRoundTrip(t *testing.T) {
 	uuid1 := uuid.MakeV4()
 	uuid2 := uuid.MakeV4()
 	ptsRecords := cdcprogresspb.ProtectedTimestampRecords{
-		ProtectedTimestampRecords: map[descpb.ID]uuid.UUID{
+		UserTables: map[descpb.ID]uuid.UUID{
 			descpb.ID(100): uuid1,
 			descpb.ID(200): uuid2,
 		},

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1999,7 +1999,19 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 	}()
 
 	if cf.spec.ProgressConfig != nil && cf.spec.ProgressConfig.PerTableProtectedTimestamps {
-		return cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater)
+		updatedPerTablePTS, err :=
+			cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater, pts)
+		if err != nil {
+			return false, err
+		}
+
+		updatedSystemTablesPTS, err :=
+			cf.advanceSystemTablesProtectedTimestamp(ctx, txn, &ptsEntries, highwater, pts)
+		if err != nil {
+			return false, err
+		}
+
+		return updatedPerTablePTS || updatedSystemTablesPTS, nil
 	}
 
 	return cf.advanceProtectedTimestamp(ctx, progress, pts, highwater)
@@ -2010,8 +2022,8 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 	txn isql.Txn,
 	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	highwater hlc.Timestamp,
+	pts protectedts.Storage,
 ) (updatedPerTablePTS bool, err error) {
-	pts := cf.FlowCtx.Cfg.ProtectedTimestampProvider.WithTxn(txn)
 	tableIDsToCreate := make(map[descpb.ID]hlc.Timestamp)
 	for tableID, frontier := range cf.frontier.Frontiers() {
 		tableHighWater := func() hlc.Timestamp {
@@ -2023,14 +2035,13 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 			return frontier.Frontier()
 		}()
 
-		if ptsEntries.ProtectedTimestampRecords[tableID] != uuid.Nil {
+		if ptsEntries.UserTables[tableID] != uuid.Nil {
 			if updated, err := cf.advancePerTableProtectedTimestampRecord(ctx, ptsEntries, tableID, tableHighWater, pts); err != nil {
 				return false, err
 			} else if updated {
 				updatedPerTablePTS = true
 			}
 		} else {
-			// TODO(#152448): Do not include system table protections in these records.
 			// TODO(#153894): Newly added/dropped tables should be caught and
 			// protected when starting the frontier, not here.
 			tableIDsToCreate[tableID] = tableHighWater
@@ -2061,7 +2072,7 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 	tableHighWater hlc.Timestamp,
 	pts protectedts.Storage,
 ) (updated bool, err error) {
-	rec, err := pts.GetRecord(ctx, ptsEntries.ProtectedTimestampRecords[tableID])
+	rec, err := pts.GetRecord(ctx, ptsEntries.UserTables[tableID])
 	if err != nil {
 		return false, err
 	}
@@ -2071,7 +2082,7 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 		return false, nil
 	}
 
-	if err := pts.UpdateTimestamp(ctx, ptsEntries.ProtectedTimestampRecords[tableID], tableHighWater); err != nil {
+	if err := pts.UpdateTimestamp(ctx, ptsEntries.UserTables[tableID], tableHighWater); err != nil {
 		return false, err
 	}
 	return true, nil
@@ -2083,19 +2094,19 @@ func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 	tableIDsToCreate map[descpb.ID]hlc.Timestamp,
 	pts protectedts.Storage,
 ) error {
-	if ptsEntries.ProtectedTimestampRecords == nil {
-		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]uuid.UUID)
+	if ptsEntries.UserTables == nil {
+		ptsEntries.UserTables = make(map[descpb.ID]uuid.UUID)
 	}
 	for tableID, tableHighWater := range tableIDsToCreate {
 		targets, err := cf.createPerTablePTSTargets(tableID)
 		if err != nil {
 			return err
 		}
-		ptr := createProtectedTimestampRecord(
+		ptr := createUserTablesProtectedTimestampRecord(
 			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, targets, tableHighWater,
 		)
 		uuid := ptr.ID.GetUUID()
-		ptsEntries.ProtectedTimestampRecords[tableID] = uuid
+		ptsEntries.UserTables[tableID] = uuid
 		if err := pts.Protect(ctx, ptr); err != nil {
 			return err
 		}
@@ -2124,14 +2135,63 @@ func (cf *changeFrontier) createPerTablePTSTargets(
 	return targets, nil
 }
 
+func (cf *changeFrontier) advanceSystemTablesProtectedTimestamp(
+	ctx context.Context,
+	txn isql.Txn,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
+	timestamp hlc.Timestamp,
+	pts protectedts.Storage,
+) (updated bool, err error) {
+	if ptsEntries.SystemTables == uuid.Nil {
+		// All changefeeds using per-table PTS records should have a system tables
+		// PTS record. If they are missing one, it should be made when starting the
+		// changefeed.
+		return false, errors.AssertionFailedf("expected system tables PTS record to be present")
+	}
+
+	rec, err := pts.GetRecord(ctx, ptsEntries.SystemTables)
+	if err != nil {
+		return false, err
+	}
+
+	if !makeSystemTablesTargetToProtect().Equal(rec.Target) {
+		if cf.knobs.PreservePTSTargets != nil && cf.knobs.PreservePTSTargets() {
+			return false, nil
+		}
+		if err := cf.remakeSystemTablesPTSRecord(ctx, txn, pts, ptsEntries, timestamp); err != nil {
+			return false, err
+		}
+		log.VEventf(
+			ctx, 2, "remade system tables PTS record %v to include all targets",
+			ptsEntries.SystemTables,
+		)
+		return true, nil
+	}
+
+	ptsUpdateLag := changefeedbase.ProtectTimestampLag.Get(&cf.FlowCtx.Cfg.Settings.SV)
+	if rec.Timestamp.AddDuration(ptsUpdateLag).After(timestamp) {
+		return false, nil
+	}
+
+	if err := pts.UpdateTimestamp(ctx, ptsEntries.SystemTables, timestamp); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// advanceProtectedTimestamp advances the single PTS record for changefeeds that
+// are not using per-table protected timestamps.
 func (cf *changeFrontier) advanceProtectedTimestamp(
 	ctx context.Context,
 	progress *jobspb.ChangefeedProgress,
 	pts protectedts.Storage,
 	timestamp hlc.Timestamp,
 ) (updated bool, err error) {
 	if progress.ProtectedTimestampRecord == uuid.Nil {
-		ptr := createProtectedTimestampRecord(
+		// For changefeeds not using per-table PTS, system tables are protected
+		// in the single PTS record for the changefeed with all other targets
+		// in a combined record.
+		ptr := createCombinedProtectedTimestampRecord(
 			ctx, cf.FlowCtx.Codec(), cf.spec.JobID, cf.targets, timestamp,
 		)
 		progress.ProtectedTimestampRecord = ptr.ID.GetUUID()
@@ -2159,7 +2219,7 @@ func (cf *changeFrontier) advanceProtectedTimestamp(
 	// If we've identified more tables that need to be protected since this
 	// changefeed was created, it will be missing here. If so, we "migrate" it
 	// to include all the appropriate targets.
-	if !makeTargetToProtect(cf.targets).Equal(rec.Target) {
+	if !makeCombinedTargetToProtect(cf.targets).Equal(rec.Target) {
 		if preservePTSTargets := cf.knobs.PreservePTSTargets != nil && cf.knobs.PreservePTSTargets(); preservePTSTargets {
 			return false, nil
 		}
@@ -2190,7 +2250,7 @@ func (cf *changeFrontier) remakePTSRecord(
 	resolved hlc.Timestamp,
 ) error {
 	prevRecordId := progress.ProtectedTimestampRecord
-	ptr := createProtectedTimestampRecord(
+	ptr := createCombinedProtectedTimestampRecord(
 		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, cf.targets, resolved,
 	)
 	if err := pts.Protect(ctx, ptr); err != nil {
@@ -2207,6 +2267,29 @@ func (cf *changeFrontier) remakePTSRecord(
 	return nil
 }
 
+func (cf *changeFrontier) remakeSystemTablesPTSRecord(
+	ctx context.Context,
+	txn isql.Txn,
+	pts protectedts.Storage,
+	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
+	resolved hlc.Timestamp,
+) error {
+	ptr := createSystemTablesProtectedTimestampRecord(
+		ctx, cf.FlowCtx.Codec(), cf.spec.JobID, resolved,
+	)
+	if err := pts.Protect(ctx, ptr); err != nil {
+		return err
+	}
+	prevRecordId := ptsEntries.SystemTables
+	if err := pts.Release(ctx, prevRecordId); err != nil {
+		return err
+	}
+	ptsEntries.SystemTables = ptr.ID.GetUUID()
+	log.Eventf(ctx, "created new system tables pts record %v to replace old pts record %v at %v",
+		ptsEntries.SystemTables, prevRecordId, resolved)
+	return writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID)
+}
+
 func (cf *changeFrontier) maybeEmitResolved(ctx context.Context, newResolved hlc.Timestamp) error {
 	if cf.freqEmitResolved == emitNoResolved || newResolved.IsEmpty() {
 		return nil

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -327,6 +327,9 @@ func changefeedPlanHook(
 			// perTablePTSRecords is the per-table protected timestamp records which exist when
 			// per-table protected timestamps are enabled.
 			var perTablePTSRecords []*ptpb.Record
+			// systemTablesPTSRecord is the system tables protected timestamp record which exists when
+			// per-table protected timestamps are enabled.
+			var systemTablesPTSRecord *ptpb.Record
 			// ptsRecords is the protected timestamp records object containing all per-table protected
 			// timestamp records. Its format matches what will be persisted to the job info table.
 			var ptsRecords *cdcprogresspb.ProtectedTimestampRecords
@@ -339,9 +342,11 @@ func changefeedPlanHook(
 			if usingPerTablePTS {
 				protectedTimestampRecords := make(map[descpb.ID]uuid.UUID)
 				if err := targets.EachTarget(func(target changefeedbase.Target) error {
+					// TODO(#155957): We are likely leaking PTS records here in
+					// the column families case.
 					ptsTargets := changefeedbase.Targets{}
 					ptsTargets.Add(target)
-					ptsRecord := createProtectedTimestampRecord(
+					ptsRecord := createUserTablesProtectedTimestampRecord(
 						ctx,
 						codec,
 						jobID,
@@ -355,11 +360,18 @@ func changefeedPlanHook(
 				}); err != nil {
 					return err
 				}
+				systemTablesPTSRecord = createSystemTablesProtectedTimestampRecord(
+					ctx,
+					codec,
+					jobID,
+					details.StatementTime,
+				)
 				ptsRecords = &cdcprogresspb.ProtectedTimestampRecords{
-					ProtectedTimestampRecords: protectedTimestampRecords,
+					UserTables:   protectedTimestampRecords,
+					SystemTables: systemTablesPTSRecord.ID.GetUUID(),
 				}
 			} else {
-				ptr = createProtectedTimestampRecord(
+				ptr = createCombinedProtectedTimestampRecord(
 					ctx,
 					codec,
 					jobID,
@@ -380,12 +392,15 @@ func changefeedPlanHook(
 					}
 				}
 				if usingPerTablePTS {
+					pts := p.ExecCfg().ProtectedTimestampProvider.WithTxn(p.InternalSQLTxn())
 					for _, perTableRecord := range perTablePTSRecords {
-						pts := p.ExecCfg().ProtectedTimestampProvider.WithTxn(p.InternalSQLTxn())
 						if err := pts.Protect(ctx, perTableRecord); err != nil {
 							return err
 						}
 					}
+					if err := pts.Protect(ctx, systemTablesPTSRecord); err != nil {
+						return err
+					}
 					if err := writeChangefeedJobInfo(
 						ctx, perTableProtectedTimestampsFilename, ptsRecords, p.InternalSQLTxn(), jobID,
 					); err != nil {
@@ -428,6 +443,9 @@ func changefeedPlanHook(
 							return err
 						}
 					}
+					if err := pts.Protect(ctx, systemTablesPTSRecord); err != nil {
+						return err
+					}
 					if err := writeChangefeedJobInfo(
 						ctx, perTableProtectedTimestampsFilename, ptsRecords, txn, jobID,
 					); err != nil {
@@ -1966,20 +1984,27 @@ func (b *changefeedResumer) OnFailOrCancel(
 	}
 
 	maybeCleanUpProtectedTimestamp(progress.GetChangefeed().ProtectedTimestampRecord)
+	// We clean up the per-table protected timestamps (and their accompanying
+	// system tables protected timestamp record) in a transaction since we need
+	// to read from the job info.
 	if err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		ptsEntries := cdcprogresspb.ProtectedTimestampRecords{}
 		if err := readChangefeedJobInfo(
 			ctx, perTableProtectedTimestampsFilename, &ptsEntries, txn, b.job.ID(),
 		); err != nil {
 			return err
 		}
-
-		if len(ptsEntries.ProtectedTimestampRecords) == 0 {
+		// In the event that the changefeed is not using per-table protected
+		// timestamps, the ptsEntries populated from the job info table
+		// (in the file perTableProtectedTimestampsFilename) will be empty.
+		// There is nothing to clean up, so we can safely return here.
+		if len(ptsEntries.UserTables) == 0 && ptsEntries.SystemTables == uuid.Nil {
 			return nil
 		}
-		for _, record := range ptsEntries.ProtectedTimestampRecords {
+		for _, record := range ptsEntries.UserTables {
 			maybeCleanUpProtectedTimestamp(record)
 		}
+		maybeCleanUpProtectedTimestamp(ptsEntries.SystemTables)
 		return deleteChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, txn, b.job.ID())
 	}); err != nil {
 		return err