From b940b33186317d9d21087a45d3048a3e358727a2 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 19:02:52 +0800 Subject: [PATCH 1/7] schemastore: mark tables deleted after dropping database --- logservice/schemastore/persist_storage.go | 16 +++++- .../persist_storage_ddl_handlers.go | 33 +++++++++--- .../schemastore/persist_storage_test.go | 52 +++++++++++++++++++ 3 files changed, 91 insertions(+), 10 deletions(-) diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index b7ec23606d..83ba354b46 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -787,6 +787,13 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { tableTriggerDDLHistory: p.tableTriggerDDLHistory, }) + // A drop-schema DDL removes the database metadata below. Collect its physical + // table IDs first so registered version stores also receive the delete version. + var dropSchemaTableIDs []int64 + if job.Type == model.ActionDropSchema { + dropSchemaTableIDs = getSchemaPhysicalTableIDs(ddlEvent.SchemaID, p.databaseMap, p.partitionMap) + } + handler.updateSchemaMetadataFunc(updateSchemaMetadataFuncArgs{ event: &ddlEvent, databaseMap: p.databaseMap, @@ -794,7 +801,7 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { partitionMap: p.partitionMap, }) - handler.iterateEventTablesFunc(&ddlEvent, func(tableIDs ...int64) { + applyDDLToTableInfoStores := func(tableIDs ...int64) { for _, tableID := range tableIDs { if store, ok := p.tableInfoStoreMap[tableID]; ok { // do some safety check @@ -807,7 +814,12 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { store.applyDDL(&ddlEvent) } } - }) + } + if job.Type == model.ActionDropSchema { + applyDDLToTableInfoStores(dropSchemaTableIDs...) + } else { + handler.iterateEventTablesFunc(&ddlEvent, applyDDLToTableInfoStores) + } return nil } diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go index 2439fb2974..605b3e3d5a 100644 --- a/logservice/schemastore/persist_storage_ddl_handlers.go +++ b/logservice/schemastore/persist_storage_ddl_handlers.go @@ -133,7 +133,7 @@ var allDDLHandlers = map[model.ActionType]*persistStorageDDLHandler{ updateFullTableInfoFunc: updateFullTableInfoForDropSchema, updateSchemaMetadataFunc: updateSchemaMetadataForDropSchema, iterateEventTablesFunc: iterateEventTablesIgnore, - extractTableInfoFunc: extractTableInfoFuncIgnore, + extractTableInfoFunc: extractTableInfoFuncForDropSchema, buildDDLEventFunc: buildDDLEventForDropSchema, }, model.ActionCreateTable: { @@ -1169,17 +1169,28 @@ func updateDDLHistoryForTableTriggerOnlyDDL(args updateDDLHistoryFuncArgs) []uin return args.tableTriggerDDLHistory } -func updateDDLHistoryForSchemaDDL(args updateDDLHistoryFuncArgs) []uint64 { - args.appendTableTriggerDDLHistory(args.ddlEvent.FinishedTs) - for tableID := range args.databaseMap[args.ddlEvent.SchemaID].Tables { - if partitionInfo, ok := args.partitionMap[tableID]; ok { - for id := range partitionInfo { - args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, id) +func getSchemaPhysicalTableIDs( + schemaID int64, + databaseMap map[int64]*BasicDatabaseInfo, + partitionMap map[int64]BasicPartitionInfo, +) []int64 { + physicalTableIDs := make([]int64, 0) + for tableID := range databaseMap[schemaID].Tables { + if partitionInfo, ok := partitionMap[tableID]; ok { + for partitionID := range partitionInfo { + physicalTableIDs = append(physicalTableIDs, partitionID) } } else { - args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, tableID) + physicalTableIDs = append(physicalTableIDs, tableID) } } + return physicalTableIDs +} + +func updateDDLHistoryForSchemaDDL(args updateDDLHistoryFuncArgs) []uint64 { + args.appendTableTriggerDDLHistory(args.ddlEvent.FinishedTs) + physicalTableIDs := getSchemaPhysicalTableIDs(args.ddlEvent.SchemaID, args.databaseMap, args.partitionMap) + args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, physicalTableIDs...) return args.tableTriggerDDLHistory } @@ -1759,6 +1770,12 @@ func extractTableInfoFuncIgnore(event *PersistedDDLEvent, tableID int64) (*commo return nil, false } +func extractTableInfoFuncForDropSchema(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) { + // Drop-schema events are only added to the DDL history of physical tables in + // the dropped schema, so reaching this extractor means this table was deleted. + return nil, true +} + func extractTableInfoFuncForDropTable(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) { if isPartitionTable(event.TableInfo) { for _, partitionID := range getAllPartitionIDs(event.TableInfo) { diff --git a/logservice/schemastore/persist_storage_test.go b/logservice/schemastore/persist_storage_test.go index d0252e226d..c583601015 100644 --- a/logservice/schemastore/persist_storage_test.go +++ b/logservice/schemastore/persist_storage_test.go @@ -2862,6 +2862,58 @@ func TestRegisterTable(t *testing.T) { }, }, }, + { + name: "drop schema", + initialDBInfos: []mockDBInfo{ + { + dbInfo: &model.DBInfo{ + ID: 50, + Name: ast.NewCIStr("test"), + }, + tables: []*model.TableInfo{ + newEligibleTableInfoForTest(99, "t1"), + newEligibleTableInfoForTest(100, "t2"), + { + ID: 102, + Name: ast.NewCIStr("pt"), + Partition: buildPartitionDefinitionsForTest([]int64{201, 202}), + }, + }, + }, + }, + preDDLTables: []int64{99, 201}, + postDDLTables: []int64{100, 202}, + ddlJobs: []*model.Job{ + buildDropSchemaJobForTest(50, 1030), + }, + queryCases: []QueryTableInfoTestCase{ + { + tableID: 99, + snapTs: 1029, + name: "t1", + }, + { + tableID: 99, + snapTs: 1030, + deleted: true, + }, + { + tableID: 100, + snapTs: 1030, + deleted: true, + }, + { + tableID: 201, + snapTs: 1030, + deleted: true, + }, + { + tableID: 202, + snapTs: 1030, + deleted: true, + }, + }, + }, } for _, tt := range testCases { t.Run(tt.name, func(t *testing.T) { From 856b7e641df90cdff7f84c8af7f8e369a906ff91 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 19:38:24 +0800 Subject: [PATCH 2/7] refactor --- logservice/schemastore/persist_storage.go | 47 +++++------ .../persist_storage_ddl_handlers.go | 81 ++++++++++++------- 2 files changed, 71 insertions(+), 57 deletions(-) diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index 83ba354b46..fd231bcb51 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -787,12 +787,27 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { tableTriggerDDLHistory: p.tableTriggerDDLHistory, }) - // A drop-schema DDL removes the database metadata below. Collect its physical - // table IDs first so registered version stores also receive the delete version. - var dropSchemaTableIDs []int64 - if job.Type == model.ActionDropSchema { - dropSchemaTableIDs = getSchemaPhysicalTableIDs(ddlEvent.SchemaID, p.databaseMap, p.partitionMap) - } + // Iterate before updating schema metadata because some DDLs, such as drop + // schema, need the old metadata to determine their affected physical tables. + handler.iterateEventTablesFunc(iterateEventTablesFuncArgs{ + event: &ddlEvent, + databaseMap: p.databaseMap, + partitionMap: p.partitionMap, + apply: func(tableIDs ...int64) { + for _, tableID := range tableIDs { + if store, ok := p.tableInfoStoreMap[tableID]; ok { + // do some safety check + switch model.ActionType(job.Type) { + case model.ActionCreateTable, model.ActionCreateTables: + // newly created tables should not be registered before this ddl are handled + log.Panic("should not be registered", zap.Int64("tableID", tableID)) + default: + } + store.applyDDL(&ddlEvent) + } + } + }, + }) handler.updateSchemaMetadataFunc(updateSchemaMetadataFuncArgs{ event: &ddlEvent, @@ -801,26 +816,6 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { partitionMap: p.partitionMap, }) - applyDDLToTableInfoStores := func(tableIDs ...int64) { - for _, tableID := range tableIDs { - if store, ok := p.tableInfoStoreMap[tableID]; ok { - // do some safety check - switch model.ActionType(job.Type) { - case model.ActionCreateTable, model.ActionCreateTables: - // newly created tables should not be registered before this ddl are handled - log.Panic("should not be registered", zap.Int64("tableID", tableID)) - default: - } - store.applyDDL(&ddlEvent) - } - } - } - if job.Type == model.ActionDropSchema { - applyDDLToTableInfoStores(dropSchemaTableIDs...) - } else { - handler.iterateEventTablesFunc(&ddlEvent, applyDDLToTableInfoStores) - } - return nil } diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go index 605b3e3d5a..5edaf11ede 100644 --- a/logservice/schemastore/persist_storage_ddl_handlers.go +++ b/logservice/schemastore/persist_storage_ddl_handlers.go @@ -69,6 +69,13 @@ type updateSchemaMetadataFuncArgs struct { partitionMap map[int64]BasicPartitionInfo } +type iterateEventTablesFuncArgs struct { + event *PersistedDDLEvent + databaseMap map[int64]*BasicDatabaseInfo + partitionMap map[int64]BasicPartitionInfo + apply func(tableIDs ...int64) +} + func (args *updateSchemaMetadataFuncArgs) addTableToDB(tableID int64, schemaID int64) { databaseInfo, ok := args.databaseMap[schemaID] if !ok { @@ -107,7 +114,7 @@ type persistStorageDDLHandler struct { // iterateEventTablesFunc iterates through all physical table IDs affected by the DDL event // and calls the provided `apply` function with those IDs. For partition tables, it includes // all partition IDs. - iterateEventTablesFunc func(event *PersistedDDLEvent, apply func(tableIDs ...int64)) + iterateEventTablesFunc func(args iterateEventTablesFuncArgs) // extractTableInfoFunc extract (table info, deleted) for the specified `tableID` from ddl event extractTableInfoFunc func(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) // buildDDLEvent build a DDLEvent from a PersistedDDLEvent @@ -132,7 +139,7 @@ var allDDLHandlers = map[model.ActionType]*persistStorageDDLHandler{ updateDDLHistoryFunc: updateDDLHistoryForSchemaDDL, updateFullTableInfoFunc: updateFullTableInfoForDropSchema, updateSchemaMetadataFunc: updateSchemaMetadataForDropSchema, - iterateEventTablesFunc: iterateEventTablesIgnore, + iterateEventTablesFunc: iterateEventTablesForDropSchema, extractTableInfoFunc: extractTableInfoFuncForDropSchema, buildDDLEventFunc: buildDDLEventForDropSchema, }, @@ -1169,28 +1176,17 @@ func updateDDLHistoryForTableTriggerOnlyDDL(args updateDDLHistoryFuncArgs) []uin return args.tableTriggerDDLHistory } -func getSchemaPhysicalTableIDs( - schemaID int64, - databaseMap map[int64]*BasicDatabaseInfo, - partitionMap map[int64]BasicPartitionInfo, -) []int64 { - physicalTableIDs := make([]int64, 0) - for tableID := range databaseMap[schemaID].Tables { - if partitionInfo, ok := partitionMap[tableID]; ok { +func updateDDLHistoryForSchemaDDL(args updateDDLHistoryFuncArgs) []uint64 { + args.appendTableTriggerDDLHistory(args.ddlEvent.FinishedTs) + for tableID := range args.databaseMap[args.ddlEvent.SchemaID].Tables { + if partitionInfo, ok := args.partitionMap[tableID]; ok { for partitionID := range partitionInfo { - physicalTableIDs = append(physicalTableIDs, partitionID) + args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, partitionID) } } else { - physicalTableIDs = append(physicalTableIDs, tableID) + args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, tableID) } } - return physicalTableIDs -} - -func updateDDLHistoryForSchemaDDL(args updateDDLHistoryFuncArgs) []uint64 { - args.appendTableTriggerDDLHistory(args.ddlEvent.FinishedTs) - physicalTableIDs := getSchemaPhysicalTableIDs(args.ddlEvent.SchemaID, args.databaseMap, args.partitionMap) - args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, physicalTableIDs...) return args.tableTriggerDDLHistory } @@ -1605,9 +1601,22 @@ func updateSchemaMetadataForRemovePartitioning(args updateSchemaMetadataFuncArgs // iterateEventTablesFunc begin // ======= -func iterateEventTablesIgnore(event *PersistedDDLEvent, apply func(tableId ...int64)) {} +func iterateEventTablesIgnore(args iterateEventTablesFuncArgs) {} + +func iterateEventTablesForDropSchema(args iterateEventTablesFuncArgs) { + for tableID := range args.databaseMap[args.event.SchemaID].Tables { + if partitionInfo, ok := args.partitionMap[tableID]; ok { + for partitionID := range partitionInfo { + args.apply(partitionID) + } + } else { + args.apply(tableID) + } + } +} -func iterateEventTablesForSingleTableDDL(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForSingleTableDDL(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply if isPartitionTable(event.TableInfo) { apply(getAllPartitionIDs(event.TableInfo)...) } else { @@ -1615,7 +1624,8 @@ func iterateEventTablesForSingleTableDDL(event *PersistedDDLEvent, apply func(ta } } -func iterateEventTablesForTruncateTable(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForTruncateTable(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply if isPartitionTable(event.TableInfo) { apply(event.PrevPartitions...) apply(getAllPartitionIDs(event.TableInfo)...) @@ -1624,17 +1634,20 @@ func iterateEventTablesForTruncateTable(event *PersistedDDLEvent, apply func(tab } } -func iterateEventTablesForAddPartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForAddPartition(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply newCreatedIDs := getCreatedIDs(event.PrevPartitions, getAllPartitionIDs(event.TableInfo)) apply(newCreatedIDs...) } -func iterateEventTablesForDropPartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForDropPartition(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply droppedIDs := getDroppedIDs(event.PrevPartitions, getAllPartitionIDs(event.TableInfo)) apply(droppedIDs...) } -func iterateEventTablesForTruncatePartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForTruncatePartition(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply physicalIDs := getAllPartitionIDs(event.TableInfo) droppedIDs := getDroppedIDs(event.PrevPartitions, physicalIDs) apply(droppedIDs...) @@ -1642,7 +1655,8 @@ func iterateEventTablesForTruncatePartition(event *PersistedDDLEvent, apply func apply(newCreatedIDs...) } -func iterateEventTablesForExchangeTablePartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForExchangeTablePartition(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply physicalIDs := getAllPartitionIDs(event.TableInfo) droppedIDs := getDroppedIDs(event.PrevPartitions, physicalIDs) if len(droppedIDs) != 1 { @@ -1653,7 +1667,8 @@ func iterateEventTablesForExchangeTablePartition(event *PersistedDDLEvent, apply apply(targetPartitionID, event.TableID) } -func iterateEventTablesForRenameTables(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForRenameTables(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply for _, info := range event.MultipleTableInfos { if info.ID == InvalidTableID { continue @@ -1666,7 +1681,8 @@ func iterateEventTablesForRenameTables(event *PersistedDDLEvent, apply func(tabl } } -func iterateEventTablesForCreateTables(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForCreateTables(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply for _, info := range event.MultipleTableInfos { if isPartitionTable(info) { apply(getAllPartitionIDs(info)...) @@ -1676,7 +1692,8 @@ func iterateEventTablesForCreateTables(event *PersistedDDLEvent, apply func(tabl } } -func iterateEventTablesForReorganizePartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForReorganizePartition(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply physicalIDs := getAllPartitionIDs(event.TableInfo) droppedIDs := getDroppedIDs(event.PrevPartitions, physicalIDs) apply(droppedIDs...) @@ -1684,7 +1701,8 @@ func iterateEventTablesForReorganizePartition(event *PersistedDDLEvent, apply fu apply(newCreatedIDs...) } -func iterateEventTablesForAlterTablePartitioning(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForAlterTablePartitioning(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply if len(event.PrevPartitions) > 0 { apply(event.PrevPartitions...) } else { @@ -1693,7 +1711,8 @@ func iterateEventTablesForAlterTablePartitioning(event *PersistedDDLEvent, apply apply(getAllPartitionIDs(event.TableInfo)...) } -func iterateEventTablesForRemovePartitioning(event *PersistedDDLEvent, apply func(tableId ...int64)) { +func iterateEventTablesForRemovePartitioning(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply apply(event.PrevPartitions...) apply(event.TableID) } From 3dd7a6c1bc31b4a95ea49d0bbcf1876acce6dfc2 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 19:41:10 +0800 Subject: [PATCH 3/7] f --- logservice/schemastore/persist_storage.go | 31 +++++++++++++---------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index fd231bcb51..d36fdd040e 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -787,25 +787,15 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { tableTriggerDDLHistory: p.tableTriggerDDLHistory, }) - // Iterate before updating schema metadata because some DDLs, such as drop - // schema, need the old metadata to determine their affected physical tables. + // Collect affected tables before updating schema metadata because some DDLs, + // such as drop schema, need the old metadata to determine them. + affectedTableIDs := make([]int64, 0) handler.iterateEventTablesFunc(iterateEventTablesFuncArgs{ event: &ddlEvent, databaseMap: p.databaseMap, partitionMap: p.partitionMap, apply: func(tableIDs ...int64) { - for _, tableID := range tableIDs { - if store, ok := p.tableInfoStoreMap[tableID]; ok { - // do some safety check - switch model.ActionType(job.Type) { - case model.ActionCreateTable, model.ActionCreateTables: - // newly created tables should not be registered before this ddl are handled - log.Panic("should not be registered", zap.Int64("tableID", tableID)) - default: - } - store.applyDDL(&ddlEvent) - } - } + affectedTableIDs = append(affectedTableIDs, tableIDs...) }, }) @@ -816,6 +806,19 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { partitionMap: p.partitionMap, }) + for _, tableID := range affectedTableIDs { + if store, ok := p.tableInfoStoreMap[tableID]; ok { + // do some safety check + switch model.ActionType(job.Type) { + case model.ActionCreateTable, model.ActionCreateTables: + // newly created tables should not be registered before this ddl are handled + log.Panic("should not be registered", zap.Int64("tableID", tableID)) + default: + } + store.applyDDL(&ddlEvent) + } + } + return nil } From 563bad5a6ed21561a52a52755b58b6019078a89c Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 19:50:12 +0800 Subject: [PATCH 4/7] f --- logservice/schemastore/persist_storage.go | 31 ++++++++++------------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index d36fdd040e..fd231bcb51 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -787,15 +787,25 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { tableTriggerDDLHistory: p.tableTriggerDDLHistory, }) - // Collect affected tables before updating schema metadata because some DDLs, - // such as drop schema, need the old metadata to determine them. - affectedTableIDs := make([]int64, 0) + // Iterate before updating schema metadata because some DDLs, such as drop + // schema, need the old metadata to determine their affected physical tables. handler.iterateEventTablesFunc(iterateEventTablesFuncArgs{ event: &ddlEvent, databaseMap: p.databaseMap, partitionMap: p.partitionMap, apply: func(tableIDs ...int64) { - affectedTableIDs = append(affectedTableIDs, tableIDs...) + for _, tableID := range tableIDs { + if store, ok := p.tableInfoStoreMap[tableID]; ok { + // do some safety check + switch model.ActionType(job.Type) { + case model.ActionCreateTable, model.ActionCreateTables: + // newly created tables should not be registered before this ddl are handled + log.Panic("should not be registered", zap.Int64("tableID", tableID)) + default: + } + store.applyDDL(&ddlEvent) + } + } }, }) @@ -806,19 +816,6 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { partitionMap: p.partitionMap, }) - for _, tableID := range affectedTableIDs { - if store, ok := p.tableInfoStoreMap[tableID]; ok { - // do some safety check - switch model.ActionType(job.Type) { - case model.ActionCreateTable, model.ActionCreateTables: - // newly created tables should not be registered before this ddl are handled - log.Panic("should not be registered", zap.Int64("tableID", tableID)) - default: - } - store.applyDDL(&ddlEvent) - } - } - return nil } From 4aa725eb4df9c0f8d70b546b27cb3a4e53bff85d Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 20:00:55 +0800 Subject: [PATCH 5/7] f --- logservice/schemastore/persist_storage.go | 6 +-- .../schemastore/persist_storage_test.go | 49 +++++++++++++++++++ 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index fd231bcb51..7446598e2e 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -796,11 +796,11 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { apply: func(tableIDs ...int64) { for _, tableID := range tableIDs { if store, ok := p.tableInfoStoreMap[tableID]; ok { - // do some safety check switch model.ActionType(job.Type) { case model.ActionCreateTable, model.ActionCreateTables: - // newly created tables should not be registered before this ddl are handled - log.Panic("should not be registered", zap.Int64("tableID", tableID)) + log.Warn("table was registered before create DDL was handled", + zap.Int64("tableID", tableID), + zap.Uint64("finishedTs", ddlEvent.FinishedTs)) default: } store.applyDDL(&ddlEvent) diff --git a/logservice/schemastore/persist_storage_test.go b/logservice/schemastore/persist_storage_test.go index c583601015..a4453de18e 100644 --- a/logservice/schemastore/persist_storage_test.go +++ b/logservice/schemastore/persist_storage_test.go @@ -2717,6 +2717,55 @@ func TestRegisterTable(t *testing.T) { postDDLTables []int64 queryCases []QueryTableInfoTestCase }{ + { + name: "create table registered before DDL", + initialDBInfos: []mockDBInfo{ + { + dbInfo: &model.DBInfo{ + ID: 50, + Name: ast.NewCIStr("test"), + }, + }, + }, + ddlJobs: []*model.Job{ + buildCreateTableJobForTest(50, 99, "t1", 1000), + }, + preDDLTables: []int64{99}, + queryCases: []QueryTableInfoTestCase{ + { + tableID: 99, + snapTs: 1000, + name: "t1", + }, + }, + }, + { + name: "create tables registered before DDL", + initialDBInfos: []mockDBInfo{ + { + dbInfo: &model.DBInfo{ + ID: 50, + Name: ast.NewCIStr("test"), + }, + }, + }, + ddlJobs: []*model.Job{ + buildCreateTablesJobForTest(50, []int64{99, 100}, []string{"t1", "t2"}, 1000), + }, + preDDLTables: []int64{99, 100}, + queryCases: []QueryTableInfoTestCase{ + { + tableID: 99, + snapTs: 1000, + name: "t1", + }, + { + tableID: 100, + snapTs: 1000, + name: "t2", + }, + }, + }, { name: "rename table", initialDBInfos: []mockDBInfo{ From fe20911d358f660ca10a08b59ce512531760c622 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 21:57:28 +0800 Subject: [PATCH 6/7] fix lint --- logservice/schemastore/persist_storage.go | 4 ++-- logservice/schemastore/persist_storage_ddl_handlers.go | 4 ++-- pkg/eventservice/event_broker_test.go | 6 ++---- 3 files changed, 6 insertions(+), 8 deletions(-) diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index 7446598e2e..3bb1d4f47c 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -796,7 +796,7 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { apply: func(tableIDs ...int64) { for _, tableID := range tableIDs { if store, ok := p.tableInfoStoreMap[tableID]; ok { - switch model.ActionType(job.Type) { + switch job.Type { case model.ActionCreateTable, model.ActionCreateTables: log.Warn("table was registered before create DDL was handled", zap.Int64("tableID", tableID), @@ -820,7 +820,7 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { } func shouldSkipDDL(job *model.Job, tableMap map[int64]*BasicTableInfo) bool { - switch model.ActionType(job.Type) { + switch job.Type { // Skipping ActionCreateTable and ActionCreateTables when the table already exists: // 1. It is possible to receive ActionCreateTable and ActionCreateTables multiple times, // and filtering duplicates in a generic way is challenging. diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go index 5edaf11ede..1027420f4b 100644 --- a/logservice/schemastore/persist_storage_ddl_handlers.go +++ b/logservice/schemastore/persist_storage_ddl_handlers.go @@ -1601,7 +1601,7 @@ func updateSchemaMetadataForRemovePartitioning(args updateSchemaMetadataFuncArgs // iterateEventTablesFunc begin // ======= -func iterateEventTablesIgnore(args iterateEventTablesFuncArgs) {} +func iterateEventTablesIgnore(_ iterateEventTablesFuncArgs) {} func iterateEventTablesForDropSchema(args iterateEventTablesFuncArgs) { for tableID := range args.databaseMap[args.event.SchemaID].Tables { @@ -1789,7 +1789,7 @@ func extractTableInfoFuncIgnore(event *PersistedDDLEvent, tableID int64) (*commo return nil, false } -func extractTableInfoFuncForDropSchema(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) { +func extractTableInfoFuncForDropSchema(_ *PersistedDDLEvent, _ int64) (*common.TableInfo, bool) { // Drop-schema events are only added to the DDL history of physical tables in // the dropped schema, so reaching this extractor means this table was deleted. return nil, true diff --git a/pkg/eventservice/event_broker_test.go b/pkg/eventservice/event_broker_test.go index 33342f506d..aa87a9de51 100644 --- a/pkg/eventservice/event_broker_test.go +++ b/pkg/eventservice/event_broker_test.go @@ -1287,14 +1287,12 @@ func TestSendHandshakeIfNeedConcurrency(t *testing.T) { // Launch multiple goroutines to call sendHandshakeIfNeed concurrently for i := 0; i < numGoroutines; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { // Wait for all goroutines to be ready startBarrier.Wait() // Call the method broker.sendHandshakeIfNeed(disp) - }() + }) } // Start all goroutines at the same time From fc456d9ebdd20b8786cfb7adedaa6db8b9fd222f Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 21:59:04 +0800 Subject: [PATCH 7/7] fix --- pkg/eventservice/event_broker_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pkg/eventservice/event_broker_test.go b/pkg/eventservice/event_broker_test.go index aa87a9de51..33342f506d 100644 --- a/pkg/eventservice/event_broker_test.go +++ b/pkg/eventservice/event_broker_test.go @@ -1287,12 +1287,14 @@ func TestSendHandshakeIfNeedConcurrency(t *testing.T) { // Launch multiple goroutines to call sendHandshakeIfNeed concurrently for i := 0; i < numGoroutines; i++ { - wg.Go(func() { + wg.Add(1) + go func() { + defer wg.Done() // Wait for all goroutines to be ready startBarrier.Wait() // Call the method broker.sendHandshakeIfNeed(disp) - }) + }() } // Start all goroutines at the same time