From 76943c95721d834d3c277a1f785e285a1f88c91d Mon Sep 17 00:00:00 2001 From: nhsmw Date: Wed, 22 Apr 2026 15:06:20 +0800 Subject: [PATCH 1/3] Implement global checkpoint timestamp retrieval Added functionality to retrieve and update global checkpoint timestamp from metadata file. --- cmd/storage-consumer/consumer.go | 51 +++++++++++++++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index ce2fec450b..f8254e12cf 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -47,6 +47,7 @@ const ( defaultChangefeedName = "storage-consumer" defaultLogInterval = 5 * time.Second fakePartitionNumForSchemaFile = -1 + metadataFileName = "metadata" ) type ( @@ -60,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 @@ -79,6 +84,8 @@ type consumer struct { dmlCount atomic.Int64 readSeq atomic.Uint64 + + globalCheckpointTs uint64 } func newConsumer(ctx context.Context) (*consumer, error) { @@ -198,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) { @@ -401,6 +431,13 @@ func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error { if err != nil { return errors.Trace(err) } + if 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 + } data, err := c.externalStorage.ReadFile(ctx, path) if err != nil { return errors.Trace(err) @@ -435,6 +472,13 @@ func (c *consumer) parseSchemaFilePath(ctx context.Context, path string) error { if err != nil { return errors.Trace(err) } + if 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 + } key := schemaKey.GetKey() if tableDefs, ok := c.tableDefMap[key]; ok { if _, ok := tableDefs[schemaKey.TableVersion]; ok { @@ -705,12 +749,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 d0dc9d6887849e2d7e4b8b7e3e6e8f45e41d4b85 Mon Sep 17 00:00:00 2001 From: nhsmw Date: Wed, 22 Apr 2026 15:11:04 +0800 Subject: [PATCH 2/3] Apply suggestions from code review Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --- cmd/storage-consumer/consumer.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index f8254e12cf..a3cf23bad9 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -431,7 +431,7 @@ func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error { if err != nil { return errors.Trace(err) } - if dmlkey.TableVersion > c.globalCheckpointTs { +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), @@ -472,7 +472,7 @@ func (c *consumer) parseSchemaFilePath(ctx context.Context, path string) error { if err != nil { return errors.Trace(err) } - if schemaKey.TableVersion > c.globalCheckpointTs { +if c.globalCheckpointTs > 0 && schemaKey.TableVersion > c.globalCheckpointTs { log.Debug("skip schema file by checkpoint", zap.String("path", path), zap.Uint64("tableVersion", schemaKey.TableVersion), From 5103e0214dc4a62c7b36494b32105e927acef58e Mon Sep 17 00:00:00 2001 From: wk989898 Date: Wed, 22 Apr 2026 07:16:29 +0000 Subject: [PATCH 3/3] fmt Signed-off-by: wk989898 --- cmd/storage-consumer/consumer.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index a3cf23bad9..00020124c4 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -431,7 +431,7 @@ func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error { if err != nil { return errors.Trace(err) } -if c.globalCheckpointTs > 0 && dmlkey.TableVersion > c.globalCheckpointTs { + 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), @@ -472,7 +472,7 @@ func (c *consumer) parseSchemaFilePath(ctx context.Context, path string) error { if err != nil { return errors.Trace(err) } -if c.globalCheckpointTs > 0 && schemaKey.TableVersion > c.globalCheckpointTs { + if c.globalCheckpointTs > 0 && schemaKey.TableVersion > c.globalCheckpointTs { log.Debug("skip schema file by checkpoint", zap.String("path", path), zap.Uint64("tableVersion", schemaKey.TableVersion),