(t *testing.T)
| 1535 | } |
| 1536 | |
| 1537 | func TestPITRIDMapOnCheckpointStorage(t *testing.T) { |
| 1538 | ctx := context.Background() |
| 1539 | s := utiltest.CreateRestoreSchemaSuite(t) |
| 1540 | tk := testkit.NewTestKit(t, s.Mock.Storage) |
| 1541 | tk.MustExec(session.CreatePITRIDMap) |
| 1542 | g := gluetidb.New() |
| 1543 | se, err := g.CreateSession(s.Mock.Storage) |
| 1544 | require.NoError(t, err) |
| 1545 | client := logclient.TEST_NewLogClient(123, 1, 2, 3, s.Mock.Domain, se) |
| 1546 | client.SetUseCheckpoint() |
| 1547 | stg, err := storage.NewLocalStorage("local://" + filepath.ToSlash(t.TempDir())) |
| 1548 | require.NoError(t, err) |
| 1549 | logCheckpointMetaManager := checkpoint.NewLogStorageMetaManager( |
| 1550 | stg, nil, 123, "test", 1) |
| 1551 | defer logCheckpointMetaManager.Close() |
| 1552 | baseTableMappingManager := &stream.TableMappingManager{ |
| 1553 | DBReplaceMap: getDBMap(), |
| 1554 | } |
| 1555 | err = client.TEST_saveIDMap(ctx, baseTableMappingManager, logCheckpointMetaManager) |
| 1556 | require.NoError(t, err) |
| 1557 | newSchemaReplaces, err := client.TEST_initSchemasMap(ctx, 1, logCheckpointMetaManager) |
| 1558 | require.NoError(t, err) |
| 1559 | require.Nil(t, newSchemaReplaces) |
| 1560 | client2 := logclient.TEST_NewLogClient(123, 1, 2, 4, s.Mock.Domain, se) |
| 1561 | stg, err = storage.NewLocalStorage("local://" + filepath.ToSlash(t.TempDir()) + "/temp_another") |
| 1562 | require.NoError(t, err) |
| 1563 | logCheckpointMetaManager2 := checkpoint.NewLogStorageMetaManager( |
| 1564 | stg, nil, 123, "test", 1) |
| 1565 | defer logCheckpointMetaManager.Close() |
| 1566 | newSchemaReplaces, err = client2.TEST_initSchemasMap(ctx, 2, logCheckpointMetaManager2) |
| 1567 | require.NoError(t, err) |
| 1568 | require.Nil(t, newSchemaReplaces) |
| 1569 | newSchemaReplaces, err = client.TEST_initSchemasMap(ctx, 2, logCheckpointMetaManager) |
| 1570 | require.NoError(t, err) |
| 1571 | |
| 1572 | require.Equal(t, len(baseTableMappingManager.DBReplaceMap), len(newSchemaReplaces)) |
| 1573 | for _, dbMap := range newSchemaReplaces { |
| 1574 | baseDbMap := baseTableMappingManager.DBReplaceMap[dbMap.IdMap.UpstreamId] |
| 1575 | require.NotNil(t, baseDbMap) |
| 1576 | require.Equal(t, baseDbMap.DbID, dbMap.IdMap.DownstreamId) |
| 1577 | require.Equal(t, baseDbMap.Name, dbMap.Name) |
| 1578 | require.Equal(t, len(baseDbMap.TableMap), len(dbMap.Tables)) |
| 1579 | for _, tableMap := range dbMap.Tables { |
| 1580 | baseTableMap := baseDbMap.TableMap[tableMap.IdMap.UpstreamId] |
| 1581 | require.NotNil(t, baseTableMap) |
| 1582 | require.Equal(t, baseTableMap.TableID, tableMap.IdMap.DownstreamId) |
| 1583 | require.Equal(t, baseTableMap.Name, tableMap.Name) |
| 1584 | require.Equal(t, len(baseTableMap.PartitionMap), len(tableMap.Partitions)) |
| 1585 | for _, partitionMap := range tableMap.Partitions { |
| 1586 | basePartitionMap, exist := baseTableMap.PartitionMap[partitionMap.UpstreamId] |
| 1587 | require.True(t, exist) |
| 1588 | require.Equal(t, basePartitionMap, partitionMap.DownstreamId) |
| 1589 | } |
| 1590 | } |
| 1591 | } |
| 1592 | } |
| 1593 | |
| 1594 | func TestRepairIngestIndex(t *testing.T) { |
nothing calls this directly
no test coverage detected