schemastore: fix view DDL handling and metadata cleanup (#6568)

logservice/schemastore/persist_storage_ddl_handlers.go

@@ -324,8 +324,8 @@ var allDDLHandlers = map[model.ActionType]*persistStorageDDLHandler{
 	model.ActionCreateView: {
 		buildPersistedDDLEventFunc: buildPersistedDDLEventForCreateView,
 		updateDDLHistoryFunc:       updateDDLHistoryForCreateView,
-		updateFullTableInfoFunc:    updateFullTableInfoIgnore,
-		updateSchemaMetadataFunc:   updateSchemaMetadataIgnore,
+		updateFullTableInfoFunc:    updateFullTableInfoForCreateView,
+		updateSchemaMetadataFunc:   updateSchemaMetadataForCreateView,
 		iterateEventTablesFunc:     iterateEventTablesIgnore,
 		extractTableInfoFunc:       extractTableInfoFuncIgnore,
 		buildDDLEventFunc:          buildDDLEventForCreateView,
@@ -351,8 +351,8 @@ var allDDLHandlers = map[model.ActionType]*persistStorageDDLHandler{
 	model.ActionDropView: {
 		buildPersistedDDLEventFunc: buildPersistedDDLEventForDropView,
 		updateDDLHistoryFunc:       updateDDLHistoryForTableTriggerOnlyDDL,
-		updateFullTableInfoFunc:    updateFullTableInfoIgnore,
-		updateSchemaMetadataFunc:   updateSchemaMetadataIgnore,
+		updateFullTableInfoFunc:    updateFullTableInfoForDropTable,
+		updateSchemaMetadataFunc:   updateSchemaMetadataForDropTable,
 		iterateEventTablesFunc:     iterateEventTablesIgnore,
 		extractTableInfoFunc:       extractTableInfoFuncIgnore,
 		buildDDLEventFunc:          buildDDLEventForDropView,
@@ -716,13 +716,18 @@ func buildPersistedDDLEventForCreateView(args buildPersistedDDLEventFuncArgs) (P
 	event := buildPersistedDDLEventCommon(args)
 	event.SchemaName = getSchemaName(args.databaseMap, event.SchemaID)
 	event.TableName = args.job.TableName
+	// CREATE OR REPLACE VIEW invalidates the old ID. Snapshot views may still
+	// have catalog entries even though incremental views are not registered.
+	if oldID, ok := findTableIDByName(args.tableMap, event.SchemaID, event.TableName); ok {
+		event.ExtraTableID = oldID
+	}
 	return event, nil
 }
 
 func buildPersistedDDLEventForDropView(args buildPersistedDDLEventFuncArgs) (PersistedDDLEvent, error) {
 	event := buildPersistedDDLEventCommon(args)
 	event.SchemaName = getSchemaName(args.databaseMap, event.SchemaID)
-	// We don't store the relationship: view_id -> table_name, get table name from args.job
+	// Incremental views are absent from tableMap; the job carries their name.
 	event.TableName = args.job.TableName
 	// The query in job maybe "DROP VIEW test1.view1, test2.view2", we need rebuild it here.
 	event.Query = fmt.Sprintf("DROP VIEW %s", common.QuoteSchema(event.SchemaName, event.TableName))
@@ -1039,6 +1044,9 @@ func buildPersistedDDLEventForRenameTable(args buildPersistedDDLEventFuncArgs) (
 	// through SQL. For malformed or unsupported jobs, fall back to the complete snapshot
 	// identity instead of combining fields from different sources.
 	if oldSchemaID == 0 || oldSchemaName == "" || oldTableName == "" {
+		if event.TableInfo.IsView() && args.tableMap[event.TableID] == nil {
+			return PersistedDDLEvent{}, cerror.ErrDDLEventError.GenWithStackByArgs()
+		}
 		oldSchemaID = getSchemaID(args.tableMap, event.TableID)
 		oldSchemaName = getSchemaName(args.databaseMap, oldSchemaID)
 		oldTableName = getTableName(args.tableMap, event.TableID)
@@ -1375,6 +1383,9 @@ func updateDDLHistoryForSchemaDDL(args updateDDLHistoryFuncArgs) []uint64 {
 
 func updateDDLHistoryForAddDropTable(args updateDDLHistoryFuncArgs) []uint64 {
 	args.appendTableTriggerDDLHistory(args.ddlEvent.FinishedTs)
+	if args.ddlEvent.TableInfo != nil && args.ddlEvent.TableInfo.IsView() {
+		return args.tableTriggerDDLHistory
+	}
 	// Note: for create table, this ddl event will not be sent to table dispatchers by default.
 	// adding it to ddl history is just for building table info store.
 	if isPartitionTable(args.ddlEvent.TableInfo) {
@@ -1508,7 +1519,7 @@ func updateDDLHistoryForRenameTables(args updateDDLHistoryFuncArgs) []uint64 {
 	args.appendTableTriggerDDLHistory(args.ddlEvent.FinishedTs)
 	// it won't be send to table dispatchers, just for build version store
 	for _, info := range args.ddlEvent.MultipleTableInfos {
-		if info.ID == InvalidTableID {
+		if info.ID == InvalidTableID || info.IsView() {
 			continue
 		}
 		if isPartitionTable(info) {
@@ -1573,6 +1584,11 @@ func updateDDLHistoryForAlterTableTTL(args updateDDLHistoryFuncArgs) []uint64 {
 
 func updateFullTableInfoIgnore(args updateFullTableInfoFuncArgs) {}
 
+func updateFullTableInfoForCreateView(args updateFullTableInfoFuncArgs) {
+	// View definitions are carried by DDL events. Remove the replaced snapshot entry.
+	delete(args.tableInfoMap, args.event.ExtraTableID)
+}
+
 func updateFullTableInfoForDropSchema(args updateFullTableInfoFuncArgs) {
 	for tableID := range args.databaseMap[args.event.SchemaID].Tables {
 		delete(args.tableInfoMap, tableID)
@@ -1671,6 +1687,13 @@ func updateSchemaMetadataForDropTable(args updateSchemaMetadataFuncArgs) {
 
 func updateSchemaMetadataIgnore(args updateSchemaMetadataFuncArgs) {}
 
+func updateSchemaMetadataForCreateView(args updateSchemaMetadataFuncArgs) {
+	if args.event.ExtraTableID != 0 {
+		args.removeTableFromDB(args.event.ExtraTableID, args.event.SchemaID)
+		delete(args.tableMap, args.event.ExtraTableID)
+	}
+}
+
 func updateSchemaMetadataForTruncateTable(args updateSchemaMetadataFuncArgs) {
 	oldTableID := args.event.TableID
 	newTableID := args.event.ExtraTableID
@@ -1691,6 +1714,11 @@ func updateSchemaMetadataForTruncateTable(args updateSchemaMetadataFuncArgs) {
 }
 
 func updateSchemaMetadataForRenameTable(args updateSchemaMetadataFuncArgs) {
+	// Incrementally created views are absent from tableMap. Views loaded from
+	// a snapshot still need their existing metadata updated.
+	if args.event.TableInfo != nil && args.event.TableInfo.IsView() && args.tableMap[args.event.TableID] == nil {
+		return
+	}
 	if filter.IsSysSchema(args.event.ExtraSchemaName) {
 		// The old table was filtered out. Register its metadata in the new schema.
 		updateSchemaMetadataForNewTableDDL(args)
@@ -1763,6 +1791,9 @@ func updateSchemaMetadataForRenameTables(args updateSchemaMetadataFuncArgs) {
 		if info.ID == InvalidTableID {
 			continue
 		}
+		if info.IsView() && args.tableMap[info.ID] == nil {
+			continue
+		}
 		if filter.IsSysSchema(args.event.ExtraSchemaNames[i]) {
 			newTableArgs := args
 			newTableArgs.event = &PersistedDDLEvent{
@@ -1872,6 +1903,9 @@ func iterateEventTablesForDropSchema(args iterateEventTablesFuncArgs) {
 
 func iterateEventTablesForSingleTableDDL(args iterateEventTablesFuncArgs) {
 	event, apply := args.event, args.apply
+	if event.TableInfo != nil && event.TableInfo.IsView() {
+		return
+	}
 	if isPartitionTable(event.TableInfo) {
 		apply(getAllPartitionIDs(event.TableInfo)...)
 	} else {
@@ -1922,7 +1956,7 @@ func iterateEventTablesForExchangeTablePartition(args iterateEventTablesFuncArgs
 func iterateEventTablesForRenameTables(args iterateEventTablesFuncArgs) {
 	event, apply := args.event, args.apply
 	for _, info := range event.MultipleTableInfos {
-		if info.ID == InvalidTableID {
+		if info.ID == InvalidTableID || info.IsView() {
 			continue
 		}
 		if isPartitionTable(info) {
@@ -1936,6 +1970,9 @@ func iterateEventTablesForRenameTables(args iterateEventTablesFuncArgs) {
 func iterateEventTablesForCreateTables(args iterateEventTablesFuncArgs) {
 	event, apply := args.event, args.apply
 	for _, info := range event.MultipleTableInfos {
+		if info.IsView() {
+			continue
+		}
 		if isPartitionTable(info) {
 			apply(getAllPartitionIDs(info)...)
 		} else {
@@ -2356,6 +2393,12 @@ func buildDDLEventForRecoverSchema(rawEvent *PersistedDDLEvent, tableFilter filt
 		if filtered {
 			continue
 		}
+		if tableInfo.IsView() {
+			if !notSync {
+				ddlEvent.MultipleTableInfos = append(ddlEvent.MultipleTableInfos, common.WrapTableInfo(rawEvent.SchemaName, tableInfo))
+			}
+			continue
+		}
 		if isPartitionTable(tableInfo) {
 			for _, partitionID := range getAllPartitionIDs(tableInfo) {
 				ddlEvent.NeedAddedTables = append(ddlEvent.NeedAddedTables, commonEvent.Table{
@@ -2743,6 +2786,19 @@ func buildDDLEventForRenameTable(rawEvent *PersistedDDLEvent, tableFilter filter
 			return commonEvent.DDLEvent{}, false, err
 		}
 	}
+	if rawEvent.TableInfo != nil && rawEvent.TableInfo.IsView() {
+		// Views have no physical dispatcher. Only the table trigger participates
+		// in the barrier, as it does for DROP VIEW.
+		ddlEvent.BlockedTables = &commonEvent.InfluencedTables{
+			InfluenceType: commonEvent.InfluenceTypeNormal,
+			TableIDs:      []int64{common.DDLSpanTableID},
+		}
+		if ignorePrevTable {
+			ddlEvent.Err = cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(rawEvent.TableID, rawEvent.Query).Error()
+		}
+		ddlEvent.NotSync = notSyncPrevTable
+		return ddlEvent, true, nil
+	}
 	if isPartitionTable(rawEvent.TableInfo) {
 		allPhysicalIDs := getAllPartitionIDs(rawEvent.TableInfo)
 		if !ignorePrevTable {
@@ -3135,7 +3191,7 @@ func buildDDLEventForRenameTables(rawEvent *PersistedDDLEvent, tableFilter filte
 	if err != nil {
 		log.Panic("split queries failed", zap.Error(err))
 	}
-	var addNames, dropNames []commonEvent.SchemaTableName
+	var addNames, dropNames, notSyncDropNames []commonEvent.SchemaTableName
 	allFiltered := true
 	resultQuerys := make([]string, 0)
 	tableInfos := make([]*common.TableInfo, 0)
@@ -3162,6 +3218,21 @@ func buildDDLEventForRenameTables(rawEvent *PersistedDDLEvent, tableFilter filte
 			continue
 		}
 		allFiltered = false
+		if tableInfo != nil && tableInfo.IsView() {
+			if ignorePrevTable {
+				ddlEvent.Err = cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(tableInfo.ID, rawEvent.Query).Error()
+			} else if !notSyncPrevTable {
+				resultQuerys = append(resultQuerys, querys[i])
+				tableInfos = append(tableInfos, common.WrapTableInfo(rawEvent.SchemaNames[i], tableInfo))
+				// GetEvents uses DropName to recover the source identity of each
+				// emitted rename. Do not add the view to the physical table names.
+				dropNames = append(dropNames, commonEvent.SchemaTableName{
+					SchemaName: rawEvent.ExtraSchemaNames[i],
+					TableName:  rawEvent.ExtraTableNames[i],
+				})
+			}
+			continue
+		}
 		if isPartitionTable(tableInfo) {
 			allPhysicalIDs := getAllPartitionIDs(tableInfo)
 			if !ignorePrevTable {
@@ -3293,10 +3364,17 @@ func buildDDLEventForRenameTables(rawEvent *PersistedDDLEvent, tableFilter filte
 				}
 			}
 		}
+		if !ignorePrevTable && notSyncPrevTable {
+			// Keep the first DropName entries aligned with the emitted queries.
+			// Ignored SQL still carries the table's name lifecycle change.
+			notSyncDropNames = append(notSyncDropNames, dropNames[len(dropNames)-1])
+			dropNames = dropNames[:len(dropNames)-1]
+		}
 	}
 	if allFiltered {
 		return commonEvent.DDLEvent{}, false, err
 	}
+	dropNames = append(dropNames, notSyncDropNames...)
 	if addNames != nil || dropNames != nil {
 		ddlEvent.TableNameChange = &commonEvent.TableNameChange{
 			AddName:  addNames,

logservice/schemastore/persist_storage_test.go

@@ -5022,6 +5022,66 @@ func TestRegisteredTableInfoForComplexDDL(t *testing.T) {
 }
 
 func TestBuildSchemaAndViewDDLEvents(t *testing.T) {
+	t.Run("drop snapshot view and reuse name", func(t *testing.T) {
+		view := newEligibleTableInfoForTest(500, "v")
+		view.View = &model.ViewInfo{}
+		storage := newPersistentStorageForTest(t.TempDir(), []mockDBInfo{{
+			dbInfo: &model.DBInfo{ID: 10, Name: parser_model.NewCIStr("test")},
+			tables: []*model.TableInfo{view},
+		}})
+		t.Cleanup(func() { require.NoError(t, storage.close()) })
+		drop := buildDropViewJobForTest(10, 1000)
+		drop.TableID, drop.TableName = view.ID, "v"
+		require.NoError(t, storage.handleDDLJob(drop))
+		require.NotContains(t, storage.tableMap, view.ID)
+		require.NotContains(t, storage.databaseMap[10].Tables, view.ID)
+		require.NoError(t, storage.handleDDLJob(buildCreateTableJobForTest(10, 600, "v", 1100)))
+		job := buildCreateTableJobForTest(10, 700, "copy", 1200)
+		job.Query = "CREATE TABLE `test`.`copy` LIKE `test`.`v`"
+		event, err := buildPersistedDDLEventForCreateTable(buildPersistedDDLEventFuncArgs{
+			job: job, databaseMap: storage.databaseMap, tableMap: storage.tableMap,
+		})
+		require.NoError(t, err)
+		require.Equal(t, int64(600), event.ExtraTableID)
+	})
+	t.Run("replace snapshot view", func(t *testing.T) {
+		helper := commonEvent.NewEventTestHelper(t)
+		t.Cleanup(helper.Close)
+		helper.DDL2Job("CREATE DATABASE view_lifecycle")
+		create := helper.DDL2Job("CREATE VIEW view_lifecycle.v AS SELECT 1 AS id")
+		replace := helper.DDL2Job("CREATE OR REPLACE VIEW view_lifecycle.v AS SELECT 2 AS id")
+		path := t.TempDir()
+		storage := newPersistentStorageForTest(path, []mockDBInfo{{
+			dbInfo: &model.DBInfo{ID: create.SchemaID, Name: parser_model.NewCIStr("view_lifecycle")},
+			tables: []*model.TableInfo{create.BinlogInfo.TableInfo},
+		}})
+		t.Cleanup(func() { require.NoError(t, storage.close()) })
+		storage.kvStorage = helper.Storage()
+		require.NoError(t, storage.handleDDLJob(replace))
+		require.NotContains(t, storage.tableMap, create.TableID)
+		require.NotContains(t, storage.databaseMap[create.SchemaID].Tables, create.TableID)
+		require.NoError(t, storage.close())
+		storage = loadPersistentStorageFromPathForTest(path, replace.BinlogInfo.FinishedTS+1)
+		require.NotContains(t, storage.tableMap, create.TableID)
+		assertPhysicalTableIDs(t, storage, replace.BinlogInfo.FinishedTS, nil)
+	})
+	t.Run("recover schema with view", func(t *testing.T) {
+		view := newEligibleTableInfoForTest(500, "v")
+		view.View = &model.ViewInfo{}
+		storage := newPersistentStorageForTest(t.TempDir(), nil)
+		t.Cleanup(func() { require.NoError(t, storage.close()) })
+		job := buildRecoverSchemaJobForTest(10, "test", []*model.TableInfo{
+			newEligibleTableInfoForTest(200, "t"), view,
+		}, 1000)
+		require.NoError(t, storage.handleDDLJob(job))
+		events, err := storage.fetchTableTriggerDDLEvents(nil, 999, 10)
+		require.NoError(t, err)
+		require.Len(t, events, 1)
+		require.Len(t, events[0].MultipleTableInfos, 2)
+		require.Equal(t, []commonEvent.Table{{SchemaID: 10, TableID: 200, Splitable: true}}, events[0].NeedAddedTables)
+		require.Equal(t, []commonEvent.SchemaTableName{{SchemaName: "test", TableName: "t"}}, events[0].TableNameChange.AddName)
+		assertPhysicalTableIDs(t, storage, 1000, []int64{200})
+	})
 	testCases := []struct {
 		name                 string
 		raw                  PersistedDDLEvent

tests/integration_tests/multi_tables_ddl/conf/normal.toml

@@ -15,7 +15,13 @@ rules = [
     "multi_tables_ddl_test.rename_mix_normal_2_done",
     "multi_tables_ddl_test.rename_mix_part_2_done",
     "multi_tables_ddl_test.finish_mark",
+    "view_ddl_test.*",
+    "view_recover_test.*",
 ]
 
+[[filter.event-filters]]
+matcher = ["view_ddl_test.rename_view_ignored_table_*", "view_ddl_test.rename_view_ignored_view_*"]
+ignore-event = ["rename table"]
+
 [sink.csv]
 include-commit-ts = true

tests/integration_tests/multi_tables_ddl/data/test.sql

@@ -177,3 +177,57 @@ rename table t13 to t14;
 /* error, t11 not match `filter.rule` and t9 match `filter.rule` */
 rename table t11 to t9;
 create table finish_mark(id int primary key);
+
+USE view_ddl_test;
+-- Regression for #6548: excluded views still pass through the shared SchemaStore.
+create database unselected_db;
+create view unselected_db.conflict_view as select 1 as id;
+create or replace view unselected_db.conflict_view as select 2 as id;
+rename table unselected_db.conflict_view to unselected_db.conflict_view_done;
+drop view unselected_db.conflict_view_done;
+drop database unselected_db;
+
+create table finish_mark(id int primary key);
+create table rename_view_base(id int primary key);
+insert into rename_view_base values(7);
+create view rename_view_single as select id from rename_view_base;
+rename table rename_view_single to rename_view_single_done;
+
+-- Mixed table/view renames must work in either order.
+create table rename_view_table_1(id int primary key);
+insert into rename_view_table_1 values(1);
+create view rename_view_mixed_1 as select id from rename_view_base;
+rename table rename_view_table_1 to rename_view_table_1_done, rename_view_mixed_1 to rename_view_mixed_1_done;
+insert into rename_view_table_1_done values(2);
+
+create table rename_view_table_2(id int primary key);
+insert into rename_view_table_2 values(1);
+create view rename_view_mixed_2 as select id from rename_view_base;
+rename table rename_view_mixed_2 to rename_view_mixed_2_done, rename_view_table_2 to rename_view_table_2_done;
+insert into rename_view_table_2_done values(2);
+
+-- Ignoring the table rename must preserve the emitted view rename metadata.
+create table rename_view_ignored_table_3(id int primary key);
+insert into rename_view_ignored_table_3 values(1);
+create view rename_view_mixed_3 as select id from rename_view_base;
+rename table rename_view_ignored_table_3 to rename_view_ignored_table_3_done, rename_view_mixed_3 to rename_view_mixed_3_done;
+
+create table rename_view_ignored_table_4(id int primary key);
+insert into rename_view_ignored_table_4 values(1);
+create view rename_view_mixed_4 as select id from rename_view_base;
+rename table rename_view_mixed_4 to rename_view_mixed_4_done, rename_view_ignored_table_4 to rename_view_ignored_table_4_done;
+
+-- Ignoring the view rename must not interfere with the table rename or its DML.
+create table rename_view_table_5(id int primary key);
+insert into rename_view_table_5 values(1);
+create view rename_view_ignored_view_5 as select id from rename_view_base;
+rename table rename_view_table_5 to rename_view_table_5_done, rename_view_ignored_view_5 to rename_view_ignored_view_5_done;
+insert into rename_view_table_5_done values(2);
+
+create table rename_view_table_6(id int primary key);
+insert into rename_view_table_6 values(1);
+create view rename_view_ignored_view_6 as select id from rename_view_base;
+rename table rename_view_ignored_view_6 to rename_view_ignored_view_6_done, rename_view_table_6 to rename_view_table_6_done;
+insert into rename_view_table_6_done values(2);
+
+insert into view_ddl_test.finish_mark values(100);