changefeedccl: fix avro schema bug

Fix bug in avro schema generation introduced by #139655, which
caused extraneous fields to appear in the schema.

Epic: none
Release note: None

Author: asg0451

pkg/ccl/changefeedccl/avro/avro.go

@@ -180,9 +180,10 @@ type EnvelopeOpts struct {
 type EnvelopeRecord struct {
 	Record
 
-	// TODO: opts should be private
-	Opts               EnvelopeOpts
-	Before, After, Rec *DataRecord
+	// NOTE: this struct gets serialized to json, but we still need to be able
+	// to access these fields outside this package, so we hide them.
+	Opts               EnvelopeOpts `json:"-"`
+	Before, After, Rec *DataRecord  `json:"-"`
 }
 
 // typeToSchema converts a database type to an avro field

pkg/ccl/changefeedccl/encoder_test.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	gosql "database/sql"
 	"encoding/base64"
+	gojson "encoding/json"
 	"fmt"
 	"math/rand"
 	"net/url"
@@ -770,6 +771,41 @@ func TestAvroSchemaNamespace(t *testing.T) {
 	cdcTest(t, testFn, feedTestForceSink("kafka"), feedTestUseRootUserConnection)
 }
 
+func TestAvroSchemaHasExpectedTopLevelFields(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 DATABASE movr`)
+		sqlDB.Exec(t, `CREATE TABLE movr.drivers (id INT PRIMARY KEY, name STRING)`)
+		sqlDB.Exec(t,
+			`INSERT INTO movr.drivers VALUES (1, 'Alice')`,
+		)
+
+		foo := feed(t, f, fmt.Sprintf(`CREATE CHANGEFEED FOR movr.drivers `+
+			`WITH format=%s`, changefeedbase.OptFormatAvro))
+		defer closeFeed(t, foo)
+
+		assertPayloads(t, foo, []string{
+			`drivers: {"id":{"long":1}}->{"after":{"drivers":{"id":{"long":1},"name":{"string":"Alice"}}}}`,
+		})
+
+		reg := foo.(*kafkaFeed).registry
+
+		schemaJSON := reg.SchemaForSubject(`drivers-value`)
+		var schema map[string]any
+		require.NoError(t, gojson.Unmarshal([]byte(schemaJSON), &schema))
+		var keys []string
+		for k := range schema {
+			keys = append(keys, k)
+		}
+		require.ElementsMatch(t, keys, []string{"type", "name", "fields"})
+	}
+
+	cdcTest(t, testFn, feedTestForceSink("kafka"), feedTestUseRootUserConnection)
+}
+
 func TestTableNameCollision(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)