pingcap · ti-chi-bot · May 10, 2023
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
@@ -279,6 +279,16 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 			Terminator:               c.Sink.Terminator,
 			DateSeparator:            c.Sink.DateSeparator,
 			EnablePartitionSeparator: c.Sink.EnablePartitionSeparator,
+<<<<<<< HEAD
+=======
+			FileIndexWidth:           c.Sink.FileIndexWidth,
+			EnableKafkaSinkV2:        c.Sink.EnableKafkaSinkV2,
+			OnlyOutputUpdatedColumns: c.Sink.OnlyOutputUpdatedColumns,
+			KafkaConfig:              kafkaConfig,
+			MySQLConfig:              mysqlConfig,
+			CloudStorageConfig:       cloudStorageConfig,
+			SafeMode:                 c.Sink.SafeMode,
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
 		}
 	}
 	if c.Mounter != nil {
@@ -383,6 +393,16 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 			Terminator:               cloned.Sink.Terminator,
 			DateSeparator:            cloned.Sink.DateSeparator,
 			EnablePartitionSeparator: cloned.Sink.EnablePartitionSeparator,
+<<<<<<< HEAD
+=======
+			FileIndexWidth:           cloned.Sink.FileIndexWidth,
+			EnableKafkaSinkV2:        cloned.Sink.EnableKafkaSinkV2,
+			OnlyOutputUpdatedColumns: cloned.Sink.OnlyOutputUpdatedColumns,
+			KafkaConfig:              kafkaConfig,
+			MySQLConfig:              mysqlConfig,
+			CloudStorageConfig:       cloudStorageConfig,
+			SafeMode:                 cloned.Sink.SafeMode,
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
 		}
 	}
 	if cloned.Consistent != nil {
@@ -502,6 +522,7 @@ type Table struct {
 // SinkConfig represents sink config for a changefeed
 // This is a duplicate of config.SinkConfig
 type SinkConfig struct {
+<<<<<<< HEAD
 	Protocol                 string            `json:"protocol"`
 	SchemaRegistry           string            `json:"schema_registry"`
 	CSVConfig                *CSVConfig        `json:"csv"`
@@ -512,6 +533,25 @@ type SinkConfig struct {
 	Terminator               string            `json:"terminator"`
 	DateSeparator            string            `json:"date_separator"`
 	EnablePartitionSeparator bool              `json:"enable_partition_separator"`
+=======
+	Protocol                 string              `json:"protocol"`
+	SchemaRegistry           string              `json:"schema_registry"`
+	CSVConfig                *CSVConfig          `json:"csv"`
+	DispatchRules            []*DispatchRule     `json:"dispatchers,omitempty"`
+	ColumnSelectors          []*ColumnSelector   `json:"column_selectors"`
+	TxnAtomicity             string              `json:"transaction_atomicity"`
+	EncoderConcurrency       int                 `json:"encoder_concurrency"`
+	Terminator               string              `json:"terminator"`
+	DateSeparator            string              `json:"date_separator"`
+	EnablePartitionSeparator bool                `json:"enable_partition_separator"`
+	FileIndexWidth           int                 `json:"file_index_width"`
+	EnableKafkaSinkV2        bool                `json:"enable_kafka_sink_v2"`
+	OnlyOutputUpdatedColumns *bool               `json:"only_output_updated_columns"`
+	SafeMode                 *bool               `json:"safe_mode,omitempty"`
+	KafkaConfig              *KafkaConfig        `json:"kafka_config,omitempty"`
+	MySQLConfig              *MySQLConfig        `json:"mysql_config,omitempty"`
+	CloudStorageConfig       *CloudStorageConfig `json:"cloud_storage_config,omitempty"`
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
 }
 
 // CSVConfig denotes the csv config

diff --git a/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go b/cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go
@@ -91,8 +91,13 @@ func TestCloudStorageWriteEventsWithoutDateSeparator(t *testing.T) {
 	require.Nil(t, err)
 
 	replicaConfig := config.GetDefaultReplicaConfig()
+<<<<<<< HEAD:cdc/sinkv2/eventsink/cloudstorage/cloud_storage_dml_sink_test.go
 	replicaConfig.Sink.Protocol = config.ProtocolOpen.String()
 
+=======
+	replicaConfig.Sink.Protocol = config.ProtocolCsv.String()
+	replicaConfig.Sink.FileIndexWidth = 6
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918)):cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink_test.go
 	errCh := make(chan error, 5)
 	s, err := NewCloudStorageSink(ctx, sinkURI, replicaConfig, errCh)
 	require.Nil(t, err)
@@ -163,6 +168,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
 	replicaConfig := config.GetDefaultReplicaConfig()
 	replicaConfig.Sink.Protocol = config.ProtocolOpen.String()
 	replicaConfig.Sink.DateSeparator = config.DateSeparatorDay.String()
+	replicaConfig.Sink.FileIndexWidth = 6
 
 	errCh := make(chan error, 5)
 	s, err := NewCloudStorageSink(ctx, sinkURI, replicaConfig, errCh)

diff --git a/cdc/sinkv2/eventsink/cloudstorage/dml_worker_test.go b/cdc/sinkv2/eventsink/cloudstorage/dml_worker_test.go
@@ -46,6 +46,7 @@ func testDMLWorker(ctx context.Context, t *testing.T, dir string) *dmlWorker {
 	require.Nil(t, err)
 	cfg := cloudstorage.NewConfig()
 	err = cfg.Apply(context.TODO(), sinkURI, config.GetDefaultReplicaConfig())
+	cfg.FileIndexWidth = 6
 	require.Nil(t, err)
 
 	statistics := metrics.NewStatistics(ctx, sink.TxnSink)

diff --git a/cmd/storage-consumer/main.go b/cmd/storage-consumer/main.go
@@ -63,6 +63,7 @@ var (
 	logFile          string
 	logLevel         string
 	flushInterval    time.Duration
+	fileIndexWidth   int
 	enableProfiling  bool
 	timezone         string
 )
@@ -80,6 +81,8 @@ func init() {
 	flag.StringVar(&logFile, "log-file", "", "log file path")
 	flag.StringVar(&logLevel, "log-level", "info", "log level")
 	flag.DurationVar(&flushInterval, "flush-interval", 10*time.Second, "flush interval")
+	flag.IntVar(&fileIndexWidth, "file-index-width",
+		config.DefaultFileIndexWidth, "file index width")
 	flag.BoolVar(&enableProfiling, "enable-profiling", false, "whether to enable profiling")
 	flag.StringVar(&timezone, "tz", "System", "Specify time zone of storage consumer")
 	flag.Parse()
@@ -532,7 +535,11 @@ func (c *consumer) syncExecDMLEvents(
 	key dmlPathKey,
 	fileIdx uint64,
 ) error {
+<<<<<<< HEAD
 	filePath := key.generateDMLFilePath(fileIdx, c.fileExtension)
+=======
+	filePath := key.GenerateDMLFilePath(fileIdx, c.fileExtension, fileIndexWidth)
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
 	log.Debug("read from dml file path", zap.String("path", filePath))
 	content, err := c.externalStorage.ReadFile(ctx, filePath)
 	if err != nil {

diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go
@@ -1271,6 +1271,21 @@ var doc = `{
                 "encoder-concurrency": {
                     "type": "integer"
                 },
+<<<<<<< HEAD
+=======
+                "file-index-digit": {
+                    "type": "integer"
+                },
+                "kafka-config": {
+                    "$ref": "#/definitions/config.KafkaConfig"
+                },
+                "mysql-config": {
+                    "$ref": "#/definitions/config.MySQLConfig"
+                },
+                "only-output-updated-columns": {
+                    "type": "boolean"
+                },
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
                 "protocol": {
                     "type": "string"
                 },
@@ -1976,6 +1991,21 @@ var doc = `{
                 "encoder_concurrency": {
                     "type": "integer"
                 },
+<<<<<<< HEAD
+=======
+                "file_index_width": {
+                    "type": "integer"
+                },
+                "kafka_config": {
+                    "$ref": "#/definitions/v2.KafkaConfig"
+                },
+                "mysql_config": {
+                    "$ref": "#/definitions/v2.MySQLConfig"
+                },
+                "only_output_updated_columns": {
+                    "type": "boolean"
+                },
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
                 "protocol": {
                     "type": "string"
                 },

diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json
@@ -1252,6 +1252,21 @@
                 "encoder-concurrency": {
                     "type": "integer"
                 },
+<<<<<<< HEAD
+=======
+                "file-index-digit": {
+                    "type": "integer"
+                },
+                "kafka-config": {
+                    "$ref": "#/definitions/config.KafkaConfig"
+                },
+                "mysql-config": {
+                    "$ref": "#/definitions/config.MySQLConfig"
+                },
+                "only-output-updated-columns": {
+                    "type": "boolean"
+                },
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
                 "protocol": {
                     "type": "string"
                 },
@@ -1957,6 +1972,21 @@
                 "encoder_concurrency": {
                     "type": "integer"
                 },
+<<<<<<< HEAD
+=======
+                "file_index_width": {
+                    "type": "integer"
+                },
+                "kafka_config": {
+                    "$ref": "#/definitions/v2.KafkaConfig"
+                },
+                "mysql_config": {
+                    "$ref": "#/definitions/v2.MySQLConfig"
+                },
+                "only_output_updated_columns": {
+                    "type": "boolean"
+                },
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
                 "protocol": {
                     "type": "string"
                 },

diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
@@ -60,6 +60,17 @@ definitions:
         type: boolean
       encoder-concurrency:
         type: integer
+<<<<<<< HEAD
+=======
+      file-index-digit:
+        type: integer
+      kafka-config:
+        $ref: '#/definitions/config.KafkaConfig'
+      mysql-config:
+        $ref: '#/definitions/config.MySQLConfig'
+      only-output-updated-columns:
+        type: boolean
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
       protocol:
         type: string
       schema-registry:
@@ -530,6 +541,17 @@ definitions:
         type: boolean
       encoder_concurrency:
         type: integer
+<<<<<<< HEAD
+=======
+      file_index_width:
+        type: integer
+      kafka_config:
+        $ref: '#/definitions/v2.KafkaConfig'
+      mysql_config:
+        $ref: '#/definitions/v2.MySQLConfig'
+      only_output_updated_columns:
+        type: boolean
+>>>>>>> 19277542cd (sink(ticdc): add index config to storage sink (#8918))
       protocol:
         type: string
       schema_registry:

diff --git a/pkg/cmd/util/helper_test.go b/pkg/cmd/util/helper_test.go
@@ -218,6 +218,7 @@ func TestAndWriteStorageSinkTOML(t *testing.T) {
 		Terminator:               "\r\n",
 		DateSeparator:            "day",
 		EnablePartitionSeparator: true,
+		FileIndexWidth:           config.DefaultFileIndexWidth,
 		CSVConfig: &config.CSVConfig{
 			Delimiter:       ",",
 			Quote:           "\"",