sink: add checkpointTs message for storage sink

coordinator/changefeed/changefeed.go

@@ -35,7 +35,7 @@ import (
 type Changefeed struct {
 	ID       common.ChangeFeedID
 	info     *atomic.Pointer[config.ChangeFeedInfo]
-	isMQSink bool
+	sinkType common.SinkType
 	isNew    bool // only true when the changefeed is newly created or resumed by overwriteCheckpointTs
 
 	// nodeIDMu protects nodeID
@@ -73,7 +73,7 @@ func NewChangefeed(cfID common.ChangeFeedID,
 		info:                  atomic.NewPointer(info),
 		configBytes:           bytes,
 		lastSavedCheckpointTs: atomic.NewUint64(checkpointTs),
-		isMQSink:              sink.IsMQScheme(uri.Scheme),
+		sinkType:              getSinkType(uri.Scheme),
 		isNew:                 isNew,
 		// Initialize the status
 		status: atomic.NewPointer(
@@ -167,8 +167,8 @@ func (c *Changefeed) ForceUpdateStatus(newStatus *heartbeatpb.MaintainerStatus)
 	return c.backoff.CheckStatus(newStatus)
 }
 
-func (c *Changefeed) IsMQSink() bool {
-	return c.isMQSink
+func (c *Changefeed) NeedCheckpointTsMessage() bool {
+	return c.sinkType == common.KafkaSinkType || c.sinkType == common.CloudStorageSinkType
 }
 
 func (c *Changefeed) SetIsNew(isNew bool) {
@@ -259,3 +259,17 @@ func RemoveMaintainerMessage(id common.ChangeFeedID, server node.ID, caseCade bo
 			Removed: removed,
 		})
 }
+
+// getSinkType returns the sink type of the url.
+func getSinkType(scheme string) common.SinkType {
+	if sink.IsMySQLCompatibleScheme(scheme) {
+		return common.MysqlSinkType
+	}
+	if sink.IsMQScheme(scheme) {
+		return common.KafkaSinkType
+	}
+	if sink.IsStorageScheme(scheme) {
+		return common.CloudStorageSinkType
+	}
+	return common.BlackHoleSinkType
+}

coordinator/changefeed/changefeed_db_test.go

@@ -232,7 +232,7 @@ func TestReplaceStoppedChangefeed(t *testing.T) {
 	require.Contains(t, db.stopped, cf.ID)
 
 	cf3 := db.GetByID(cf.ID)
-	require.Equal(t, true, cf3.isMQSink)
+	require.Equal(t, true, cf3.NeedCheckpointTsMessage())
 
 	cf4ID := common.NewChangeFeedIDWithName("test4")
 	cf4 := &config.ChangeFeedInfo{

coordinator/changefeed/changefeed_test.go

@@ -39,7 +39,7 @@ func TestNewChangefeed(t *testing.T) {
 	require.Equal(t, cfID, cf.ID)
 	require.Equal(t, info, cf.GetInfo())
 	require.Equal(t, checkpointTs, cf.GetLastSavedCheckPointTs())
-	require.True(t, cf.IsMQSink())
+	require.True(t, cf.NeedCheckpointTsMessage())
 }
 
 func TestChangefeed_GetSetInfo(t *testing.T) {
@@ -100,7 +100,7 @@ func TestChangefeed_IsMQSink(t *testing.T) {
 	}
 	cf := NewChangefeed(cfID, info, 100, true)
 
-	require.True(t, cf.IsMQSink())
+	require.True(t, cf.NeedCheckpointTsMessage())
 }
 
 func TestChangefeed_GetSetLastSavedCheckPointTs(t *testing.T) {

coordinator/controller_test.go

@@ -125,11 +125,11 @@ func TestUpdateChangefeed(t *testing.T) {
 
 	backend.EXPECT().UpdateChangefeed(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(errors.New("failed")).Times(1)
 	require.NotNil(t, controller.UpdateChangefeed(context.Background(), newConfig))
-	require.Equal(t, false, changefeedDB.GetByID(cfID).IsMQSink())
+	require.Equal(t, false, changefeedDB.GetByID(cfID).NeedCheckpointTsMessage())
 
 	backend.EXPECT().UpdateChangefeed(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
 	require.Nil(t, controller.UpdateChangefeed(context.Background(), newConfig))
-	require.Equal(t, true, changefeedDB.GetByID(cfID).IsMQSink())
+	require.Equal(t, true, changefeedDB.GetByID(cfID).NeedCheckpointTsMessage())
 	require.Equal(t, 1, changefeedDB.GetStoppedSize())
 }
 

coordinator/coordinator.go

@@ -340,7 +340,7 @@ func (c *coordinator) saveCheckpointTs(ctx context.Context, cfs map[common.Chang
 			continue
 		}
 		cf.SetLastSavedCheckPointTs(cp)
-		if cf.IsMQSink() {
+		if cf.NeedCheckpointTsMessage() {
 			msg := cf.NewCheckpointTsMessage(cf.GetLastSavedCheckPointTs())
 			c.sendMessages([]*messaging.TargetMessage{msg})
 		}

pkg/sink/cloudstorage/path_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/config"
@@ -39,7 +40,7 @@ import (
 
 func testFilePathGenerator(ctx context.Context, t *testing.T, dir string) *FilePathGenerator {
 	uri := fmt.Sprintf("file:///%s?flush-interval=2s", dir)
-	storage, err := util.GetExternalStorageFromURI(ctx, uri)
+	storage, err := helper.GetExternalStorageFromURI(ctx, uri)
 	require.NoError(t, err)
 
 	sinkURI, err := url.Parse(uri)
@@ -353,7 +354,7 @@ func TestRemoveExpiredFilesWithoutPartition(t *testing.T) {
 	defer cancel()
 	dir := t.TempDir()
 	uri := fmt.Sprintf("file:///%s?flush-interval=2s", dir)
-	storage, err := util.GetExternalStorageFromURI(ctx, uri)
+	storage, err := helper.GetExternalStorageFromURI(ctx, uri)
 	require.NoError(t, err)
 	sinkURI, err := url.Parse(uri)
 	require.NoError(t, err)

pkg/sink/codec/csv/csv_message.go

@@ -25,7 +25,6 @@ import (
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
-	"github.com/pingcap/tidb/pkg/kv"
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/types"
@@ -87,7 +86,7 @@ type csvMessage struct {
 	preColumns []any
 	// newRecord indicates whether we encounter a new record.
 	newRecord bool
-	HandleKey kv.Handle
+	// HandleKey kv.Handle
 }
 
 func newCSVMessage(config *common.Config) *csvMessage {
@@ -138,7 +137,8 @@ func (c *csvMessage) encodeMeta(opType string, b *strings.Builder) {
 		}
 	}
 	if c.config.OutputHandleKey {
-		c.formatValue(c.HandleKey.String(), b)
+		log.Warn("not support output handle key")
+		// c.formatValue(c.HandleKey.String(), b)
 	}
 }
 
@@ -355,33 +355,31 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe
 		if err != nil {
 			return nil, err
 		}
+	} else if e.IsInsert() {
+		// This is a insert operation.
+		csvMsg.opType = operationInsert
+		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), e.TableInfo)
+		if err != nil {
+			return nil, err
+		}
 	} else {
-		if e.GetPreRows() == nil {
-			// This is a insert operation.
-			csvMsg.opType = operationInsert
-			csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), e.TableInfo)
-			if err != nil {
-				return nil, err
+		// This is a update operation.
+		csvMsg.opType = operationUpdate
+		if csvConfig.OutputOldValue {
+			if e.GetPreRows().Len() != e.GetRows().Len() {
+				return nil, cerror.WrapError(cerror.ErrCSVDecodeFailed,
+					fmt.Errorf("the column length of preColumns %d doesn't equal to that of columns %d",
+						e.GetPreRows().Len(), e.GetRows().Len()))
 			}
-		} else {
-			// This is a update operation.
-			csvMsg.opType = operationUpdate
-			if csvConfig.OutputOldValue {
-				if e.GetPreRows().Len() != e.GetRows().Len() {
-					return nil, cerror.WrapError(cerror.ErrCSVDecodeFailed,
-						fmt.Errorf("the column length of preColumns %d doesn't equal to that of columns %d",
-							e.GetPreRows().Len(), e.GetRows().Len()))
-				}
-				csvMsg.preColumns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), e.TableInfo)
-				if err != nil {
-					return nil, err
-				}
-			}
-			csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), e.TableInfo)
+			csvMsg.preColumns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), e.TableInfo)
 			if err != nil {
 				return nil, err
 			}
 		}
+		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), e.TableInfo)
+		if err != nil {
+			return nil, err
+		}
 	}
 	return csvMsg, nil
 }