changefeedccl: fix ALTER CHANGEFEED max PTS age bug

This patch fixes a bug where modifying a changefeed with ALTER CHANGEFEED
that either unset or left the `gc_protect_expires_after` option unset would
cause the changefeed's max PTS age to become unbounded instead of being set to
the default value configured by the `changefeed.protect_timestamp.max_age`
cluster setting.

This bug was occurring because the ALTER CHANGEFEED code was manually querying
the option instead of relying on the CREATE CHANGEFEED code path that also
consulted the cluster setting. This inconsistency has now been corrected.

Release note (bug fix): A bug where modifying a changefeed with ALTER CHANGEFEED
that either unset or left the `gc_protect_expires_after` option unset would
cause the changefeed's max PTS age to become unbounded instead of being set to
the default value configured by the `changefeed.protect_timestamp.max_age`
cluster setting.

Author: andyyang890

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -227,11 +227,11 @@ func alterChangefeedPlanHook(
 		newPayload.Details = jobspb.WrapPayloadDetails(newDetails)
 		newPayload.Description = jobRecord.Description
 		newPayload.DescriptorIDs = jobRecord.DescriptorIDs
-		newExpiration, err := newOptions.GetPTSExpiration()
-		if err != nil {
-			return err
-		}
-		newPayload.MaximumPTSAge = newExpiration
+
+		// The maximum PTS age on jobRecord will be set correctly (based on either
+		// the option or cluster setting) by createChangefeedJobRecord.
+		newPayload.MaximumPTSAge = jobRecord.MaximumPTSAge
+
 		j, err := p.ExecCfg().JobRegistry.LoadJobWithTxn(ctx, jobID, p.InternalSQLTxn())
 		if err != nil {
 			return err

pkg/ccl/changefeedccl/protected_timestamps_test.go

@@ -424,27 +424,114 @@ func TestChangefeedCanceledWhenPTSIsOld(t *testing.T) {
 		// single row with multiple versions.
 		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b INT)`)
 
-		feed, err := f.Feed("CREATE CHANGEFEED FOR TABLE foo WITH protect_data_from_gc_on_pause, gc_protect_expires_after='24h'")
-		require.NoError(t, err)
-		defer func() {
-			closeFeed(t, feed)
-		}()
+		t.Run("canceled due to gc_protect_expires_after option", func(t *testing.T) {
+			testutils.RunValues(t, "initially-protected-with", []string{"none", "option", "setting"},
+				func(t *testing.T, initialProtect string) {
+					defer func() {
+						sqlDB.Exec(t, `RESET CLUSTER SETTING changefeed.protect_timestamp.max_age`)
+					}()
+
+					if initialProtect == "option" {
+						// We set the cluster setting to something small to make sure that
+						// the option alone is able to protect the PTS record.
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '1us'`)
+					} else {
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '24h'`)
+					}
 
-		jobFeed := feed.(cdctest.EnterpriseTestFeed)
-		require.NoError(t, jobFeed.Pause())
+					feedStmt := `CREATE CHANGEFEED FOR TABLE foo`
+					switch initialProtect {
+					case "none":
+						feedStmt += ` WITH gc_protect_expires_after='1us'`
+					case "option":
+						feedStmt += ` WITH gc_protect_expires_after='24h'`
+					}
+
+					feed, err := f.Feed(feedStmt)
+					require.NoError(t, err)
+					defer func() {
+						closeFeed(t, feed)
+					}()
 
-		// While the job is paused, take opportunity to test that alter changefeed
-		// works when setting gc_protect_expires_after option.
+					jobFeed := feed.(cdctest.EnterpriseTestFeed)
 
-		// Verify we can set it to 0 -- i.e. disable.
-		sqlDB.Exec(t, fmt.Sprintf("ALTER CHANGEFEED %d SET gc_protect_expires_after = '0s'", jobFeed.JobID()))
-		// Now, set it to something very small.
-		sqlDB.Exec(t, fmt.Sprintf("ALTER CHANGEFEED %d SET gc_protect_expires_after = '250ms'", jobFeed.JobID()))
+					if initialProtect != "none" {
+						require.NoError(t, jobFeed.Pause())
+
+						// Wait a little bit and make sure the job ISN'T canceled.
+						require.ErrorContains(t, jobFeed.WaitDurationForState(10*time.Second, func(s jobs.State) bool {
+							return s == jobs.StateCanceled
+						}), `still waiting for job status; current status is "paused"`)
+
+						if initialProtect == "option" {
+							// Set the cluster setting back to something high to make sure the
+							// option alone can cause the changefeed to be canceled.
+							sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '24h'`)
+						}
+
+						// Set option to something small so that job will be canceled.
+						sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET gc_protect_expires_after = '1us'`, jobFeed.JobID()))
+					}
 
-		// Stale PTS record should trigger job cancellation.
-		require.NoError(t, jobFeed.WaitForState(func(s jobs.State) bool {
-			return s == jobs.StateCanceled
-		}))
+					// Stale PTS record should trigger job cancellation.
+					require.NoError(t, jobFeed.WaitForState(func(s jobs.State) bool {
+						return s == jobs.StateCanceled
+					}))
+				})
+		})
+
+		t.Run("canceled due to changefeed.protect_timestamp.max_age setting", func(t *testing.T) {
+			testutils.RunValues(t, "initially-protected-with", []string{"none", "option", "setting"},
+				func(t *testing.T, initialProtect string) {
+					defer func() {
+						sqlDB.Exec(t, `RESET CLUSTER SETTING changefeed.protect_timestamp.max_age`)
+					}()
+
+					if initialProtect == "setting" {
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '24h'`)
+					} else {
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '1us'`)
+					}
+
+					// Set the max age cluster setting to something small.
+					feedStmt := `CREATE CHANGEFEED FOR TABLE foo`
+					if initialProtect == "option" {
+						feedStmt += ` WITH gc_protect_expires_after='24h'`
+					}
+					feed, err := f.Feed(feedStmt)
+					require.NoError(t, err)
+					defer func() {
+						closeFeed(t, feed)
+					}()
+
+					jobFeed := feed.(cdctest.EnterpriseTestFeed)
+
+					if initialProtect != "none" {
+						require.NoError(t, jobFeed.Pause())
+
+						// Wait a little bit and make sure the job ISN'T canceled.
+						require.ErrorContains(t, jobFeed.WaitDurationForState(10*time.Second, func(s jobs.State) bool {
+							return s == jobs.StateCanceled
+						}), `still waiting for job status; current status is "paused"`)
+
+						switch initialProtect {
+						case "option":
+							// Reset the option so that it defaults to the cluster setting.
+							sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET gc_protect_expires_after = '0s'`, jobFeed.JobID()))
+						case "setting":
+							// Modify the cluster setting and do an ALTER CHANGEFEED so that
+							// the new value is picked up.
+							sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '1us'`)
+							sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET diff`, jobFeed.JobID()))
+						}
+					}
+
+					// Stale PTS record should trigger job cancellation.
+					require.NoError(t, jobFeed.WaitForState(func(s jobs.State) bool {
+						return s == jobs.StateCanceled
+					}))
+				})
+		})
 	}
 
 	cdcTestWithSystem(t, testFn, feedTestEnterpriseSinks)

pkg/ccl/changefeedccl/testfeed_test.go

@@ -491,7 +491,7 @@ func (f *jobFeed) WaitDurationForState(
 		if statusPred(jobs.State(status)) {
 			return nil
 		}
-		return errors.Newf("still waiting for job status; current %s", status)
+		return errors.Newf("still waiting for job status; current status is %q", status)
 	}, dur)
 }