From 452e35ee4ecb73c6e9e47c587d99afa8a6766e9d Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 22:13:41 +0800 Subject: [PATCH 1/3] schemastore: update TableInfo for surviving partitions --- logservice/schemastore/multi_version_test.go | 126 ++++++++++++++++++ .../persist_storage_ddl_handlers.go | 26 ++-- 2 files changed, 139 insertions(+), 13 deletions(-) diff --git a/logservice/schemastore/multi_version_test.go b/logservice/schemastore/multi_version_test.go index 8cfb443c7e..f1e616c3ee 100644 --- a/logservice/schemastore/multi_version_test.go +++ b/logservice/schemastore/multi_version_test.go @@ -17,6 +17,7 @@ import ( "testing" "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/tidb/pkg/meta/model" "github.com/stretchr/testify/require" ) @@ -233,6 +234,131 @@ func TestBuildVersionedTableInfoStore(t *testing.T) { } } +func TestPartitionDDLUpdatesSurvivingPartitionTableInfo(t *testing.T) { + const ( + logicalTableID = int64(100) + survivorID = int64(201) + oldUpdateTS = uint64(1000) + newUpdateTS = uint64(2000) + ddlFinishedTS = uint64(3000) + ) + + testCases := []struct { + name string + ddlType model.ActionType + previousIDs []int64 + currentIDs []int64 + expectedUpdateTS uint64 + droppedID int64 + }{ + { + name: "add partition", + ddlType: model.ActionAddTablePartition, + previousIDs: []int64{201, 202}, + currentIDs: []int64{201, 202, 203}, + expectedUpdateTS: newUpdateTS, + }, + { + name: "drop partition", + ddlType: model.ActionDropTablePartition, + previousIDs: []int64{201, 202, 203}, + currentIDs: []int64{201, 202}, + expectedUpdateTS: newUpdateTS, + droppedID: 203, + }, + { + name: "reorganize partition", + ddlType: model.ActionReorganizePartition, + previousIDs: []int64{201, 202, 203}, + currentIDs: []int64{201, 204, 205}, + expectedUpdateTS: newUpdateTS, + droppedID: 202, + }, + { + name: "truncate partition", + ddlType: model.ActionTruncateTablePartition, + previousIDs: []int64{201, 202, 203}, + currentIDs: []int64{201, 202, 204}, + expectedUpdateTS: oldUpdateTS, + droppedID: 203, + }, + } + + newPartitionTableInfo := func(partitionIDs []int64, updateTS uint64) *model.TableInfo { + tableInfo := newEligibleTableInfoForTest(logicalTableID, "t") + tableInfo.Partition = buildPartitionDefinitionsForTest(partitionIDs) + tableInfo.UpdateTS = updateTS + return tableInfo + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + event := &PersistedDDLEvent{ + Type: byte(tc.ddlType), + SchemaID: 10, + TableID: logicalTableID, + SchemaName: "test", + TableName: "t", + TableInfo: newPartitionTableInfo(tc.currentIDs, tc.expectedUpdateTS), + PrevPartitions: tc.previousIDs, + FinishedTs: ddlFinishedTS, + } + handler := allDDLHandlers[tc.ddlType] + + newStore := func(initialized bool) *versionedTableInfoStore { + store := newEmptyVersionedTableInfoStore(survivorID) + store.addInitialTableInfo( + common.WrapTableInfo("test", newPartitionTableInfo(tc.previousIDs, oldUpdateTS)), + oldUpdateTS, + ) + if initialized { + store.setTableInfoInitialized() + } + return store + } + assertUpdated := func(store *versionedTableInfoStore) { + tableInfo, err := store.getTableInfo(ddlFinishedTS) + require.NoError(t, err) + require.Equal(t, tc.expectedUpdateTS, tableInfo.GetUpdateTS()) + require.Len(t, store.infos, 2) + require.Equal(t, ddlFinishedTS, store.infos[1].Version) + } + + // Verify that the online path applies the DDL to a registered surviving partition. + liveStore := newStore(true) + applied := false + handler.iterateEventTablesFunc(event, func(tableIDs ...int64) { + for _, tableID := range tableIDs { + if tableID == survivorID { + liveStore.applyDDL(event) + applied = true + } + } + }) + require.True(t, applied) + assertUpdated(liveStore) + + // Verify that the history path records and extracts the same update. + tablesDDLHistory := make(map[int64][]uint64) + handler.updateDDLHistoryFunc(updateDDLHistoryFuncArgs{ + ddlEvent: event, + tablesDDLHistory: tablesDDLHistory, + }) + require.Equal(t, []uint64{ddlFinishedTS}, tablesDDLHistory[survivorID]) + historyStore := newStore(false) + historyStore.applyDDLFromPersistStorage(event) + historyStore.setTableInfoInitialized() + assertUpdated(historyStore) + + if tc.droppedID != 0 { + tableInfo, deleted := handler.extractTableInfoFunc(event, tc.droppedID) + require.Nil(t, tableInfo) + require.True(t, deleted) + } + }) + } +} + func TestGCMultiVersionTableInfo(t *testing.T) { tableID := int64(100) store := newEmptyVersionedTableInfoStore(tableID) diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go index 2439fb2974..0828e23226 100644 --- a/logservice/schemastore/persist_storage_ddl_handlers.go +++ b/logservice/schemastore/persist_storage_ddl_handlers.go @@ -1614,19 +1614,16 @@ func iterateEventTablesForTruncateTable(event *PersistedDDLEvent, apply func(tab } func iterateEventTablesForAddPartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { - newCreatedIDs := getCreatedIDs(event.PrevPartitions, getAllPartitionIDs(event.TableInfo)) - apply(newCreatedIDs...) + apply(getAllPartitionIDs(event.TableInfo)...) } func iterateEventTablesForDropPartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { - droppedIDs := getDroppedIDs(event.PrevPartitions, getAllPartitionIDs(event.TableInfo)) - apply(droppedIDs...) + apply(event.PrevPartitions...) } func iterateEventTablesForTruncatePartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { physicalIDs := getAllPartitionIDs(event.TableInfo) - droppedIDs := getDroppedIDs(event.PrevPartitions, physicalIDs) - apply(droppedIDs...) + apply(event.PrevPartitions...) newCreatedIDs := getCreatedIDs(event.PrevPartitions, physicalIDs) apply(newCreatedIDs...) } @@ -1667,8 +1664,7 @@ func iterateEventTablesForCreateTables(event *PersistedDDLEvent, apply func(tabl func iterateEventTablesForReorganizePartition(event *PersistedDDLEvent, apply func(tableId ...int64)) { physicalIDs := getAllPartitionIDs(event.TableInfo) - droppedIDs := getDroppedIDs(event.PrevPartitions, physicalIDs) - apply(droppedIDs...) + apply(event.PrevPartitions...) newCreatedIDs := getCreatedIDs(event.PrevPartitions, physicalIDs) apply(newCreatedIDs...) } @@ -1798,8 +1794,7 @@ func extractTableInfoFuncForTruncateTable(event *PersistedDDLEvent, tableID int6 } func extractTableInfoFuncForAddPartition(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) { - newCreatedIDs := getCreatedIDs(event.PrevPartitions, getAllPartitionIDs(event.TableInfo)) - for _, partition := range newCreatedIDs { + for _, partition := range getAllPartitionIDs(event.TableInfo) { if tableID == partition { return common.WrapTableInfo(event.SchemaName, event.TableInfo), false } @@ -1808,12 +1803,18 @@ func extractTableInfoFuncForAddPartition(event *PersistedDDLEvent, tableID int64 } func extractTableInfoFuncForDropPartition(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) { - droppedIDs := getDroppedIDs(event.PrevPartitions, getAllPartitionIDs(event.TableInfo)) + physicalIDs := getAllPartitionIDs(event.TableInfo) + droppedIDs := getDroppedIDs(event.PrevPartitions, physicalIDs) for _, partition := range droppedIDs { if tableID == partition { return nil, true } } + for _, partition := range physicalIDs { + if tableID == partition { + return common.WrapTableInfo(event.SchemaName, event.TableInfo), false + } + } return nil, false } @@ -1825,8 +1826,7 @@ func extractTableInfoFuncForTruncateAndReorganizePartition(event *PersistedDDLEv return nil, true } } - newCreatedIDs := getCreatedIDs(event.PrevPartitions, physicalIDs) - for _, partition := range newCreatedIDs { + for _, partition := range physicalIDs { if tableID == partition { return common.WrapTableInfo(event.SchemaName, event.TableInfo), false } From b26ad25ca2031fb933c74608a64190f2393ad895 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 12 Aug 2026 23:02:30 +0800 Subject: [PATCH 2/3] fix lint error --- .../persist_storage_ddl_handlers.go | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go index 0828e23226..266c403d7a 100644 --- a/logservice/schemastore/persist_storage_ddl_handlers.go +++ b/logservice/schemastore/persist_storage_ddl_handlers.go @@ -16,6 +16,7 @@ package schemastore import ( "errors" "fmt" + "slices" "strings" "github.com/pingcap/log" @@ -1794,10 +1795,8 @@ func extractTableInfoFuncForTruncateTable(event *PersistedDDLEvent, tableID int6 } func extractTableInfoFuncForAddPartition(event *PersistedDDLEvent, tableID int64) (*common.TableInfo, bool) { - for _, partition := range getAllPartitionIDs(event.TableInfo) { - if tableID == partition { - return common.WrapTableInfo(event.SchemaName, event.TableInfo), false - } + if slices.Contains(getAllPartitionIDs(event.TableInfo), tableID) { + return common.WrapTableInfo(event.SchemaName, event.TableInfo), false } return nil, false } @@ -1810,10 +1809,8 @@ func extractTableInfoFuncForDropPartition(event *PersistedDDLEvent, tableID int6 return nil, true } } - for _, partition := range physicalIDs { - if tableID == partition { - return common.WrapTableInfo(event.SchemaName, event.TableInfo), false - } + if slices.Contains(physicalIDs, tableID) { + return common.WrapTableInfo(event.SchemaName, event.TableInfo), false } return nil, false } @@ -1826,10 +1823,8 @@ func extractTableInfoFuncForTruncateAndReorganizePartition(event *PersistedDDLEv return nil, true } } - for _, partition := range physicalIDs { - if tableID == partition { - return common.WrapTableInfo(event.SchemaName, event.TableInfo), false - } + if slices.Contains(physicalIDs, tableID) { + return common.WrapTableInfo(event.SchemaName, event.TableInfo), false } return nil, false } From 2dbfce31e6c71e529f5200f917e917a1768bd635 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 13 Aug 2026 15:42:38 +0800 Subject: [PATCH 3/3] improve test --- logservice/schemastore/multi_version_test.go | 80 ++++++++++++-------- 1 file changed, 47 insertions(+), 33 deletions(-) diff --git a/logservice/schemastore/multi_version_test.go b/logservice/schemastore/multi_version_test.go index f1e616c3ee..ff46360e09 100644 --- a/logservice/schemastore/multi_version_test.go +++ b/logservice/schemastore/multi_version_test.go @@ -244,43 +244,48 @@ func TestPartitionDDLUpdatesSurvivingPartitionTableInfo(t *testing.T) { ) testCases := []struct { - name string - ddlType model.ActionType - previousIDs []int64 - currentIDs []int64 - expectedUpdateTS uint64 - droppedID int64 + name string + ddlType model.ActionType + previousIDs []int64 + currentIDs []int64 + expectedAffectedIDs []int64 + expectedUpdateTS uint64 + droppedID int64 }{ { - name: "add partition", - ddlType: model.ActionAddTablePartition, - previousIDs: []int64{201, 202}, - currentIDs: []int64{201, 202, 203}, - expectedUpdateTS: newUpdateTS, + name: "add partition", + ddlType: model.ActionAddTablePartition, + previousIDs: []int64{201, 202}, + currentIDs: []int64{201, 202, 203}, + expectedAffectedIDs: []int64{201, 202, 203}, + expectedUpdateTS: newUpdateTS, }, { - name: "drop partition", - ddlType: model.ActionDropTablePartition, - previousIDs: []int64{201, 202, 203}, - currentIDs: []int64{201, 202}, - expectedUpdateTS: newUpdateTS, - droppedID: 203, + name: "drop partition", + ddlType: model.ActionDropTablePartition, + previousIDs: []int64{201, 202, 203}, + currentIDs: []int64{201, 202}, + expectedAffectedIDs: []int64{201, 202, 203}, + expectedUpdateTS: newUpdateTS, + droppedID: 203, }, { - name: "reorganize partition", - ddlType: model.ActionReorganizePartition, - previousIDs: []int64{201, 202, 203}, - currentIDs: []int64{201, 204, 205}, - expectedUpdateTS: newUpdateTS, - droppedID: 202, + name: "reorganize partition", + ddlType: model.ActionReorganizePartition, + previousIDs: []int64{201, 202, 203}, + currentIDs: []int64{201, 204, 205}, + expectedAffectedIDs: []int64{201, 202, 203, 204, 205}, + expectedUpdateTS: newUpdateTS, + droppedID: 202, }, { - name: "truncate partition", - ddlType: model.ActionTruncateTablePartition, - previousIDs: []int64{201, 202, 203}, - currentIDs: []int64{201, 202, 204}, - expectedUpdateTS: oldUpdateTS, - droppedID: 203, + name: "truncate partition", + ddlType: model.ActionTruncateTablePartition, + previousIDs: []int64{201, 202, 203}, + currentIDs: []int64{201, 202, 204}, + expectedAffectedIDs: []int64{201, 202, 203, 204}, + expectedUpdateTS: oldUpdateTS, + droppedID: 203, }, } @@ -326,16 +331,16 @@ func TestPartitionDDLUpdatesSurvivingPartitionTableInfo(t *testing.T) { // Verify that the online path applies the DDL to a registered surviving partition. liveStore := newStore(true) - applied := false + affectedIDs := make([]int64, 0, len(tc.expectedAffectedIDs)) handler.iterateEventTablesFunc(event, func(tableIDs ...int64) { + affectedIDs = append(affectedIDs, tableIDs...) for _, tableID := range tableIDs { if tableID == survivorID { liveStore.applyDDL(event) - applied = true } } }) - require.True(t, applied) + require.ElementsMatch(t, tc.expectedAffectedIDs, affectedIDs) assertUpdated(liveStore) // Verify that the history path records and extracts the same update. @@ -344,12 +349,21 @@ func TestPartitionDDLUpdatesSurvivingPartitionTableInfo(t *testing.T) { ddlEvent: event, tablesDDLHistory: tablesDDLHistory, }) - require.Equal(t, []uint64{ddlFinishedTS}, tablesDDLHistory[survivorID]) + require.Len(t, tablesDDLHistory, len(tc.expectedAffectedIDs)) + for _, tableID := range tc.expectedAffectedIDs { + require.Equal(t, []uint64{ddlFinishedTS}, tablesDDLHistory[tableID]) + } historyStore := newStore(false) historyStore.applyDDLFromPersistStorage(event) historyStore.setTableInfoInitialized() assertUpdated(historyStore) + for _, tableID := range tc.currentIDs { + tableInfo, deleted := handler.extractTableInfoFunc(event, tableID) + require.NotNil(t, tableInfo) + require.False(t, deleted) + require.Equal(t, tc.expectedUpdateTS, tableInfo.GetUpdateTS()) + } if tc.droppedID != 0 { tableInfo, deleted := handler.extractTableInfoFunc(event, tc.droppedID) require.Nil(t, tableInfo)