changefeedccl: remove extra characters from topic names

Previously, when creating topic names, we would include quotes
around identifiers that included capital letters resulting in
some unexpected topic names. We removed this behavior from when
we render table names as strings.

This change is behind a flag: feature.changefeed.bare_table_names.
That flag is on by default.

Epic: none
Fixes: #148181

Release note (sql change): Fix a bug where extra quotes or escaped
quote characters would be added to topic names in changefeeds.
Can be turned off by setting feature.changefeed.bare_table_names to
false.

Author: aerfrei

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -1817,6 +1817,9 @@ func getQualifiedTableName(
 	if err != nil {
 		return "", err
 	}
+	if changefeedbase.UseBareTableNames.Get(&execCfg.Settings.SV) {
+		return tree.AsStringWithFlags(&tbName, tree.FmtBareIdentifiers), nil
+	}
 	return tbName.String(), nil
 }
 

pkg/ccl/changefeedccl/changefeed_test.go

@@ -788,6 +788,78 @@ func TestChangefeedBasicConfluentKafka(t *testing.T) {
 	cdcTest(t, testFn, feedTestForceSink("kafka"))
 }
 
+func TestChangefeedQuotedTableNameTopicName(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `CREATE TABLE "MyTable" (a INT PRIMARY KEY, b STRING)`)
+		sqlDB.Exec(t, `INSERT INTO "MyTable" VALUES (0, 'initial')`)
+		sqlDB.Exec(t, `UPSERT INTO "MyTable" VALUES (0, 'updated')`)
+
+		foo := feed(t, f,
+			`CREATE CHANGEFEED FOR d.public."MyTable" WITH diff, full_table_name`)
+		defer closeFeed(t, foo)
+
+		// The topic name should be d.public.MyTable and not d.public._u0022_MyTable_u0022_
+		// or d.public."MyTable".
+		assertPayloads(t, foo, []string{
+			`d.public.MyTable: [0]->{"after": {"a": 0, "b": "updated"}, "before": null}`,
+		})
+	}
+
+	cdcTest(t, testFn)
+}
+
+// TestChangefeedQuotedIdentifiersTopicName is similar to
+// TestChangefeedQuotedTableNameTopicName, but for quoted identifiers
+// in the SELECT clause instead of the table name.
+func TestChangefeedQuotedIdentifiersTopicName(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		sqlDB.Exec(t, `CREATE TABLE mytable (
+			id INT PRIMARY KEY, 
+			"SomeField" JSONB,
+			"AnotherField" JSONB
+		)`)
+
+		sqlDB.Exec(t, `INSERT INTO mytable VALUES (
+			1, 
+			'{"PropA": "value1", "prop_b": "value2"}'::jsonb,
+			'{"PropC": "value3", "prop_d": "value4"}'::jsonb
+		)`)
+
+		sqlDB.Exec(t, `INSERT INTO mytable VALUES (
+			2, 
+			'{"PropA": "value5", "prop_b": "value6"}'::jsonb,
+			'{"PropC": "value7", "prop_d": "value8"}'::jsonb
+		)`)
+
+		foo := feed(t, f, `CREATE CHANGEFEED WITH diff, full_table_name, on_error=pause, envelope=wrapped AS SELECT 
+			id,
+			"SomeField"->>'PropA' AS "PropA",
+			"SomeField"->>'prop_b' AS "PropB",
+			"AnotherField"->>'PropC' AS "PropC",
+			"AnotherField"->>'prop_d' AS "PropD"
+		FROM public.mytable`)
+		defer closeFeed(t, foo)
+
+		// The topic should show up as d.public.mytable and not as
+		// d.public.u0022_mytable_u0022 or d.public."MyTable".
+		assertPayloads(t, foo, []string{
+			`d.public.mytable: [1]->{"after": {"PropA": "value1", "PropB": "value2", "PropC": "value3", "PropD": "value4", "id": 1}, "before": null}`,
+			`d.public.mytable: [2]->{"after": {"PropA": "value5", "PropB": "value6", "PropC": "value7", "PropD": "value8", "id": 2}, "before": null}`,
+		})
+	}
+
+	cdcTest(t, testFn)
+}
+
 func TestChangefeedDiff(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -972,7 +1044,7 @@ func TestChangefeedFullTableName(t *testing.T) {
 		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'a')`)
 
 		t.Run(`envelope=row`, func(t *testing.T) {
-			foo := feed(t, f, `CREATE CHANGEFEED FOR foo WITH full_table_name`, optOutOfMetamorphicEnrichedEnvelope{reason: "broken for webhook; see #145927"})
+			foo := feed(t, f, `CREATE CHANGEFEED FOR foo WITH full_table_name`)
 			defer closeFeed(t, foo)
 			assertPayloads(t, foo, []string{`d.public.foo: [1]->{"after": {"a": 1, "b": "a"}}`})
 		})

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -380,3 +380,11 @@ var KafkaV2ErrorDetailsEnabled = settings.RegisterBoolSetting(
 	true,
 	settings.WithPublic,
 )
+
+// UseBareTableNames is used to enable and disable the use of bare table names
+// in changefeed topics.
+var UseBareTableNames = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"changefeed.bare_table_names.enabled",
+	"set to true to use bare table names in changefeed topics, false to use quoted table names; default is true",
+	true)

pkg/ccl/changefeedccl/helpers_test.go

@@ -1082,6 +1082,14 @@ func maybeForceEnrichedEnvelope(
 				return create, args, false, nil
 			}
 		}
+		if strings.EqualFold(opt.Key.String(), "full_table_name") {
+			// TODO(#145927): full_table_name is not supported in enriched envelopes.
+			switch f.(type) {
+			case *webhookFeedFactory:
+				t.Logf("did not force enriched envelope for %s because full_table_name was specified for webhook sink", create)
+				return create, args, false, nil
+			}
+		}
 		if strings.EqualFold(opt.Key.String(), "envelope") {
 			envelopeKV = &opt
 		}