changefeedccl: remove the default values from SHOW

CHANGEFEED JOB output

Currently, when a user alters a changefeed, we
include the default options in the SHOW CHANGEFEED
JOB output. In this PR we prevent the default values
from being displayed.

Release note (enterprise change): Remove the default
values from the SHOW CHANGEFEED JOB output

pkg/ccl/changefeedccl/BUILD.bazel

@@ -67,6 +67,7 @@ go_library(
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/flowinfra",
+        "//pkg/sql/parser",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/pgwire/pgnotice",

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/resolver"
+	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -78,7 +79,11 @@ func alterChangefeedPlanHook(
 
 		newChangefeedStmt := &tree.CreateChangefeed{}
 
-		newOptions, newSinkURI, err := generateNewOpts(ctx, p, alterChangefeedStmt.Cmds, prevDetails)
+		prevOpts, err := getPrevOpts(job.Payload().Description, prevDetails.Opts)
+		if err != nil {
+			return err
+		}
+		newOptions, newSinkURI, err := generateNewOpts(ctx, p, alterChangefeedStmt.Cmds, prevOpts, prevDetails.SinkURI)
 		if err != nil {
 			return err
 		}
@@ -191,22 +196,11 @@ func generateNewOpts(
 	ctx context.Context,
 	p sql.PlanHookState,
 	alterCmds tree.AlterChangefeedCmds,
-	details jobspb.ChangefeedDetails,
+	prevOpts map[string]string,
+	prevSinkURI string,
 ) (map[string]string, string, error) {
-	sinkURI := details.SinkURI
-	newOptions := make(map[string]string, len(details.Opts))
-
-	// pull the options that are set for the existing changefeed.
-	for key, value := range details.Opts {
-		// There are some options (e.g. topics) that we set during the creation of
-		// a changefeed, but we do not allow these options to be set by the user.
-		// Hence, we can not include these options in our new CREATE CHANGEFEED
-		// statement.
-		if _, ok := changefeedbase.ChangefeedOptionExpectValues[key]; !ok {
-			continue
-		}
-		newOptions[key] = value
-	}
+	sinkURI := prevSinkURI
+	newOptions := prevOpts
 
 	for _, cmd := range alterCmds {
 		switch v := cmd.(type) {
@@ -231,7 +225,7 @@ func generateNewOpts(
 						return nil, ``, err
 					}
 
-					prevSinkURI, err := url.Parse(details.SinkURI)
+					prevSinkURI, err := url.Parse(sinkURI)
 					if err != nil {
 						return nil, ``, err
 					}
@@ -652,3 +646,22 @@ func fetchSpansForDescs(
 
 	return spans, err
 }
+
+func getPrevOpts(prevDescription string, opts map[string]string) (map[string]string, error) {
+	prevStmt, err := parser.ParseOne(prevDescription)
+	if err != nil {
+		return nil, err
+	}
+
+	prevChangefeedStmt, ok := prevStmt.AST.(*tree.CreateChangefeed)
+	if !ok {
+		return nil, errors.Errorf(`could not parse job description`)
+	}
+
+	prevOpts := make(map[string]string, len(prevChangefeedStmt.Options))
+	for _, opt := range prevChangefeedStmt.Options {
+		prevOpts[opt.Key.String()] = opts[opt.Key.String()]
+	}
+
+	return prevOpts, nil
+}

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go

@@ -422,7 +422,7 @@ func TestShowChangefeedJobsAlterChangefeed(t *testing.T) {
 		out = obtainJobRowFn()
 
 		require.Equal(t, jobID, out.id, "Expected id:%d but found id:%d", jobID, out.id)
-		require.Equal(t, "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/' WITH envelope = 'wrapped', format = 'json', on_error = 'fail', schema_change_events = 'default', schema_change_policy = 'backfill', virtual_columns = 'omitted'", out.description, "Expected description:%s but found description:%s", "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/'", out.description)
+		require.Equal(t, "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/'", out.description, "Expected description:%s but found description:%s", "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/'", out.description)
 		require.Equal(t, sinkURI, out.SinkURI, "Expected sinkUri:%s but found sinkUri:%s", sinkURI, out.SinkURI)
 		require.Equal(t, "bar", out.topics, "Expected topics:%s but found topics:%s", "bar", sortedTopics)
 		require.Equal(t, "{d.public.bar}", string(out.FullTableNames), "Expected fullTableNames:%s but found fullTableNames:%s", "{d.public.bar}", string(out.FullTableNames))
@@ -433,7 +433,7 @@ func TestShowChangefeedJobsAlterChangefeed(t *testing.T) {
 		out = obtainJobRowFn()
 
 		require.Equal(t, jobID, out.id, "Expected id:%d but found id:%d", jobID, out.id)
-		require.Equal(t, "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/' WITH envelope = 'wrapped', format = 'json', on_error = 'fail', resolved = '5s', schema_change_events = 'default', schema_change_policy = 'backfill', virtual_columns = 'omitted'", out.description, "Expected description:%s but found description:%s", "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/'", out.description)
+		require.Equal(t, "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/' WITH resolved = '5s'", out.description, "Expected description:%s but found description:%s", "CREATE CHANGEFEED FOR TABLE bar INTO 'kafka://does.not.matter/ WITH resolved = '5s''", out.description)
 		require.Equal(t, sinkURI, out.SinkURI, "Expected sinkUri:%s but found sinkUri:%s", sinkURI, out.SinkURI)
 		require.Equal(t, "bar", out.topics, "Expected topics:%s but found topics:%s", "bar", sortedTopics)
 		require.Equal(t, "{d.public.bar}", string(out.FullTableNames), "Expected fullTableNames:%s but found fullTableNames:%s", "{d.public.bar}", string(out.FullTableNames))