sink(ticdc): add hex encode to csv (#9366)

close #9373

cdc/api/v2/model.go

@@ -266,10 +266,11 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 		var csvConfig *config.CSVConfig
 		if c.Sink.CSVConfig != nil {
 			csvConfig = &config.CSVConfig{
-				Delimiter:       c.Sink.CSVConfig.Delimiter,
-				Quote:           c.Sink.CSVConfig.Quote,
-				NullString:      c.Sink.CSVConfig.NullString,
-				IncludeCommitTs: c.Sink.CSVConfig.IncludeCommitTs,
+				Delimiter:            c.Sink.CSVConfig.Delimiter,
+				Quote:                c.Sink.CSVConfig.Quote,
+				NullString:           c.Sink.CSVConfig.NullString,
+				IncludeCommitTs:      c.Sink.CSVConfig.IncludeCommitTs,
+				BinaryEncodingMethod: c.Sink.CSVConfig.BinaryEncodingMethod,
 			}
 		}
 		var kafkaConfig *config.KafkaConfig
@@ -491,10 +492,11 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		var csvConfig *CSVConfig
 		if cloned.Sink.CSVConfig != nil {
 			csvConfig = &CSVConfig{
-				Delimiter:       cloned.Sink.CSVConfig.Delimiter,
-				Quote:           cloned.Sink.CSVConfig.Quote,
-				NullString:      cloned.Sink.CSVConfig.NullString,
-				IncludeCommitTs: cloned.Sink.CSVConfig.IncludeCommitTs,
+				Delimiter:            cloned.Sink.CSVConfig.Delimiter,
+				Quote:                cloned.Sink.CSVConfig.Quote,
+				NullString:           cloned.Sink.CSVConfig.NullString,
+				IncludeCommitTs:      cloned.Sink.CSVConfig.IncludeCommitTs,
+				BinaryEncodingMethod: cloned.Sink.CSVConfig.BinaryEncodingMethod,
 			}
 		}
 		var kafkaConfig *KafkaConfig
@@ -766,10 +768,11 @@ type SinkConfig struct {
 // CSVConfig denotes the csv config
 // This is the same as config.CSVConfig
 type CSVConfig struct {
-	Delimiter       string `json:"delimiter"`
-	Quote           string `json:"quote"`
-	NullString      string `json:"null"`
-	IncludeCommitTs bool   `json:"include_commit_ts"`
+	Delimiter            string `json:"delimiter"`
+	Quote                string `json:"quote"`
+	NullString           string `json:"null"`
+	IncludeCommitTs      bool   `json:"include_commit_ts"`
+	BinaryEncodingMethod string `json:"binary_encoding_method"`
 }
 
 // LargeMessageHandleConfig denotes the large message handling config

cdc/api/v2/model_test.go

@@ -45,9 +45,10 @@ var defaultAPIConfig = &ReplicaConfig{
 	},
 	Sink: &SinkConfig{
 		CSVConfig: &CSVConfig{
-			Quote:      string(config.DoubleQuoteChar),
-			Delimiter:  config.Comma,
-			NullString: config.NULL,
+			Quote:                string(config.DoubleQuoteChar),
+			Delimiter:            config.Comma,
+			NullString:           config.NULL,
+			BinaryEncodingMethod: config.BinaryEncodingBase64,
 		},
 		EncoderConcurrency:               util.AddressOf(16),
 		Terminator:                       util.AddressOf(config.CRLF),

docs/swagger/docs.go

@@ -1345,6 +1345,10 @@ var doc = `{
         "config.CSVConfig": {
             "type": "object",
             "properties": {
+                "binary-encoding-method": {
+                    "description": "encoding method of binary type",
+                    "type": "string"
+                },
                 "delimiter": {
                     "description": "delimiter between fields",
                     "type": "string"
@@ -1977,6 +1981,9 @@ var doc = `{
         "v2.CSVConfig": {
             "type": "object",
             "properties": {
+                "binary_encoding_method": {
+                    "type": "string"
+                },
                 "delimiter": {
                     "type": "string"
                 },

docs/swagger/swagger.json

@@ -1326,6 +1326,10 @@
         "config.CSVConfig": {
             "type": "object",
             "properties": {
+                "binary-encoding-method": {
+                    "description": "encoding method of binary type",
+                    "type": "string"
+                },
                 "delimiter": {
                     "description": "delimiter between fields",
                     "type": "string"
@@ -1958,6 +1962,9 @@
         "v2.CSVConfig": {
             "type": "object",
             "properties": {
+                "binary_encoding_method": {
+                    "type": "string"
+                },
                 "delimiter": {
                     "type": "string"
                 },

docs/swagger/swagger.yaml

@@ -1,6 +1,9 @@
 definitions:
   config.CSVConfig:
     properties:
+      binary-encoding-method:
+        description: encoding method of binary type
+        type: string
       delimiter:
         description: delimiter between fields
         type: string
@@ -435,6 +438,8 @@ definitions:
     type: object
   v2.CSVConfig:
     properties:
+      binary_encoding_method:
+        type: string
       delimiter:
         type: string
       include_commit_ts:

pkg/cmd/util/helper_test.go

@@ -201,9 +201,10 @@ func TestAndWriteExampleReplicaTOML(t *testing.T) {
 			{Matcher: []string{"test3.*", "test4.*"}, Columns: []string{"!a", "column3"}},
 		},
 		CSVConfig: &config.CSVConfig{
-			Quote:      string(config.DoubleQuoteChar),
-			Delimiter:  string(config.Comma),
-			NullString: config.NULL,
+			Quote:                string(config.DoubleQuoteChar),
+			Delimiter:            string(config.Comma),
+			NullString:           config.NULL,
+			BinaryEncodingMethod: config.BinaryEncodingBase64,
 		},
 		Terminator:                       util.AddressOf("\r\n"),
 		DateSeparator:                    util.AddressOf(config.DateSeparatorDay.String()),
@@ -236,10 +237,11 @@ func TestAndWriteStorageSinkTOML(t *testing.T) {
 		FileIndexWidth:           util.AddressOf(config.DefaultFileIndexWidth),
 		EnableKafkaSinkV2:        util.AddressOf(false),
 		CSVConfig: &config.CSVConfig{
-			Delimiter:       ",",
-			Quote:           "\"",
-			NullString:      "\\N",
-			IncludeCommitTs: false,
+			Delimiter:            ",",
+			Quote:                "\"",
+			NullString:           "\\N",
+			IncludeCommitTs:      false,
+			BinaryEncodingMethod: config.BinaryEncodingBase64,
 		},
 		OnlyOutputUpdatedColumns:         util.AddressOf(false),
 		DeleteOnlyOutputHandleKeyColumns: util.AddressOf(false),

pkg/config/config_test_data.go

@@ -203,7 +203,8 @@ const (
       "delimiter": ",",
       "quote": "\"",
       "null": "\\N",
-      "include-commit-ts": true
+      "include-commit-ts": true,
+      "binary-encoding-method":"base64"
     },
     "date-separator": "month",
     "enable-partition-separator": true,
@@ -334,7 +335,8 @@ const (
       "delimiter": ",",
       "quote": "\"",
       "null": "\\N",
-      "include-commit-ts": true
+      "include-commit-ts": true,
+      "binary-encoding-method":"base64"
     },
     "terminator": "\r\n",
 	"transaction-atomicity": "",

pkg/config/replica_config.go

@@ -55,9 +55,10 @@ var defaultReplicaConfig = &ReplicaConfig{
 	},
 	Sink: &SinkConfig{
 		CSVConfig: &CSVConfig{
-			Quote:      string(DoubleQuoteChar),
-			Delimiter:  Comma,
-			NullString: NULL,
+			Quote:                string(DoubleQuoteChar),
+			Delimiter:            Comma,
+			NullString:           NULL,
+			BinaryEncodingMethod: BinaryEncodingBase64,
 		},
 		EncoderConcurrency:               util.AddressOf(16),
 		Terminator:                       util.AddressOf(CRLF),

pkg/config/replica_config_test.go

@@ -49,10 +49,11 @@ func TestReplicaConfigMarshal(t *testing.T) {
 		},
 	}
 	conf.Sink.CSVConfig = &CSVConfig{
-		Delimiter:       ",",
-		Quote:           "\"",
-		NullString:      `\N`,
-		IncludeCommitTs: true,
+		Delimiter:            ",",
+		Quote:                "\"",
+		NullString:           `\N`,
+		IncludeCommitTs:      true,
+		BinaryEncodingMethod: BinaryEncodingBase64,
 	}
 	conf.Sink.TxnAtomicity = util.AddressOf(unknownTxnAtomicity)
 	conf.Sink.DateSeparator = util.AddressOf("month")

pkg/config/sink.go

@@ -63,6 +63,11 @@ const (
 	MaxFileIndexWidth = 20 // enough for 2^64 files
 	// DefaultFileIndexWidth is the default width of file index.
 	DefaultFileIndexWidth = MaxFileIndexWidth
+
+	// BinaryEncodingHex encodes binary data to hex string.
+	BinaryEncodingHex = "hex"
+	// BinaryEncodingBase64 encodes binary data to base64 string.
+	BinaryEncodingBase64 = "base64"
 )
 
 // AtomicityLevel represents the atomicity level of a changefeed.
@@ -164,6 +169,8 @@ type CSVConfig struct {
 	NullString string `toml:"null" json:"null"`
 	// whether to include commit ts
 	IncludeCommitTs bool `toml:"include-commit-ts" json:"include-commit-ts"`
+	// encoding method of binary type
+	BinaryEncodingMethod string `toml:"binary-encoding-method" json:"binary-encoding-method"`
 }
 
 func (c *CSVConfig) validateAndAdjust() error {
@@ -199,6 +206,14 @@ func (c *CSVConfig) validateAndAdjust() error {
 			errors.New("csv config quote and delimiter cannot be the same"))
 	}
 
+	// validate binary encoding method
+	switch c.BinaryEncodingMethod {
+	case BinaryEncodingHex, BinaryEncodingBase64:
+	default:
+		return cerror.WrapError(cerror.ErrSinkInvalidConfig,
+			errors.New("csv config binary-encoding-method can only be hex or base64"))
+	}
+
 	return nil
 }
 

pkg/config/sink_test.go

@@ -281,8 +281,9 @@ func TestValidateAndAdjustCSVConfig(t *testing.T) {
 		{
 			name: "valid quote",
 			config: &CSVConfig{
-				Quote:     "\"",
-				Delimiter: ",",
+				Quote:                "\"",
+				Delimiter:            ",",
+				BinaryEncodingMethod: BinaryEncodingBase64,
 			},
 			wantErr: "",
 		},
@@ -303,8 +304,9 @@ func TestValidateAndAdjustCSVConfig(t *testing.T) {
 		{
 			name: "valid delimiter1",
 			config: &CSVConfig{
-				Quote:     "\"",
-				Delimiter: ",",
+				Quote:                "\"",
+				Delimiter:            ",",
+				BinaryEncodingMethod: BinaryEncodingHex,
 			},
 			wantErr: "",
 		},
@@ -332,6 +334,15 @@ func TestValidateAndAdjustCSVConfig(t *testing.T) {
 			},
 			wantErr: "csv config quote and delimiter cannot be the same",
 		},
+		{
+			name: "invalid binary encoding method",
+			config: &CSVConfig{
+				Quote:                "\"",
+				Delimiter:            ",",
+				BinaryEncodingMethod: "invalid",
+			},
+			wantErr: "csv config binary-encoding-method can only be hex or base64",
+		},
 	}
 	for _, c := range tests {
 		tc := c

pkg/sink/codec/common/config.go

@@ -58,11 +58,12 @@ type Config struct {
 	AvroEnableWatermark bool
 
 	// for sinking to cloud storage
-	Delimiter       string
-	Quote           string
-	NullString      string
-	IncludeCommitTs bool
-	Terminator      string
+	Delimiter            string
+	Quote                string
+	NullString           string
+	IncludeCommitTs      bool
+	Terminator           string
+	BinaryEncodingMethod string
 
 	// for open protocol
 	OnlyOutputUpdatedColumns bool
@@ -174,6 +175,7 @@ func (c *Config) Apply(sinkURI *url.URL, replicaConfig *config.ReplicaConfig) er
 			c.Quote = replicaConfig.Sink.CSVConfig.Quote
 			c.NullString = replicaConfig.Sink.CSVConfig.NullString
 			c.IncludeCommitTs = replicaConfig.Sink.CSVConfig.IncludeCommitTs
+			c.BinaryEncodingMethod = replicaConfig.Sink.CSVConfig.BinaryEncodingMethod
 		}
 		if replicaConfig.Sink.KafkaConfig != nil {
 			c.LargeMessageHandle = replicaConfig.Sink.KafkaConfig.LargeMessageHandle

pkg/sink/codec/csv/csv_decoder.go

@@ -109,7 +109,7 @@ func (b *batchDecoder) NextRowChangedEvent() (*model.RowChangedEvent, error) {
 		return nil, cerror.WrapError(cerror.ErrCSVDecodeFailed, errors.New("no csv row can be found"))
 	}
 
-	e, err := csvMsg2RowChangedEvent(b.msg, b.tableInfo.Columns)
+	e, err := csvMsg2RowChangedEvent(b.codecConfig, b.msg, b.tableInfo.Columns)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}