diff --git a/logservice/schemastore/persist_storage.go b/logservice/schemastore/persist_storage.go index b7ec23606d..3bb1d4f47c 100644 --- a/logservice/schemastore/persist_storage.go +++ b/logservice/schemastore/persist_storage.go @@ -787,33 +787,40 @@ func (p *persistentStorage) handleDDLJob(job *model.Job) error { tableTriggerDDLHistory: p.tableTriggerDDLHistory, }) - handler.updateSchemaMetadataFunc(updateSchemaMetadataFuncArgs{ + // 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, - tableMap: p.tableMap, partitionMap: p.partitionMap, - }) - - handler.iterateEventTablesFunc(&ddlEvent, 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: + apply: func(tableIDs ...int64) { + for _, tableID := range tableIDs { + if store, ok := p.tableInfoStoreMap[tableID]; ok { + switch job.Type { + case model.ActionCreateTable, model.ActionCreateTables: + log.Warn("table was registered before create DDL was handled", + zap.Int64("tableID", tableID), + zap.Uint64("finishedTs", ddlEvent.FinishedTs)) + default: + } + store.applyDDL(&ddlEvent) } - store.applyDDL(&ddlEvent) } - } + }, + }) + + handler.updateSchemaMetadataFunc(updateSchemaMetadataFuncArgs{ + event: &ddlEvent, + databaseMap: p.databaseMap, + tableMap: p.tableMap, + partitionMap: p.partitionMap, }) return nil } 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 2439fb2974..1027420f4b 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,8 +139,8 @@ var allDDLHandlers = map[model.ActionType]*persistStorageDDLHandler{ updateDDLHistoryFunc: updateDDLHistoryForSchemaDDL, updateFullTableInfoFunc: updateFullTableInfoForDropSchema, updateSchemaMetadataFunc: updateSchemaMetadataForDropSchema, - iterateEventTablesFunc: iterateEventTablesIgnore, - extractTableInfoFunc: extractTableInfoFuncIgnore, + iterateEventTablesFunc: iterateEventTablesForDropSchema, + extractTableInfoFunc: extractTableInfoFuncForDropSchema, buildDDLEventFunc: buildDDLEventForDropSchema, }, model.ActionCreateTable: { @@ -1173,8 +1180,8 @@ 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) + for partitionID := range partitionInfo { + args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, partitionID) } } else { args.appendTablesDDLHistory(args.ddlEvent.FinishedTs, tableID) @@ -1594,9 +1601,22 @@ func updateSchemaMetadataForRemovePartitioning(args updateSchemaMetadataFuncArgs // iterateEventTablesFunc begin // ======= -func iterateEventTablesIgnore(event *PersistedDDLEvent, apply func(tableId ...int64)) {} +func iterateEventTablesIgnore(_ iterateEventTablesFuncArgs) {} -func iterateEventTablesForSingleTableDDL(event *PersistedDDLEvent, apply func(tableId ...int64)) { +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(args iterateEventTablesFuncArgs) { + event, apply := args.event, args.apply if isPartitionTable(event.TableInfo) { apply(getAllPartitionIDs(event.TableInfo)...) } else { @@ -1604,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)...) @@ -1613,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...) @@ -1631,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 { @@ -1642,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 @@ -1655,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)...) @@ -1665,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...) @@ -1673,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 { @@ -1682,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) } @@ -1759,6 +1789,12 @@ func extractTableInfoFuncIgnore(event *PersistedDDLEvent, tableID int64) (*commo return nil, false } +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 +} + 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..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{ @@ -2862,6 +2911,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) {