changefeedccl: validate avro_schema_prefix follows Avro naming rules

An invalid avro_schema_prefix caused changefeeds to hang indefinitely.
This validates that avro_schema_prefix follows Avro naming rules at
statement time.

Release note: Invalid avro_schema_prefix is now caught during statement
time. The prefix must start with [A-Za-z_] and subsequently contain only
[A-Za-z0-9_], as specified in the avro specification
https://avro.apache.org/docs/1.8.1/spec.html

Part of: #3513

Author: KeithCh

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"regexp"
 	"strings"
 	"time"
 
@@ -490,6 +491,22 @@ var redactSimple = func(string) (string, error) {
 	return "redacted", nil
 }
 
+// Regex from https://avro.apache.org/docs/1.8.1/spec.html.
+var avroNameRegexp = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
+
+func validateAvroSchemaPrefix(prefix string) error {
+	if prefix == "" {
+		return nil
+	}
+	if !avroNameRegexp.MatchString(prefix) {
+		return errors.Errorf(
+			`%s must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]`,
+			OptAvroSchemaPrefix,
+		)
+	}
+	return nil
+}
+
 // RedactUserFromURI takes a URI string and removes the user from it.
 // If there is no user, the original URI is returned.
 func RedactUserFromURI(uri string) (string, error) {

pkg/ccl/changefeedccl/changefeedbase/options_test.go

@@ -79,3 +79,53 @@ func TestEncodingOptionsValidations(t *testing.T) {
 	}
 
 }
+
+func TestAvroSchemaPrefixValidation(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	cases := []struct {
+		prefix    string
+		expectErr string
+	}{
+		// Valid prefixes
+		{"", ""},
+		{"a", ""},
+		{"A", ""},
+		{"_", ""},
+		{"abc", ""},
+		{"ABC", ""},
+		{"_abc", ""},
+		{"a1", ""},
+		{"A1_b2_C3", ""},
+		{"crdb_cdc_", ""},
+		{"super", ""},
+		{"____", ""},
+
+		// Invalid prefixes - starts with invalid character
+		{"1abc", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+		{"123", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+		{"-abc", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+
+		// Invalid prefixes - contains invalid characters
+		{"abc-def", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+		{"abc.def", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+		{"abc def", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+		{"abc!", "must start with [A-Za-z_] and subsequently contain only [A-Za-z0-9_]"},
+	}
+
+	for _, c := range cases {
+		t.Run(fmt.Sprintf("prefix=%q", c.prefix), func(t *testing.T) {
+			opts := MakeStatementOptions(map[string]string{
+				OptAvroSchemaPrefix: c.prefix,
+			})
+			err := opts.ValidateForCreateChangefeed(false)
+			if c.expectErr == "" {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), c.expectErr)
+			}
+		})
+	}
+}