From 2d9de650f85364ba40fbd6af8beaa504625a83bc Mon Sep 17 00:00:00 2001 From: nhsmw Date: Thu, 23 Apr 2026 13:56:44 +0800 Subject: [PATCH 1/3] This is an automated cherry-pick of #4886 Signed-off-by: ti-chi-bot --- cmd/storage-consumer/consumer.go | 76 +++++++++++++++++++++++++++++++- 1 file changed, 75 insertions(+), 1 deletion(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 05ec67cf4b..c7e1a82866 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -42,8 +42,15 @@ import ( ) const ( +<<<<<<< HEAD defaultChangefeedName = "storage-consumer" defaultLogInterval = 5 * time.Second +======= + defaultChangefeedName = "storage-consumer" + defaultLogInterval = 5 * time.Second + fakePartitionNumForSchemaFile = -1 + metadataFileName = "metadata" +>>>>>>> 170515398 (consumer: skip data file by global checkpointTs in storage consumer (#4886)) ) type ( @@ -57,6 +64,10 @@ type indexRange struct { end uint64 } +type storageMetadata struct { + CheckpointTs uint64 `json:"checkpoint-ts"` +} + type consumer struct { replicationCfg *config.ReplicaConfig codecCfg *common.Config @@ -76,6 +87,8 @@ type consumer struct { dmlCount atomic.Int64 readSeq atomic.Uint64 + + globalCheckpointTs uint64 } func newConsumer(ctx context.Context) (*consumer, error) { @@ -195,7 +208,30 @@ func diffDMLMaps( return resMap } -// getNewFiles returns newly created dml files in specific ranges +func (c *consumer) getGlobalCheckpointTs(ctx context.Context) error { + exists, err := c.externalStorage.FileExists(ctx, metadataFileName) + if err != nil { + return errors.Trace(err) + } + if !exists { + return nil + } + + data, err := c.externalStorage.ReadFile(ctx, metadataFileName) + if err != nil { + return errors.Trace(err) + } + var metadata storageMetadata + if err := json.Unmarshal(data, &metadata); err != nil { + return errors.Trace(err) + } + if metadata.CheckpointTs > c.globalCheckpointTs { + c.globalCheckpointTs = metadata.CheckpointTs + } + return nil +} + +// getNewFiles returns newly created dml files in specific ranges that are visible under checkpointTs. func (c *consumer) getNewFiles( ctx context.Context, ) (map[cloudstorage.DMLPathKey]fileIndexRange, error) { @@ -396,7 +432,26 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { } } +<<<<<<< HEAD func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cloudstorage.DMLPathKey) { +======= +func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error { + var dmlkey cloudstorage.DmlPathKey + dispatcherID, err := dmlkey.ParseIndexFilePath( + putil.GetOrZero(c.replicationCfg.Sink.DateSeparator), + path, + ) + if err != nil { + return errors.Trace(err) + } + if c.globalCheckpointTs > 0 && dmlkey.TableVersion > c.globalCheckpointTs { + log.Debug("skip dml index file by checkpoint", + zap.String("path", path), + zap.Uint64("tableVersion", dmlkey.TableVersion), + zap.Uint64("checkpointTs", c.globalCheckpointTs)) + return nil + } +>>>>>>> 170515398 (consumer: skip data file by global checkpointTs in storage consumer (#4886)) data, err := c.externalStorage.ReadFile(ctx, path) if err != nil { log.Panic("read dml index file failed", @@ -423,7 +478,21 @@ func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cl func (c *consumer) parseSchemaFilePath(ctx context.Context, path string) { var schemaKey cloudstorage.SchemaPathKey +<<<<<<< HEAD schemaKey.Parse(path) +======= + checksumInFile, err := schemaKey.ParseSchemaFilePath(path) + if err != nil { + return errors.Trace(err) + } + if c.globalCheckpointTs > 0 && schemaKey.TableVersion > c.globalCheckpointTs { + log.Debug("skip schema file by checkpoint", + zap.String("path", path), + zap.Uint64("tableVersion", schemaKey.TableVersion), + zap.Uint64("checkpointTs", c.globalCheckpointTs)) + return nil + } +>>>>>>> 170515398 (consumer: skip data file by global checkpointTs in storage consumer (#4886)) key := schemaKey.GetKey() if schemaFiles, ok := c.schemaFileMap[key]; ok { if _, ok := schemaFiles[schemaKey.TableVersion]; ok { @@ -641,12 +710,17 @@ func (c *consumer) handle(ctx context.Context) error { } round++ + err := c.getGlobalCheckpointTs(ctx) + if err != nil { + return errors.Trace(err) + } dmlFileMap, err := c.getNewFiles(ctx) if err != nil { return errors.Trace(err) } log.Info("storage consumer scan done", zap.Uint64("round", round), + zap.Uint64("checkpointTs", c.globalCheckpointTs), zap.Int("dmlPathKeyCount", len(dmlFileMap))) err = c.handleNewFiles(ctx, dmlFileMap, round) From a7b8e39331eb35b158d92cfcede2cd2a35194e8b Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 10 Aug 2026 09:43:45 +0000 Subject: [PATCH 2/3] fix Signed-off-by: wk989898 --- cmd/storage-consumer/consumer.go | 79 +++++++++++++++++++------------- 1 file changed, 48 insertions(+), 31 deletions(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index c7e1a82866..e52c39d31b 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -35,6 +35,10 @@ import ( "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/codec/csv" putil "github.com/pingcap/ticdc/pkg/util" + timodel "github.com/pingcap/tidb/pkg/meta/model" + "github.com/pingcap/tidb/pkg/parser" + "github.com/pingcap/tidb/pkg/parser/ast" + "github.com/pingcap/tidb/br/pkg/storage" "go.uber.org/atomic" "go.uber.org/zap" @@ -42,15 +46,9 @@ import ( ) const ( -<<<<<<< HEAD defaultChangefeedName = "storage-consumer" defaultLogInterval = 5 * time.Second -======= - defaultChangefeedName = "storage-consumer" - defaultLogInterval = 5 * time.Second - fakePartitionNumForSchemaFile = -1 - metadataFileName = "metadata" ->>>>>>> 170515398 (consumer: skip data file by global checkpointTs in storage consumer (#4886)) + metadataFileName = "metadata" ) type ( @@ -432,26 +430,14 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { } } -<<<<<<< HEAD func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cloudstorage.DMLPathKey) { -======= -func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error { - var dmlkey cloudstorage.DmlPathKey - dispatcherID, err := dmlkey.ParseIndexFilePath( - putil.GetOrZero(c.replicationCfg.Sink.DateSeparator), - path, - ) - if err != nil { - return errors.Trace(err) - } if c.globalCheckpointTs > 0 && dmlkey.TableVersion > c.globalCheckpointTs { log.Debug("skip dml index file by checkpoint", zap.String("path", path), zap.Uint64("tableVersion", dmlkey.TableVersion), zap.Uint64("checkpointTs", c.globalCheckpointTs)) - return nil + return } ->>>>>>> 170515398 (consumer: skip data file by global checkpointTs in storage consumer (#4886)) data, err := c.externalStorage.ReadFile(ctx, path) if err != nil { log.Panic("read dml index file failed", @@ -478,21 +464,14 @@ func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error { func (c *consumer) parseSchemaFilePath(ctx context.Context, path string) { var schemaKey cloudstorage.SchemaPathKey -<<<<<<< HEAD schemaKey.Parse(path) -======= - checksumInFile, err := schemaKey.ParseSchemaFilePath(path) - if err != nil { - return errors.Trace(err) - } if c.globalCheckpointTs > 0 && schemaKey.TableVersion > c.globalCheckpointTs { log.Debug("skip schema file by checkpoint", zap.String("path", path), zap.Uint64("tableVersion", schemaKey.TableVersion), zap.Uint64("checkpointTs", c.globalCheckpointTs)) - return nil + return } ->>>>>>> 170515398 (consumer: skip data file by global checkpointTs in storage consumer (#4886)) key := schemaKey.GetKey() if schemaFiles, ok := c.schemaFileMap[key]; ok { if _, ok := schemaFiles[schemaKey.TableVersion]; ok { @@ -570,6 +549,41 @@ func (c *consumer) mustGetSchemaFile(key cloudstorage.SchemaPathKey) cloudstorag return *schemaFile } +func getRenameTableOldTableKey(schemaFile cloudstorage.SchemaFile) (string, bool) { + if schemaFile.Type != byte(timodel.ActionRenameTable) { + return "", false + } + schemaName := schemaFile.Schema + stmt, err := parser.New().ParseOneStmt(schemaFile.Query, "", "") + if err != nil { + log.Panic("parse statement failed", zap.Any("DDL", schemaFile.Query), zap.Error(err)) + } + // The query in job maybe "RENAME TABLE table1 to table2" + renameStmt, ok := stmt.(*ast.RenameTableStmt) + if !ok || len(renameStmt.TableToTables) == 0 { + log.Panic("invalid rename table statement", zap.Any("DDL", schemaFile.Query)) + } + oldTable := renameStmt.TableToTables[0].OldTable + if oldTable.Schema.O != "" { + schemaName = oldTable.Schema.O + } + tableName := oldTable.Name.O + return commonType.QuoteSchema(schemaName, tableName), true +} + +func (c *consumer) updateTableDDLWatermark(schemaFile cloudstorage.SchemaFile) string { + key := commonType.QuoteSchema(schemaFile.Schema, schemaFile.Table) + if c.tableDDLWatermark[key] < schemaFile.TableVersion { + c.tableDDLWatermark[key] = schemaFile.TableVersion + } + if oldTableKey, ok := getRenameTableOldTableKey(schemaFile); ok { + if c.tableDDLWatermark[oldTableKey] < schemaFile.TableVersion { + c.tableDDLWatermark[oldTableKey] = schemaFile.TableVersion + } + } + return key +} + func (c *consumer) handleNewFiles( ctx context.Context, dmlFileMap map[cloudstorage.DMLPathKey]fileIndexRange, @@ -627,12 +641,13 @@ func (c *consumer) handleNewFiles( if err := c.sink.WriteBlockEvent(ddlEvent); err != nil { return errors.Trace(err) } - c.tableDDLWatermark[tableKey] = key.TableVersion + watermarkKey := c.updateTableDDLWatermark(schemaFile) // TODO: need to cleanup schemaFileMap in the future. log.Info("execute ddl event successfully", zap.String("query", schemaFile.Query), zap.String("schema", key.Schema), zap.String("table", key.Table), - zap.Uint64("ddlWatermark", c.tableDDLWatermark[tableKey])) + zap.Uint64("ddlWatermark", c.tableDDLWatermark[tableKey]), + zap.String("watermarkKey", watermarkKey)) continue } @@ -675,7 +690,9 @@ func (c *consumer) handleNewFiles( } } } - c.flushDMLEvents(ctx, tableID) + if err := c.flushDMLEvents(ctx, tableID); err != nil { + return err + } } return nil From 266b51079744bd153ca2e03e01f1c2bcc00a88b9 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 10 Aug 2026 09:46:41 +0000 Subject: [PATCH 3/3] fmt Signed-off-by: wk989898 --- cmd/storage-consumer/consumer.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index e52c39d31b..399a01e28d 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -35,11 +35,10 @@ import ( "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/codec/csv" putil "github.com/pingcap/ticdc/pkg/util" + "github.com/pingcap/tidb/br/pkg/storage" timodel "github.com/pingcap/tidb/pkg/meta/model" "github.com/pingcap/tidb/pkg/parser" "github.com/pingcap/tidb/pkg/parser/ast" - - "github.com/pingcap/tidb/br/pkg/storage" "go.uber.org/atomic" "go.uber.org/zap" "golang.org/x/sync/errgroup"