pingcap · ti-chi-bot · Mar 21, 2025 · Mar 3, 2025 · Mar 10, 2025 · Mar 10, 2025
diff --git a/go.mod b/go.mod
@@ -5,6 +5,8 @@ go 1.23.4
 toolchain go1.23.7
 
 require (
+	cloud.google.com/go/storage v1.39.1
+	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.16.0
 	github.com/BurntSushi/toml v1.4.1-0.20240526193622-a339e1f7089c
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/IBM/sarama v1.41.2
@@ -64,13 +66,17 @@ require (
 	go.etcd.io/etcd/api/v3 v3.5.12
 	go.etcd.io/etcd/client/pkg/v3 v3.5.12
 	go.etcd.io/etcd/client/v3 v3.5.12
+	go.etcd.io/etcd/pkg/v3 v3.5.12
 	go.etcd.io/etcd/server/v3 v3.5.12
 	go.uber.org/atomic v1.11.0
 	go.uber.org/goleak v1.3.0
+	go.uber.org/mock v0.4.0
+	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/net v0.33.0
 	golang.org/x/oauth2 v0.24.0
 	golang.org/x/sync v0.10.0
+	golang.org/x/sys v0.28.0
 	golang.org/x/term v0.27.0
 	golang.org/x/text v0.21.0
 	golang.org/x/time v0.7.0
@@ -83,11 +89,9 @@ require (
 	cloud.google.com/go/compute/metadata v0.3.0 // indirect
 	cloud.google.com/go/iam v1.1.7 // indirect
 	cloud.google.com/go/kms v1.15.8 // indirect
-	cloud.google.com/go/storage v1.39.1 // indirect
 	github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect
 	github.com/99designs/keyring v1.2.1 // indirect
 	github.com/AthenZ/athenz v1.10.39 // indirect
-	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.16.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.7.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.0.0 // indirect
@@ -311,7 +315,6 @@ require (
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.etcd.io/bbolt v1.3.9 // indirect
 	go.etcd.io/etcd/client/v2 v2.305.12 // indirect
-	go.etcd.io/etcd/pkg/v3 v3.5.12 // indirect
 	go.etcd.io/etcd/raft/v3 v3.5.12 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
@@ -323,13 +326,10 @@ require (
 	go.opentelemetry.io/otel/sdk v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
-	go.uber.org/mock v0.4.0 // indirect
-	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
 	golang.org/x/mod v0.22.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/tools v0.28.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	google.golang.org/api v0.170.0 // indirect

diff --git a/pkg/common/event/redo.go b/pkg/common/event/redo.go
@@ -101,22 +101,14 @@ const (
 func (r *DMLEvent) ToRedoLog() *RedoLog {
 	r.FinishGetRow()
 	row, valid := r.GetNextRow()
-	if !valid {
-		log.Panic("DMLEvent.ToRedoLog must be called with a valid row")
-	}
 	r.FinishGetRow()
 
 	redoLog := &RedoLog{
 		RedoRow: RedoDMLEvent{
 			Row: &DMLEventInRedoLog{
-				StartTs:  r.StartTs,
-				CommitTs: r.CommitTs,
-				Table: &common.TableName{
-					Schema:      r.TableInfo.GetSchemaName(),
-					Table:       r.TableInfo.GetTableName(),
-					TableID:     r.PhysicalTableID,
-					IsPartition: r.TableInfo.IsPartitionTable(),
-				},
+				StartTs:      r.StartTs,
+				CommitTs:     r.CommitTs,
+				Table:        nil,
 				Columns:      nil,
 				PreColumns:   nil,
 				IndexColumns: nil,
@@ -127,60 +119,95 @@ func (r *DMLEvent) ToRedoLog() *RedoLog {
 		Type: RedoLogTypeRow,
    	}
 
-	columnCount := len(r.TableInfo.GetColumns())
-	columns := make([]*RedoColumn, 0, columnCount)
-	switch row.RowType {
-	case RowTypeInsert:
-		redoLog.RedoRow.Columns = make([]RedoColumnValue, 0, columnCount)
-	case RowTypeDelete:
-		redoLog.RedoRow.PreColumns = make([]RedoColumnValue, 0, columnCount)
-	case RowTypeUpdate:
-		redoLog.RedoRow.Columns = make([]RedoColumnValue, 0, columnCount)
-		redoLog.RedoRow.PreColumns = make([]RedoColumnValue, 0, columnCount)
-	default:
-	}
+	if valid && r.TableInfo != nil {
+		redoLog.RedoRow.Row.Table = new(common.TableName)
+		*redoLog.RedoRow.Row.Table = r.TableInfo.TableName
+
+		columnCount := len(r.TableInfo.GetColumns())
+		columns := make([]*RedoColumn, 0, columnCount)
+		switch row.RowType {
+		case RowTypeInsert:
+			redoLog.RedoRow.Columns = make([]RedoColumnValue, 0, columnCount)
+		case RowTypeDelete:
+			redoLog.RedoRow.PreColumns = make([]RedoColumnValue, 0, columnCount)
+		case RowTypeUpdate:
+			redoLog.RedoRow.Columns = make([]RedoColumnValue, 0, columnCount)
+			redoLog.RedoRow.PreColumns = make([]RedoColumnValue, 0, columnCount)
+		default:
+		}
 
-	for i, column := range r.TableInfo.GetColumns() {
-		if common.IsColCDCVisible(column) {
-			columns = append(columns, &RedoColumn{
-				Name:      column.Name.String(),
-				Type:      column.GetType(),
-				Charset:   column.GetCharset(),
-				Collation: column.GetCollate(),
-			})
-			switch row.RowType {
-			case RowTypeInsert:
-				v := parseColumnValue(&row.Row, column, i)
-				redoLog.RedoRow.Columns = append(redoLog.RedoRow.Columns, v)
-			case RowTypeDelete:
-				v := parseColumnValue(&row.PreRow, column, i)
-				redoLog.RedoRow.PreColumns = append(redoLog.RedoRow.PreColumns, v)
-			case RowTypeUpdate:
-				v := parseColumnValue(&row.Row, column, i)
-				redoLog.RedoRow.Columns = append(redoLog.RedoRow.Columns, v)
-				v = parseColumnValue(&row.PreRow, column, i)
-				redoLog.RedoRow.PreColumns = append(redoLog.RedoRow.PreColumns, v)
-			default:
+		for i, column := range r.TableInfo.GetColumns() {
+			if common.IsColCDCVisible(column) {
+				columns = append(columns, &RedoColumn{
+					Name:      column.Name.String(),
+					Type:      column.GetType(),
+					Charset:   column.GetCharset(),
+					Collation: column.GetCollate(),
+				})
+				switch row.RowType {
+				case RowTypeInsert:
+					v := parseColumnValue(&row.Row, column, i)
+					redoLog.RedoRow.Columns = append(redoLog.RedoRow.Columns, v)
+				case RowTypeDelete:
+					v := parseColumnValue(&row.PreRow, column, i)
+					redoLog.RedoRow.PreColumns = append(redoLog.RedoRow.PreColumns, v)
+				case RowTypeUpdate:
+					v := parseColumnValue(&row.Row, column, i)
+					redoLog.RedoRow.Columns = append(redoLog.RedoRow.Columns, v)
+					v = parseColumnValue(&row.PreRow, column, i)
+					redoLog.RedoRow.PreColumns = append(redoLog.RedoRow.PreColumns, v)
+				default:
+				}
 			}
 		}
+		redoLog.RedoRow.Row.Columns = columns
+		redoLog.RedoRow.Row.PreColumns = columns
 	}
-	redoLog.RedoRow.Row.Columns = columns
-	redoLog.RedoRow.Row.PreColumns = columns
 
 	return redoLog
 }
 
 // ToRedoLog converts ddl event to redo log
 func (d *DDLEvent) ToRedoLog() *RedoLog {
-	ddlInRedoLog := &DDLEventInRedoLog{
-		StartTs:  d.GetStartTs(),
-		CommitTs: d.GetCommitTs(),
-		Query:    d.Query,
+	redoLog := &RedoLog{
+		RedoDDL: RedoDDLEvent{
+			DDL: &DDLEventInRedoLog{
+				StartTs:  d.GetStartTs(),
+				CommitTs: d.GetCommitTs(),
+				Query:    d.Query,
+			},
+			Type: d.Type,
+		},
+		Type: RedoLogTypeDDL,
 	}
-	return &RedoLog{
-		RedoDDL: RedoDDLEvent{DDL: ddlInRedoLog},
-		Type:    RedoLogTypeDDL,
+	if d.TableInfo != nil {
+		redoLog.RedoDDL.TableName = d.TableInfo.TableName
 	}
+
+	return redoLog
+}
+
+// GetCommitTs returns commit timestamp of the log event.
+func (r *RedoLog) GetCommitTs() common.Ts {
+	switch r.Type {
+	case RedoLogTypeRow:
+		return r.RedoRow.Row.CommitTs
+	case RedoLogTypeDDL:
+		return r.RedoDDL.DDL.CommitTs
+	default:
+		log.Panic("Unexpected redo log type")
+		return 0
+	}
+}
+
+// IsDelete checks whether it's a deletion or not.
+func (r RedoDMLEvent) IsDelete() bool {
+	return len(r.Row.PreColumns) > 0 && len(r.Row.Columns) == 0
+}
+
+// IsUpdate checks whether it's a update or not.
+func (r RedoDMLEvent) IsUpdate() bool {
+	return len(r.Row.PreColumns) > 0 && len(r.Row.Columns) > 0
 }
 
 func parseColumnValue(row *chunk.Row, column *model.ColumnInfo, i int) RedoColumnValue {

diff --git a/pkg/common/event/redo_gen_compatibility_test.go b/pkg/common/event/redo_gen_compatibility_test.go
@@ -18,9 +18,9 @@ import (
 
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/types"
 	parserModel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/chunk"
 	oldArchModel "github.com/pingcap/tiflow/cdc/model"
 	"github.com/stretchr/testify/require"
@@ -30,26 +30,26 @@ func TestDMLCompatibility(t *testing.T) {
 	tableInfo := &model.TableInfo{
 		Columns: []*model.ColumnInfo{
 			&model.ColumnInfo{
-				Name:                parserModel.NewCIStr("c1"),
-				Offset:              0,
+				Name: parserModel.NewCIStr("c1"),
+				Offset: 0,
 				GeneratedExprString: "xxx",
-				FieldType:           *types.NewFieldType(mysql.TypeString),
+				FieldType: *types.NewFieldType(mysql.TypeString),
 			},
 			&model.ColumnInfo{
-				Name:                parserModel.NewCIStr("c2"),
-				Offset:              1,
+				Name: parserModel.NewCIStr("c2"),
+				Offset: 1,
 				GeneratedExprString: "",
-				FieldType:           *types.NewFieldType(mysql.TypeString),
+				FieldType: *types.NewFieldType(mysql.TypeString),
 			},
 		},
 	}
 
-	rowEvent := &DMLEvent{
-		StartTs:   100,
-		CommitTs:  200,
-		TableInfo: common.WrapTableInfo("test", tableInfo),
+	rowEvent := &DMLEvent {
+		StartTs: 100,
+		CommitTs: 200,
+		TableInfo: common.WrapTableInfo(1, "test", tableInfo),
 		Rows: chunk.NewEmptyChunk([]*types.FieldType{
-			&tableInfo.Columns[0].FieldType,
+			&tableInfo.Columns[0].FieldType, 
 			&tableInfo.Columns[1].FieldType,
 		}),
 		RowTypes: []RowType{RowTypeUpdate},

diff --git a/pkg/common/types.go b/pkg/common/types.go
@@ -31,8 +31,9 @@ const (
 var DefaultEndian = binary.LittleEndian
 
 type (
-	Ts      = uint64
-	TableID = int64
+	Ts        = uint64
+	TableID   = int64
+	CaptureID = string
 )
 
 type CoordinatorID string

diff --git a/pkg/errors/error.go b/pkg/errors/error.go
@@ -486,4 +486,46 @@ var (
 		"user %s unauthorized, error: %s",
 		errors.RFCCodeText("CDC:ErrUnauthorized"),
 	)
+
+	ErrExternalStorageAPI = errors.Normalize(
+		"external storage api",
+		errors.RFCCodeText("CDC:ErrS3StorageAPI"),
+	)
+	ErrConsistentStorage = errors.Normalize(
+		"consistent storage (%s) not support",
+		errors.RFCCodeText("CDC:ErrConsistentStorage"),
+	)
+	ErrStorageInitialize = errors.Normalize(
+		"fail to open storage for redo log",
+		errors.RFCCodeText("CDC:ErrStorageInitialize"),
+	)
+
+	ErrRedoConfigInvalid = errors.Normalize(
+		"redo log config invalid",
+		errors.RFCCodeText("CDC:ErrRedoConfigInvalid"),
+	)
+	ErrRedoDownloadFailed = errors.Normalize(
+		"redo log down load to local failed",
+		errors.RFCCodeText("CDC:ErrRedoDownloadFailed"),
+	)
+	ErrRedoWriterStopped = errors.Normalize(
+		"redo log writer stopped",
+		errors.RFCCodeText("CDC:ErrRedoWriterStopped"),
+	)
+	ErrRedoFileOp = errors.Normalize(
+		"redo file operation",
+		errors.RFCCodeText("CDC:ErrRedoFileOp"),
+	)
+	ErrRedoFileSizeExceed = errors.Normalize(
+		"redo file size %d exceeds maximum %d",
+		errors.RFCCodeText("CDC:ErrRedoFileSizeExceed"),
+	)
+	ErrRedoMetaFileNotFound = errors.Normalize(
+		"no redo meta file found in dir: %s",
+		errors.RFCCodeText("CDC:ErrRedoMetaFileNotFound"),
+	)
+	ErrRedoMetaInitialize = errors.Normalize(
+		"initialize meta for redo log",
+		errors.RFCCodeText("CDC:ErrRedoMetaInitialize"),
+	)
 )
diff --git a/pkg/fsutil/disk_info_freebsd.go b/pkg/fsutil/disk_info_freebsd.go
@@ -0,0 +1,54 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+//go:build freebsd
+
+package fsutil
+
+import (
+	"os"
+	"path/filepath"
+	"syscall"
+
+	cerror "github.com/pingcap/tiflow/pkg/errors"
+)
+
+// GetDiskInfo return the disk space information of the given directory
+// the caller should guarantee that dir exist
+func GetDiskInfo(dir string) (*DiskInfo, error) {
+	f := filepath.Join(dir, "file.test")
+	if err := os.WriteFile(f, []byte(""), 0o600); err != nil {
+		return nil, cerror.WrapError(cerror.ErrGetDiskInfo, err)
+	}
+
+	fs := syscall.Statfs_t{}
+	if err := syscall.Statfs(dir, &fs); err != nil {
+		return nil, cerror.WrapError(cerror.ErrGetDiskInfo, err)
+	}
+
+	info := &DiskInfo{
+		All:   fs.Blocks * uint64(fs.Bsize) / gb,
+		Avail: uint64(fs.Bavail) * uint64(fs.Bsize) / gb,
+		Free:  fs.Bfree * uint64(fs.Bsize) / gb,
+	}
+	info.Used = info.All - info.Free
+	info.AvailPercentage = float32(info.Avail) / float32(info.All) * 100
+
+	if err := os.Remove(f); err != nil {
+		if !os.IsNotExist(err) {
+			return info, cerror.WrapError(cerror.ErrGetDiskInfo, err)
+		}
+	}
+
+	return info, nil
+}