Merge branch 'master' into json-use-jsoniter

.github/workflows/check_and_build.yaml

@@ -42,7 +42,7 @@ jobs:
       - name: Setup Go environment
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Cache Tools
         id: cache-tools
@@ -68,7 +68,7 @@ jobs:
       - name: Setup Go environment
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Build
         run: make build

.github/workflows/dataflow_engine_chaos.yaml

@@ -46,7 +46,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Cache go modules
         uses: actions/cache@v2

.github/workflows/dataflow_engine_e2e.yaml

@@ -25,7 +25,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Build images
         run: make engine_image
@@ -53,7 +53,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Build images
         run: make engine_image
@@ -81,7 +81,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Build images
         run: make engine_image
@@ -109,7 +109,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Build images
         run: make engine_image

.github/workflows/dm_binlog_999999.yaml

@@ -16,10 +16,10 @@ jobs:
     runs-on: ubuntu-20.04
 
     steps:
-      - name: Set up Go 1.18
+      - name: Set up Go 1.19
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Check out code
         uses: actions/checkout@v2

.github/workflows/dm_chaos.yaml

@@ -37,10 +37,10 @@ jobs:
     # Steps represent a sequence of tasks that will be executed as part of the job
     steps:
       # Set up Go for building DM, now it's v1.16
-      - name: Set up Go 1.18
+      - name: Set up Go 1.19
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
       - name: Print Go version
         run: go version
 

.github/workflows/dm_upstream_switch.yaml

@@ -16,10 +16,10 @@ jobs:
     runs-on: ubuntu-20.04
 
     steps:
-      - name: Set up Go 1.18
+      - name: Set up Go 1.19
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Check out code
         uses: actions/checkout@v2

.github/workflows/ticdc_integration.yaml

@@ -37,7 +37,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Cache Vendor
         id: cache-vendor
@@ -99,7 +99,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Cache Vendor
         id: cache-vendor
@@ -161,7 +161,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Cache Vendor
         id: cache-vendor

.github/workflows/upgrade_dm_via_tiup.yaml

@@ -29,10 +29,10 @@ jobs:
       working-directory: ${{ github.workspace }}/go/src/github.com/pingcap/tiflow
 
     steps:
-      - name: Set up Go 1.18
+      - name: Set up Go 1.19
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Check out code
         uses: actions/checkout@v2
@@ -80,10 +80,10 @@ jobs:
           ["v2.0.1", "v2.0.7", "v5.3.0"]
 
     steps:
-      - name: Set up Go 1.18
+      - name: Set up Go 1.19
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Check out code
         uses: actions/checkout@v2
@@ -186,10 +186,10 @@ jobs:
     env:
       working-directory: ${{ github.workspace }}/go/src/github.com/pingcap/tiflow
     steps:
-      - name: Set up Go 1.18
+      - name: Set up Go 1.19
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18
+          go-version: 1.19
 
       - name: Check out code
         uses: actions/checkout@v2

.golangci.yml

@@ -1,5 +1,5 @@
 run:
-  go: 1.18
+  go: 1.19
 linters:
   enable:
     - unconvert
@@ -101,6 +101,11 @@ issues:
         - errcheck
         - gosec
         - makezero
+    # G113 Potential uncontrolled memory consumption in Rat.SetString (CVE-2022-23772)
+    # only affects gp < 1.16.14. and go < 1.17.7
+    - text: "(G113)"
+      linters:
+        - gosec
   include:
     - EXC0012 # exported (.+) should have comment( \(or a comment on this block\))? or be unexported)
     - EXC0014 # comment on exported (.+) should be of the form "(.+)..."

CONTRIBUTING.md

@@ -14,7 +14,7 @@ your contribution accepted.
 
 Developing TiDB-CDC requires:
 
-* [Go 1.18+](https://go.dev/doc/code)
+* [Go 1.19+](https://go.dev/doc/code)
 * An internet connection to download the dependencies
 
 Simply run `make` to build the program.

README_DM.md

@@ -19,7 +19,7 @@ To check the code style and build binaries, you can simply run:
 make build
 ```
 
-Note that DM supports building with Go version `Go >= 1.16`, and unit test preparation can be found in [Running/Unit Test](dm/tests/README.md#Unit-Test)
+Note that DM supports building with Go version `Go >= 1.19`, and unit test preparation can be found in [Running/Unit Test](dm/tests/README.md#Unit-Test)
 
 If you only want to build binaries, you can run:
 

README_TiCDC.md

@@ -29,7 +29,7 @@ $ make cdc
 $ make test
 ```
 
-Note that TiCDC supports building with Go version `Go >= 1.18`.
+Note that TiCDC supports building with Go version `Go >= 1.19`.
 
 When TiCDC is built successfully, you can find binary in the `bin` directory. Instructions for unit test and integration
 test can be found in [Running tests](./tests/integration_tests/README.md).

cdc/api/v1/api.go

@@ -173,10 +173,8 @@ func (h *OpenAPI) ListChangefeed(c *gin.Context) {
 			ID:         cfID.ID,
 		}
 
-		if cfInfo != nil {
-			resp.FeedState = cfInfo.State
-			resp.RunningError = cfInfo.Error
-		}
+		resp.FeedState = cfInfo.State
+		resp.RunningError = cfInfo.Error
 
 		if cfStatus != nil {
 			resp.CheckpointTSO = cfStatus.CheckpointTs

cdc/api/v2/model.go

@@ -177,9 +177,22 @@ func (c *ReplicaConfig) ToInternalReplicaConfig() *config.ReplicaConfig {
 				Columns: selector.Columns,
 			})
 		}
+		var csvConfig *config.CSVConfig
+		if c.Sink.CSVConfig != nil {
+			csvConfig = &config.CSVConfig{
+				Delimiter:       c.Sink.CSVConfig.Delimiter,
+				Quote:           c.Sink.CSVConfig.Quote,
+				Terminator:      c.Sink.CSVConfig.Terminator,
+				NullString:      c.Sink.CSVConfig.NullString,
+				DateSeparator:   c.Sink.CSVConfig.DateSeparator,
+				IncludeCommitTs: c.Sink.CSVConfig.IncludeCommitTs,
+			}
+		}
+
 		res.Sink = &config.SinkConfig{
 			DispatchRules:   dispatchRules,
 			Protocol:        c.Sink.Protocol,
+			CSVConfig:       csvConfig,
 			TxnAtomicity:    config.AtomicityLevel(c.Sink.TxnAtomicity),
 			ColumnSelectors: columnSelectors,
 			SchemaRegistry:  c.Sink.SchemaRegistry,
@@ -259,10 +272,23 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 				Columns: selector.Columns,
 			})
 		}
+		var csvConfig *CSVConfig
+		if cloned.Sink.CSVConfig != nil {
+			csvConfig = &CSVConfig{
+				Delimiter:       cloned.Sink.CSVConfig.Delimiter,
+				Quote:           cloned.Sink.CSVConfig.Quote,
+				Terminator:      cloned.Sink.CSVConfig.Terminator,
+				NullString:      cloned.Sink.CSVConfig.NullString,
+				DateSeparator:   cloned.Sink.CSVConfig.DateSeparator,
+				IncludeCommitTs: cloned.Sink.CSVConfig.IncludeCommitTs,
+			}
+		}
+
 		res.Sink = &SinkConfig{
 			Protocol:        cloned.Sink.Protocol,
 			SchemaRegistry:  cloned.Sink.SchemaRegistry,
 			DispatchRules:   dispatchRules,
+			CSVConfig:       csvConfig,
 			ColumnSelectors: columnSelectors,
 			TxnAtomicity:    string(cloned.Sink.TxnAtomicity),
 		}
@@ -392,11 +418,23 @@ type Table struct {
 type SinkConfig struct {
 	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"`
 }
 
+// CSVConfig denotes the csv config
+// This is the same as config.CSVConfig
+type CSVConfig struct {
+	Delimiter       string `json:"delimiter"`
+	Quote           string `json:"quote"`
+	Terminator      string `json:"terminator"`
+	NullString      string `json:"null"`
+	DateSeparator   string `json:"date_separator"`
+	IncludeCommitTs bool   `json:"include_commit_ts"`
+}
+
 // DispatchRule represents partition rule for a table
 // This is a duplicate of config.DispatchRule
 type DispatchRule struct {

cdc/capture/capture.go

@@ -445,6 +445,7 @@ func (c *captureImpl) campaignOwner(ctx cdcContext.Context) error {
 		err = c.runEtcdWorker(ownerCtx, owner,
 			orchestrator.NewGlobalState(c.EtcdClient.GetClusterID()),
 			ownerFlushInterval, util.RoleOwner.String())
+		c.owner.AsyncStop()
 		c.setOwner(nil)
 
 		// if owner exits, resign the owner key,

cdc/entry/schema/snapshot.go

@@ -1055,7 +1055,6 @@ func (s *snapshot) iterPartitions(includeIneligible bool, f func(id int64, i *mo
 		}
 		return true
 	})
-	return
 }
 
 func (s *snapshot) iterSchemas(f func(i *timodel.DBInfo)) {
@@ -1128,7 +1127,7 @@ func (s *snapshot) drop() {
 
 	schemas := make([]versionedID, 0, s.schemas.Len())
 	var schemaID int64 = -1
-	var schemaDroped bool = false
+	schemaDroped := false
 	s.schemas.Ascend(func(x versionedID) bool {
 		if x.tag >= tag {
 			if x.id != schemaID {
@@ -1148,7 +1147,7 @@ func (s *snapshot) drop() {
 
 	tables := make([]versionedID, 0, s.tables.Len())
 	var tableID int64 = -1
-	var tableDroped bool = false
+	tableDroped := false
 	s.tables.Ascend(func(x versionedID) bool {
 		if x.tag >= tag {
 			if x.id != tableID {
@@ -1178,7 +1177,7 @@ func (s *snapshot) drop() {
 
 	partitions := make([]versionedID, 0, s.partitions.Len())
 	var partitionID int64 = -1
-	var partitionDroped bool = false
+	partitionDroped := false
 	s.partitions.Ascend(func(x versionedID) bool {
 		if x.tag >= tag {
 			if x.id != partitionID {
@@ -1204,8 +1203,8 @@ func (s *snapshot) drop() {
 	}
 
 	schemaNames := make([]versionedEntityName, 0, s.schemaNameToID.Len())
-	var schemaName string = ""
-	var schemaNameDroped bool = false
+	schemaName := ""
+	schemaNameDroped := false
 	s.schemaNameToID.Ascend(func(x versionedEntityName) bool {
 		if x.tag >= tag {
 			if x.entity != schemaName {
@@ -1225,8 +1224,8 @@ func (s *snapshot) drop() {
 
 	tableNames := make([]versionedEntityName, 0, s.tableNameToID.Len())
 	schemaID = -1
-	var tableName string = ""
-	var tableNameDroped bool = false
+	tableName := ""
+	tableNameDroped := false
 	s.tableNameToID.Ascend(func(x versionedEntityName) bool {
 		if x.tag >= tag {
 			if x.prefix != schemaID || x.entity != tableName {

cdc/entry/schema_storage_test.go

@@ -235,7 +235,7 @@ func TestTable(t *testing.T) {
 	}
 	preTableInfo, err := snap.PreTableInfo(job)
 	require.Nil(t, err)
-	require.Equal(t, preTableInfo.TableName, model.TableName{Schema: "Test", Table: "T"})
+	require.Equal(t, preTableInfo.TableName, model.TableName{Schema: "Test", Table: "T", TableID: 2})
 	require.Equal(t, preTableInfo.ID, int64(2))
 
 	err = snap.HandleDDL(job)
@@ -262,7 +262,7 @@ func TestTable(t *testing.T) {
 	}
 	preTableInfo, err = snap.PreTableInfo(job)
 	require.Nil(t, err)
-	require.Equal(t, preTableInfo.TableName, model.TableName{Schema: "Test", Table: "T"})
+	require.Equal(t, preTableInfo.TableName, model.TableName{Schema: "Test", Table: "T", TableID: 9})
 	require.Equal(t, preTableInfo.ID, int64(9))
 
 	err = snap.HandleDDL(job)