From e92e01737b7eb009f677e5056eae24f23a2f94b4 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 16 Oct 2026 07:44:59 +0000 Subject: [PATCH 1/2] init Signed-off-by: wk989898 --- .../persist_storage_ddl_handlers.go | 94 ++++++++++- .../schemastore/persist_storage_test.go | 60 +++++++ .../multi_tables_ddl/conf/normal.toml | 6 + .../multi_tables_ddl/data/test.sql | 54 ++++++ .../integration_tests/multi_tables_ddl/run.sh | 155 +++++++++++++++++- 5 files changed, 352 insertions(+), 17 deletions(-) diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go index 8f480ea593..bf45c3b333 100644 --- a/logservice/schemastore/persist_storage_ddl_handlers.go +++ b/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, diff --git a/logservice/schemastore/persist_storage_test.go b/logservice/schemastore/persist_storage_test.go index d74197836e..fc9ca16e10 100644 --- a/logservice/schemastore/persist_storage_test.go +++ b/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: ast.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: ast.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 diff --git a/tests/integration_tests/multi_tables_ddl/conf/normal.toml b/tests/integration_tests/multi_tables_ddl/conf/normal.toml index a85957996a..28dd25b85c 100644 --- a/tests/integration_tests/multi_tables_ddl/conf/normal.toml +++ b/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 diff --git a/tests/integration_tests/multi_tables_ddl/data/test.sql b/tests/integration_tests/multi_tables_ddl/data/test.sql index 5a6047266d..c3b3a0a68a 100644 --- a/tests/integration_tests/multi_tables_ddl/data/test.sql +++ b/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); diff --git a/tests/integration_tests/multi_tables_ddl/run.sh b/tests/integration_tests/multi_tables_ddl/run.sh index e5937ec053..1d789e8959 100755 --- a/tests/integration_tests/multi_tables_ddl/run.sh +++ b/tests/integration_tests/multi_tables_ddl/run.sh @@ -33,6 +33,103 @@ stop() { stop_test $WORK_DIR } +function check_view_result() { + ensure 20 "run_sql \"$1\" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} && check_contains \"$2\"" +} + +function test_view_ddls() { + # Check progress after excluded view DDLs, then wait for each downstream object. + ensure 30 "run_sql 'SELECT COUNT(*) AS marker_count FROM view_ddl_test.finish_mark WHERE id = 100;' \ + ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} && check_contains 'marker_count: 1'" + for view in single_done mixed_1_done mixed_2_done mixed_3_done mixed_4_done ignored_view_5 ignored_view_6; do + ensure 10 "run_sql 'SELECT COUNT(*) AS view_count FROM view_ddl_test.rename_view_$view WHERE id = 7;' \ + ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} && check_contains 'view_count: 1'" + done + for table in table_1_done table_2_done table_5_done table_6_done; do + ensure 10 "run_sql 'SELECT COUNT(*) AS row_count FROM view_ddl_test.rename_view_$table WHERE id IN (1, 2);' \ + ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} && check_contains 'row_count: 2'" + done + for table in ignored_table_3 ignored_table_4; do + ensure 10 "run_sql 'SELECT COUNT(*) AS row_count FROM view_ddl_test.rename_view_$table WHERE id = 1;' \ + ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} && check_contains 'row_count: 1'" + done + run_sql "SELECT COUNT(*) AS stale_count FROM information_schema.tables + WHERE table_schema = 'view_ddl_test' AND table_name IN ( + 'rename_view_single', 'rename_view_mixed_1', 'rename_view_mixed_2', + 'rename_view_mixed_3', 'rename_view_mixed_4', 'rename_view_table_1', + 'rename_view_table_2', 'rename_view_table_5', 'rename_view_table_6', + 'rename_view_ignored_table_3_done', 'rename_view_ignored_table_4_done', + 'rename_view_ignored_view_5_done', 'rename_view_ignored_view_6_done');" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + check_contains 'stale_count: 0' + run_sql "SELECT COUNT(*) AS excluded_count FROM information_schema.schemata WHERE schema_name = 'unselected_db';" \ + ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + check_contains 'excluded_count: 0' + + run_sql "DROP VIEW view_ddl_test.rename_view_single_done; + INSERT INTO view_ddl_test.finish_mark VALUES (101);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + ensure 30 "run_sql 'SELECT COUNT(*) AS marker_count FROM view_ddl_test.finish_mark WHERE id = 101;' \ + ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} && check_contains 'marker_count: 1'" + check_table_not_exists view_ddl_test.rename_view_single_done ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + + # Views present at initialization must be cleaned up when dropped or replaced. + run_sql "DROP VIEW view_ddl_test.snapshot_drop; + CREATE TABLE view_ddl_test.snapshot_drop(id INT PRIMARY KEY); + INSERT INTO view_ddl_test.snapshot_drop VALUES (1); + CREATE TABLE view_ddl_test.snapshot_copy LIKE view_ddl_test.snapshot_drop; + INSERT INTO view_ddl_test.snapshot_copy VALUES (2); + CREATE OR REPLACE VIEW view_ddl_test.snapshot_replace AS SELECT id, val + 100 AS val FROM view_ddl_test.base; + RENAME TABLE view_ddl_test.snapshot_rename TO view_ddl_test.snapshot_renamed;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT id FROM view_ddl_test.snapshot_copy;" "id: 2" + check_view_result "SELECT val FROM view_ddl_test.snapshot_replace WHERE id = 1;" "val: 110" + check_view_result "SELECT val FROM view_ddl_test.snapshot_renamed WHERE id = 1;" "val: 10" + check_table_not_exists view_ddl_test.snapshot_rename ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + + # Verify each definition before the next DDL so a missed replacement cannot pass. + run_sql "CREATE VIEW view_ddl_test.incremental AS SELECT id, val FROM view_ddl_test.base;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT val FROM view_ddl_test.incremental WHERE id = 1;" "val: 10" + run_sql "CREATE OR REPLACE VIEW view_ddl_test.incremental AS SELECT id, val + 200 AS val FROM view_ddl_test.base;" \ + ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT val FROM view_ddl_test.incremental WHERE id = 1;" "val: 210" + run_sql "RENAME TABLE view_ddl_test.incremental TO view_ddl_test.incremental_renamed; + INSERT INTO view_ddl_test.base VALUES (2, 20);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT val FROM view_ddl_test.incremental_renamed WHERE id = 2;" "val: 220" + check_table_not_exists view_ddl_test.incremental ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + run_sql "DROP VIEW view_ddl_test.incremental_renamed, view_ddl_test.snapshot_replace, + view_ddl_test.snapshot_renamed;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + for view in incremental_renamed snapshot_replace snapshot_renamed; do + check_table_not_exists view_ddl_test.$view ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + done + + # Database recovery must restore view definitions without scheduling view IDs. + run_sql "CREATE DATABASE view_recover_test; + CREATE TABLE view_recover_test.base(id INT PRIMARY KEY, val INT); + INSERT INTO view_recover_test.base VALUES (1, 10); + CREATE VIEW view_recover_test.v AS SELECT id, val + 100 AS val FROM view_recover_test.base;" \ + ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT val FROM view_recover_test.v WHERE id = 1;" "val: 110" + run_sql "DROP DATABASE view_recover_test;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_db_not_exists view_recover_test ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + run_sql "FLASHBACK DATABASE view_recover_test;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT val FROM view_recover_test.v WHERE id = 1;" "val: 110" + run_sql "INSERT INTO view_recover_test.base VALUES (2, 20);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_view_result "SELECT val FROM view_recover_test.v WHERE id = 2;" "val: 120" + + local view_ids + view_ids=$(mysql -uroot -h${UP_TIDB_HOST} -P${UP_TIDB_PORT} -N -B -e \ + "SELECT JSON_ARRAYAGG(TIDB_TABLE_ID) FROM information_schema.tables + WHERE table_schema IN ('view_ddl_test', 'view_recover_test') AND table_type = 'VIEW';") + curl -fsS "http://${CDC_HOST}:${CDC_PORT}/api/v2/changefeeds/$cf_normal/tables?keyspace=${KEYSPACE_NAME}" | + jq -e --argjson views "$view_ids" \ + '[.items[].table_ids[]] as $tables | all($views[]; . as $id | $tables | index($id) | not)' + + run_sql "DROP DATABASE view_recover_test; DROP DATABASE view_ddl_test; + INSERT INTO multi_tables_ddl_test.finish_mark VALUES (1000);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + check_db_not_exists view_recover_test ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + check_db_not_exists view_ddl_test ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + check_view_result "SELECT COUNT(*) AS marker_count FROM multi_tables_ddl_test.finish_mark WHERE id = 1000;" "marker_count: 1" + check_changefeed_state "http://${UP_PD_HOST_1}:${UP_PD_PORT_1}" $cf_normal "normal" "null" "" +} + function test_system_schema_rename() { local rename_type local marker=0 @@ -67,14 +164,6 @@ function test_system_schema_rename() { } function run() { - if [ "$SINK_TYPE" == "storage" ]; then - return - fi - # TODO(dongmen): enable pulsar in the future. - if [ "$SINK_TYPE" == "pulsar" ]; then - exit 0 - fi - start_tidb_cluster --workdir $WORK_DIR bin/minio server --address $S3_ENDPOINT "$WORK_DIR/s3" & @@ -90,6 +179,16 @@ function run() { done s3cmd --access_key=$MINIO_ACCESS_KEY --secret_key=$MINIO_SECRET_KEY --host=$S3_ENDPOINT --host-bucket=$S3_ENDPOINT --no-ssl mb s3://logbucket + # Seed and mirror views before the start TS to exercise snapshot metadata. + local view_schema="CREATE DATABASE view_ddl_test; + CREATE TABLE view_ddl_test.base(id INT PRIMARY KEY, val INT); + INSERT INTO view_ddl_test.base VALUES (1, 10); + CREATE VIEW view_ddl_test.snapshot_drop AS SELECT id, val FROM view_ddl_test.base; + CREATE VIEW view_ddl_test.snapshot_replace AS SELECT id, val FROM view_ddl_test.base; + CREATE VIEW view_ddl_test.snapshot_rename AS SELECT id, val FROM view_ddl_test.base;" + run_sql "$view_schema" ${UP_TIDB_HOST} ${UP_TIDB_PORT} + run_sql "$view_schema" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1) @@ -129,8 +228,45 @@ function run() { run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME_2?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" "$CUR/conf/error-1.toml" run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME_3?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" "$CUR/conf/error-2.toml" ;; + "storage" | "pulsar") + if [ "$SINK_TYPE" == "pulsar" ]; then + run_pulsar_cluster $WORK_DIR normal + fi + local index topic changefeed config + for index in 1 2 3; do + case $index in + 1) + topic=$TOPIC_NAME_1 + changefeed=$cf_normal + config="$CUR/conf/normal.toml" + ;; + 2) + topic=$TOPIC_NAME_2 + changefeed=$cf_err1 + config="$CUR/conf/error-1.toml" + ;; + 3) + topic=$TOPIC_NAME_3 + changefeed=$cf_err2 + config="$CUR/conf/error-2.toml" + ;; + esac + if [ "$SINK_TYPE" == "storage" ]; then + SINK_URI="file://$WORK_DIR/storage_test/$topic?flush-interval=5s&protocol=csv" + else + SINK_URI="pulsar://127.0.0.1:6650/$topic?protocol=canal-json&enable-tidb-extension=true" + fi + cdc_cli_changefeed create -c=$changefeed --start-ts=$start_ts --sink-uri="$SINK_URI" --config="$config" + if [ "$SINK_TYPE" == "storage" ]; then + run_storage_consumer $WORK_DIR "$SINK_URI" "$config" $index + else + log_suffix=$index run_pulsar_consumer --upstream-uri "$SINK_URI" --config "$config" + fi + done + ;; *) - SINK_URI="mysql://normal:123456@127.0.0.1:3306/" + # FLASHBACK DATABASE in the view lifecycle test requires elevated privileges. + SINK_URI="mysql://root@127.0.0.1:3306/" cdc_cli_changefeed create -c=$cf_normal --start-ts=$start_ts --sink-uri="$SINK_URI" --config="$CUR/conf/normal.toml" cdc_cli_changefeed create -c=$cf_err1 --start-ts=$start_ts --sink-uri="$SINK_URI" --config="$CUR/conf/error-1.toml" cdc_cli_changefeed create -c=$cf_err2 --start-ts=$start_ts --sink-uri="$SINK_URI" --config="$CUR/conf/error-2.toml" @@ -160,6 +296,7 @@ function run() { check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml 60 test_system_schema_rename + test_view_ddls cleanup_process $CDC_BINARY } From fb9c1d3e03fa497aa7e87a4a83bbeae6bcf08729 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Thu, 5 Nov 2026 08:23:58 +0000 Subject: [PATCH 2/2] schemastore: adapt view DDL tests to release-8.5 --- logservice/schemastore/persist_storage_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/logservice/schemastore/persist_storage_test.go b/logservice/schemastore/persist_storage_test.go index fc9ca16e10..a8b88bfa7e 100644 --- a/logservice/schemastore/persist_storage_test.go +++ b/logservice/schemastore/persist_storage_test.go @@ -5026,7 +5026,7 @@ func TestBuildSchemaAndViewDDLEvents(t *testing.T) { view := newEligibleTableInfoForTest(500, "v") view.View = &model.ViewInfo{} storage := newPersistentStorageForTest(t.TempDir(), []mockDBInfo{{ - dbInfo: &model.DBInfo{ID: 10, Name: ast.NewCIStr("test")}, + dbInfo: &model.DBInfo{ID: 10, Name: parser_model.NewCIStr("test")}, tables: []*model.TableInfo{view}, }}) t.Cleanup(func() { require.NoError(t, storage.close()) }) @@ -5052,7 +5052,7 @@ func TestBuildSchemaAndViewDDLEvents(t *testing.T) { 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: ast.NewCIStr("view_lifecycle")}, + 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()) })