Merge #77043

77043: changefeedccl: allow users to alter the sink URI of an existing changefeed r=sherman-grewal a=sherman-grewal changefeedccl: allow users to alter the sink URI of an existing changefeed References #75895 In this PR, we introduce the capability to alter the sink URI of an existing changefeed. This can be achieved by executing the following statement: ALTER CHANGEFEED <job_id> SET sink = '<sink_uri>' Note that the sink type cannot be altered. That is, the sink type must be the same type that was chosen when the changefeed was initially created. Release note (enterprise change): Users may now alter the sink URI of an existing changefeed. This can be achieved by executing the following statement: ALTER CHANGEFEED <job_id> SET sink = '<sink_uri>' Where the sink type of the new sink must match the sink type of the old sink that was chosen at the creation of the changefeed. Co-authored-by: Sherman Grewal <[email protected]>
cockroachdb · Feb 26, 2022 · c7b4a53 · c7b4a53
2 parents 4077017 + 067f213
commit c7b4a53
Show file tree

Hide file tree

Showing 3 changed files with 116 additions and 8 deletions.
diff --git a/pkg/ccl/changefeedccl/alter_changefeed_stmt.go b/pkg/ccl/changefeedccl/alter_changefeed_stmt.go
@@ -10,6 +10,7 @@ package changefeedccl
 
 import (
 	"context"
+	"net/url"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backupresolver"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
@@ -163,7 +164,7 @@ func alterChangefeedPlanHook(
 					delete(newDescs, desc.GetID())
 				}
 			case *tree.AlterChangefeedSetOptions:
-				optsFn, err := p.TypeAsStringOpts(ctx, v.Options, changefeedbase.ChangefeedOptionExpectValues)
+				optsFn, err := p.TypeAsStringOpts(ctx, v.Options, changefeedbase.AlterChangefeedOptionExpectValues)
 				if err != nil {
 					return err
 				}
@@ -174,21 +175,44 @@ func alterChangefeedPlanHook(
 				}
 
 				for key, value := range opts {
-					if _, ok := changefeedbase.ChangefeedOptionExpectValues[key]; !ok {
-						return pgerror.Newf(pgcode.InvalidParameterValue, `invalid option %q`, key)
-					}
 					if _, ok := changefeedbase.AlterChangefeedUnsupportedOptions[key]; ok {
 						return pgerror.Newf(pgcode.InvalidParameterValue, `cannot alter option %q`, key)
 					}
-					opt := tree.KVOption{Key: tree.Name(key)}
-					if len(value) > 0 {
-						opt.Value = tree.NewDString(value)
+					if key == changefeedbase.OptSink {
+						newSinkURI, err := url.Parse(value)
+						if err != nil {
+							return err
+						}
+
+						prevSinkURI, err := url.Parse(prevDetails.SinkURI)
+						if err != nil {
+							return err
+						}
+
+						if newSinkURI.Scheme != prevSinkURI.Scheme {
+							return pgerror.Newf(
+								pgcode.InvalidParameterValue,
+								`new sink type %q does not match original sink type %q, sink type cannot be altered`,
+								newSinkURI.Scheme,
+								prevSinkURI.Scheme,
+							)
+						}
+
+						newChangefeedStmt.SinkURI = tree.NewDString(value)
+					} else {
+						opt := tree.KVOption{Key: tree.Name(key)}
+						if len(value) > 0 {
+							opt.Value = tree.NewDString(value)
+						}
+						optionsMap[key] = opt
 					}
-					optionsMap[key] = opt
 				}
 			case *tree.AlterChangefeedUnsetOptions:
 				optKeys := v.Options.ToStrings()
 				for _, key := range optKeys {
+					if key == changefeedbase.OptSink {
+						return pgerror.Newf(pgcode.InvalidParameterValue, `cannot unset option %q`, key)
+					}
 					if _, ok := changefeedbase.ChangefeedOptionExpectValues[key]; !ok {
 						return pgerror.Newf(pgcode.InvalidParameterValue, `invalid option %q`, key)
 					}

diff --git a/pkg/ccl/changefeedccl/alter_changefeed_test.go b/pkg/ccl/changefeedccl/alter_changefeed_test.go
@@ -218,6 +218,11 @@ func TestAlterChangefeedErrors(t *testing.T) {
 			`pq: cannot alter option "initial_scan"`,
 			fmt.Sprintf(`ALTER CHANGEFEED %d UNSET initial_scan`, feed.JobID()),
 		)
+
+		sqlDB.ExpectErr(t,
+			`cannot unset option "sink"`,
+			fmt.Sprintf(`ALTER CHANGEFEED %d UNSET sink`, feed.JobID()),
+		)
 	}
 
 	t.Run(`kafka`, kafkaTest(testFn))
@@ -304,3 +309,67 @@ func TestAlterChangefeedPersistSinkURI(t *testing.T) {
 
 	require.Equal(t, details.SinkURI, `s3://fake-bucket-name/fake/path?AWS_ACCESS_KEY_ID=123&AWS_SECRET_ACCESS_KEY=456`)
 }
+
+func TestAlterChangefeedChangeSinkTypeError(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, db *gosql.DB, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(db)
+		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+
+		testFeed := feed(t, f, `CREATE CHANGEFEED FOR foo`)
+		defer closeFeed(t, testFeed)
+
+		feed, ok := testFeed.(cdctest.EnterpriseTestFeed)
+		require.True(t, ok)
+
+		sqlDB.Exec(t, `PAUSE JOB $1`, feed.JobID())
+		waitForJobStatus(sqlDB, t, feed.JobID(), `paused`)
+
+		sqlDB.ExpectErr(t,
+			`pq: new sink type "s3" does not match original sink type "kafka", sink type cannot be altered`,
+			fmt.Sprintf(`ALTER CHANGEFEED %d SET sink = 's3://fake-bucket-name/fake/path?AWS_ACCESS_KEY_ID=123&AWS_SECRET_ACCESS_KEY=456'`, feed.JobID()),
+		)
+	}
+
+	t.Run(`kafka`, kafkaTest(testFn))
+}
+
+func TestAlterChangefeedChangeSinkURI(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, db *gosql.DB, f cdctest.TestFeedFactory) {
+		registry := f.Server().JobRegistry().(*jobs.Registry)
+		ctx := context.Background()
+
+		sqlDB := sqlutils.MakeSQLRunner(db)
+		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+
+		testFeed := feed(t, f, `CREATE CHANGEFEED FOR foo`)
+		defer closeFeed(t, testFeed)
+
+		feed, ok := testFeed.(cdctest.EnterpriseTestFeed)
+		require.True(t, ok)
+
+		sqlDB.Exec(t, `PAUSE JOB $1`, feed.JobID())
+		waitForJobStatus(sqlDB, t, feed.JobID(), `paused`)
+
+		newSinkURI := `kafka://new_kafka_uri`
+
+		sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET sink = '%s'`, feed.JobID(), newSinkURI))
+
+		sqlDB.Exec(t, fmt.Sprintf(`RESUME JOB %d`, feed.JobID()))
+		waitForJobStatus(sqlDB, t, feed.JobID(), `running`)
+
+		job, err := registry.LoadJob(ctx, feed.JobID())
+		require.NoError(t, err)
+		details, ok := job.Details().(jobspb.ChangefeedDetails)
+		require.True(t, ok)
+
+		require.Equal(t, newSinkURI, details.SinkURI)
+	}
+
+	t.Run(`kafka`, kafkaTest(testFn))
+}
diff --git a/pkg/ccl/changefeedccl/changefeedbase/options.go b/pkg/ccl/changefeedccl/changefeedbase/options.go
@@ -118,6 +118,10 @@ const (
 	OptKafkaSinkConfig   = `kafka_sink_config`
 	OptWebhookSinkConfig = `webhook_sink_config`
 
+	// OptSink allows users to alter the Sink URI of an existing changefeed.
+	// Note that this option is only allowed for alter changefeed statements.
+	OptSink = `sink`
+
 	SinkParamCACert                 = `ca_cert`
 	SinkParamClientCert             = `client_cert`
 	SinkParamClientKey              = `client_key`
@@ -237,3 +241,14 @@ var NoLongerExperimental = map[string]string{
 // AlterChangefeedUnsupportedOptions are changefeed options that we do not allow
 // users to alter
 var AlterChangefeedUnsupportedOptions = makeStringSet(OptCursor, OptInitialScan, OptNoInitialScan)
+
+// AlterChangefeedOptionExpectValues is used to parse alter changefeed options
+// using PlanHookState.TypeAsStringOpts().
+var AlterChangefeedOptionExpectValues = func() map[string]sql.KVStringOptValidate {
+	alterChangefeedOptions := make(map[string]sql.KVStringOptValidate, len(ChangefeedOptionExpectValues)+1)
+	for key, value := range ChangefeedOptionExpectValues {
+		alterChangefeedOptions[key] = value
+	}
+	alterChangefeedOptions[OptSink] = sql.KVStringOptRequireValue
+	return alterChangefeedOptions
+}()