Skip to content

consumer: skip data file by global checkpointTs in storage consumer (#4886) #5939

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ti-chi-bot wants to merge 3 commits into
base: release-8.5
Choose a base branch
from
Open

consumer: skip data file by global checkpointTs in storage consumer (#4886) #5939

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2d9de65
This is an automated cherry-pick of #4886
wk989898 Apr 23, 2026
a7b8e39
fix
wk989898 Aug 10, 2026
266b510
fmt
wk989898 Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 94 additions & 4 deletions cmd/storage-consumer/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,9 @@ import (
"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"
"go.uber.org/atomic"
"go.uber.org/zap"
"golang.org/x/sync/errgroup"
Expand All @@ -44,6 +47,7 @@ import (
const (
defaultChangefeedName = "storage-consumer"
defaultLogInterval = 5 * time.Second
metadataFileName = "metadata"
)

type (
Expand All @@ -57,6 +61,10 @@ type indexRange struct {
end uint64
}

type storageMetadata struct {
CheckpointTs uint64 `json:"checkpoint-ts"`
}

type consumer struct {
replicationCfg *config.ReplicaConfig
codecCfg *common.Config
Expand All @@ -76,6 +84,8 @@ type consumer struct {

dmlCount atomic.Int64
readSeq atomic.Uint64

globalCheckpointTs uint64
}

func newConsumer(ctx context.Context) (*consumer, error) {
Expand Down Expand Up @@ -195,7 +205,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) {
Expand Down Expand Up @@ -397,6 +430,13 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error {
}

func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cloudstorage.DMLPathKey) {
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
}
data, err := c.externalStorage.ReadFile(ctx, path)
if err != nil {
log.Panic("read dml index file failed",
Expand Down Expand Up @@ -424,6 +464,13 @@ func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cl
func (c *consumer) parseSchemaFilePath(ctx context.Context, path string) {
var schemaKey cloudstorage.SchemaPathKey
schemaKey.Parse(path)
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
}
key := schemaKey.GetKey()
if schemaFiles, ok := c.schemaFileMap[key]; ok {
if _, ok := schemaFiles[schemaKey.TableVersion]; ok {
Expand Down Expand Up @@ -501,6 +548,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,
Expand Down Expand Up @@ -558,12 +640,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
}

Expand Down Expand Up @@ -606,7 +689,9 @@ func (c *consumer) handleNewFiles(
}
}
}
c.flushDMLEvents(ctx, tableID)
if err := c.flushDMLEvents(ctx, tableID); err != nil {
return err
}
}

return nil
Expand Down Expand Up @@ -641,12 +726,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)
Expand Down
Loading