Merge cockroachdb#107923

107923: changefeedccl: Correctly handle dropped UDTs r=miretskiy a=miretskiy

Schema feed is responsible for the determination
of correct schema descriptor corresponding to an
event occuring at certain time.

Due to the need to use low level API to retrieve
historical descriptor versions, it is possible that the schema feed would observe descriptors that were dropped/deleted/truncated.

A long outstanding bug in schema feed would
incorrectly check for low level value being 'nil' in order to determine if the descriptor was dropped.
This check is incorrect since the iterator may return a non-nil empty byte array reprsenting a tombstone.
A non-nil, but empty KV value would lead schema feed to incorrectly attempt to deserialize protocol message, resulting in a `value type is not BYTES: UNKNOWN` being returned.

This PR fixes the above issue, and adds a test
to verify correct behavior.

Fixes https://github.com/cockroachlabs/support/issues/2408

Release note (enterprise change): Fix a rare changefeed issue which is triggered when the parent database or types are dropped, and instead of exiting with a descriptive error message, the changefeed would observe an opaque error instead (`value type is not BYTES: UNKNOWN`)

Co-authored-by: Yevgeniy Miretskiy <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/schemafeed/schema_feed.go

@@ -857,12 +857,19 @@ func (tf *schemaFeed) fetchDescriptorVersions(
 					if err != nil {
 						return err
 					}
-					if unsafeValue == nil {
+
+					if len(unsafeValue) == 0 {
+						if isType {
+							return changefeedbase.WithTerminalError(
+								errors.Wrapf(catalog.ErrDescriptorDropped, "type descriptor %d dropped", id))
+						}
+
 						name := origName
 						if name == "" {
 							name = changefeedbase.StatementTimeName(fmt.Sprintf("desc(%d)", id))
 						}
-						return errors.Errorf(`"%v" was dropped or truncated`, name)
+						return changefeedbase.WithTerminalError(
+							errors.Wrapf(catalog.ErrDescriptorDropped, `table "%v"[%d] was dropped or truncated`, name, id))
 					}
 
 					// Unmarshal the descriptor.

pkg/ccl/changefeedccl/schemafeed/schema_feed_test.go

@@ -262,3 +262,57 @@ func TestFetchDescriptorVersionsCPULimiterPagination(t *testing.T) {
 	require.Len(t, desc, 2)
 	require.Equal(t, 2, numRequests)
 }
+
+func TestSchemaFeedHandlesCascadeDatabaseDrop(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	srv, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
+	sqlServer := s.SQLServer().(*sql.Server)
+
+	sqlDB := sqlutils.MakeSQLRunner(db)
+
+	beforeCreate := s.Clock().Now()
+
+	// Create a database, containing user defined type along with table using that type.
+	sqlDB.ExecMultiple(t,
+		`CREATE DATABASE test`,
+		`USE test`,
+		`CREATE TYPE status AS ENUM ('open', 'closed', 'inactive')`,
+		`CREATE TABLE foo(a INT, t status DEFAULT 'open')`,
+		`USE defaultdb`,
+	)
+
+	var targets changefeedbase.Targets
+	var tableID descpb.ID
+	sqlDB.QueryRow(t, "SELECT 'test.foo'::regclass::int").Scan(&tableID)
+	targets.Add(changefeedbase.Target{
+		Type:              jobspb.ChangefeedTargetSpecification_PRIMARY_FAMILY_ONLY,
+		TableID:           tableID,
+		FamilyName:        "primary",
+		StatementTimeName: "foo",
+	})
+	sf := New(ctx, &sqlServer.GetExecutorConfig().DistSQLSrv.ServerConfig,
+		TestingAllEventFilter, targets, s.Clock().Now(), nil, changefeedbase.CanHandle{
+			MultipleColumnFamilies: true,
+			VirtualColumns:         true,
+		}).(*schemaFeed)
+
+	// initialize type dependencies in schema feed.
+	require.NoError(t, sf.primeInitialTableDescs(ctx))
+
+	// DROP database with cascade to cause the type along with the table to be dropped.
+	// Dropped tables are marked as being dropped (i.e. there is an MVCC version of the
+	// descriptor that has a state indicating that the table is being dropped).
+	// However, dependent UDTs are simply deleted so, there is an MVCC tombstone for that type.
+	sqlDB.Exec(t, `DROP DATABASE test CASCADE;`)
+
+	// Fetching descriptor versions from before the initial create statement
+	// up until the current time should result in a catalog.ErrDescriptorDropped error.
+	_, err := sf.fetchDescriptorVersions(ctx, beforeCreate, s.Clock().Now())
+	require.True(t, errors.Is(err, catalog.ErrDescriptorDropped),
+		"expected dropped descriptor error, found: %v", err)
+}