Merge #153177 #153367

153177: changefeedccl: remove additional job info write from pts management r=andyyang890,KeithCh a=aerfrei

In per table PTS management, if a new table started lagging while another table stopped lagging, we would do two separate job info table writes. We would do one to create the new PTS record and one to remove the old one.

With this change, we do only one job info table write.

Epic: CRDB-1421

Release note: None

153367: build: fix some small errors r=rail a=rickystewart

These were found by an LLM. None are severe, but to minimize the chances of unexpected errors or avoid further confusion we address them.

Epic: none
Release note: None

Co-authored-by: Aerin Freilich <[email protected]>
Co-authored-by: Ricky Stewart <[email protected]>

Author: 3 people

build/bazelbuilder/Dockerfile

@@ -89,10 +89,11 @@ RUN if [[ ${TARGETPLATFORM} != "linux/s390x" ]]; then \
 # NB: As above, this is not available on `s390x`.
 RUN if [[ ${TARGETPLATFORM} != "linux/s390x" ]]; then \
  case ${TARGETPLATFORM} in \
-     "linux/amd64") ARCH=x86_64; SHASUM= ;; \
+     "linux/amd64") ARCH=x86_64; SHASUM=8ba7e746ca05f225e5a73952bbc03f4086a5f65fd94f3717df6f75f212587159 ;; \
      "linux/arm64") ARCH=arm; SHASUM=e6153461e3154ebce61d35b73005bdd14a0ecacd42e5008f66e25b4ad231e5c9 ;; \
  esac  \
  && curl -fsSL "https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/google-cloud-cli-linux-$ARCH.tar.gz" -o gcloud.tar.gz \
+ && echo "$SHASUM gcloud.tar.gz" | sha256sum -c - \
  && tar -xzf gcloud.tar.gz \
  && rm gcloud.tar.gz ; \
  fi

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -2022,14 +2022,19 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 	}
 
 	if len(tableIDsToRelease) > 0 {
-		if err := cf.releasePerTableProtectedTimestampRecords(ctx, txn, ptsEntries, tableIDsToRelease, pts); err != nil {
+		if err := cf.releasePerTableProtectedTimestampRecords(ctx, ptsEntries, tableIDsToRelease, pts); err != nil {
 			return hlc.Timestamp{}, false, err
 		}
-		updatedPerTablePTS = true
 	}
 
 	if len(tableIDsToCreate) > 0 {
-		if err := cf.createPerTableProtectedTimestampRecords(ctx, txn, ptsEntries, tableIDsToCreate, pts); err != nil {
+		if err := cf.createPerTableProtectedTimestampRecords(ctx, ptsEntries, tableIDsToCreate, pts); err != nil {
+			return hlc.Timestamp{}, false, err
+		}
+	}
+
+	if len(tableIDsToRelease) > 0 || len(tableIDsToCreate) > 0 {
+		if err := writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID); err != nil {
 			return hlc.Timestamp{}, false, err
 		}
 		updatedPerTablePTS = true
@@ -2040,7 +2045,6 @@ func (cf *changeFrontier) managePerTableProtectedTimestamps(
 
 func (cf *changeFrontier) releasePerTableProtectedTimestampRecords(
 	ctx context.Context,
-	txn isql.Txn,
 	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	tableIDs []descpb.ID,
 	pts protectedts.Storage,
@@ -2051,7 +2055,7 @@ func (cf *changeFrontier) releasePerTableProtectedTimestampRecords(
 		}
 		delete(ptsEntries.ProtectedTimestampRecords, tableID)
 	}
-	return writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID)
+	return nil
 }
 
 func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
@@ -2079,7 +2083,6 @@ func (cf *changeFrontier) advancePerTableProtectedTimestampRecord(
 
 func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 	ctx context.Context,
-	txn isql.Txn,
 	ptsEntries *cdcprogresspb.ProtectedTimestampRecords,
 	tableIDsToCreate map[descpb.ID]hlc.Timestamp,
 	pts protectedts.Storage,
@@ -2088,7 +2091,7 @@ func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 		ptsEntries.ProtectedTimestampRecords = make(map[descpb.ID]*uuid.UUID)
 	}
 	for tableID, tableHighWater := range tableIDsToCreate {
-		targets, err := cf.createPerTablePTSTarget(tableID)
+		targets, err := cf.createPerTablePTSTargets(tableID)
 		if err != nil {
 			return err
 		}
@@ -2101,22 +2104,23 @@ func (cf *changeFrontier) createPerTableProtectedTimestampRecords(
 			return err
 		}
 	}
-	return writeChangefeedJobInfo(ctx, perTableProtectedTimestampsFilename, ptsEntries, txn, cf.spec.JobID)
+	return nil
 }
 
-func (cf *changeFrontier) createPerTablePTSTarget(
+func (cf *changeFrontier) createPerTablePTSTargets(
 	tableID descpb.ID,
 ) (changefeedbase.Targets, error) {
 	targets := changefeedbase.Targets{}
-	if cf.targets.Size > 0 {
-		if found, err := cf.targets.EachHavingTableID(tableID, func(target changefeedbase.Target) error {
-			targets.Add(target)
-			return nil
-		}); err != nil {
-			return changefeedbase.Targets{}, err
-		} else if !found {
-			return changefeedbase.Targets{}, errors.AssertionFailedf("attempted to create a per-table PTS record for table %d, but no target was found", tableID)
-		}
+	if found, err := cf.targets.EachHavingTableID(tableID, func(target changefeedbase.Target) error {
+		targets.Add(target)
+		return nil
+	}); err != nil {
+		return changefeedbase.Targets{}, err
+	} else if !found {
+		return changefeedbase.Targets{}, errors.AssertionFailedf(
+			"attempted to create a per-table PTS record for table %d, but no target was found",
+			tableID,
+		)
 	}
 	if targets.Size != 1 {
 		return changefeedbase.Targets{}, errors.AssertionFailedf("expected 1 target, got %d", targets.Size)

pkg/cmd/bazci/bazci.go

@@ -320,6 +320,9 @@ func sendBepDataToBeaverHub(bepFilepath string) error {
 }
 
 func bazciImpl(cmd *cobra.Command, args []string) error {
+	if len(args) == 0 {
+		return errors.Newf("must provide some subcommand (`build`, `run`, `test`, `coverage`, `merge-test-xmls`, or `munge-test-xml`)")
+	}
 	if args[0] != buildSubcmd && args[0] != runSubcmd && args[0] != coverageSubcmd &&
 		args[0] != testSubcmd && args[0] != mungeTestXMLSubcmd && args[0] != mergeTestXMLsSubcmd {
 		return errors.Newf("First argument must be `build`, `run`, `test`, `coverage`, `merge-test-xmls`, or `munge-test-xml`; got %v", args[0])