fb/latency(cdc): merge master to resolve conflicts.

Makefile

@@ -463,6 +463,9 @@ tools/bin/msgp: tools/check/go.mod
 tools/bin/protoc:
 	./scripts/download-protoc.sh
 
+tools/bin/goimports:
+	cd tools/check && $(GO) build -mod=mod -o ../bin/goimports golang.org/x/tools/cmd/goimports
+
 check_failpoint_ctl: tools/bin/failpoint-ctl
 
 failpoint-enable: check_failpoint_ctl
@@ -473,7 +476,7 @@ failpoint-disable: check_failpoint_ctl
 
 engine: df-master df-executor df-master-client df-demo
 
-df-proto:
+df-proto: tools/bin/protoc tools/bin/protoc-gen-gogofaster tools/bin/goimports
 	./engine/generate-proto.sh
 
 df-master:

cdc/api/v2/api.go

@@ -0,0 +1,38 @@
+// Copyright 2022 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.
+
+package v2
+
+import (
+	"github.com/gin-gonic/gin"
+	"github.com/pingcap/tiflow/cdc/api/middleware"
+	"github.com/pingcap/tiflow/cdc/capture"
+)
+
+// OpenAPIV2 provides CDC v2 APIs
+type OpenAPIV2 struct {
+	capture *capture.Capture
+}
+
+// NewOpenAPI creates a new openAPIs.
+func NewOpenAPI(c *capture.Capture) *OpenAPIV2 {
+	return &OpenAPIV2{capture: c}
+}
+
+// RegisterOpenAPIRoutes registers routes for OpenAPI
+func RegisterOpenAPIRoutes(router *gin.Engine, api *OpenAPIV2) {
+	v2 := router.Group("/api/v2")
+
+	v2.Use(middleware.LogMiddleware())
+	v2.Use(middleware.ErrorHandleMiddleware())
+}

cdc/api/v2/changefeed.go

@@ -0,0 +1,14 @@
+// Copyright 2022 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.
+
+package v2

cdc/api/v2/model.go

@@ -0,0 +1,20 @@
+// Copyright 2022 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.
+
+package v2
+
+// Tso contains timestamp get from PD
+type Tso struct {
+	Timestamp int64 `json:"timestamp"`
+	LogicTime int64 `json:"logic-time"`
+}

cdc/capture/capture.go

@@ -146,14 +146,12 @@ func (c *Capture) reset(ctx context.Context) error {
 	if c.tableActorSystem != nil {
 		c.tableActorSystem.Stop()
 	}
-	if conf.Debug.EnableTableActor {
-		c.tableActorSystem = system.NewSystem()
-		err = c.tableActorSystem.Start(ctx)
-		if err != nil {
-			return errors.Annotate(
-				cerror.WrapError(cerror.ErrNewCaptureFailed, err),
-				"create table actor system")
-		}
+	c.tableActorSystem = system.NewSystem()
+	err = c.tableActorSystem.Start(ctx)
+	if err != nil {
+		return errors.Annotate(
+			cerror.WrapError(cerror.ErrNewCaptureFailed, err),
+			"create table actor system")
 	}
 	if conf.Debug.EnableDBSorter {
 		if c.sorterSystem != nil {

cdc/entry/codec.go

@@ -247,24 +247,24 @@ func unflatten(datum types.Datum, ft *types.FieldType, loc *time.Location) (type
 	if datum.IsNull() {
 		return datum, nil
 	}
-	switch ft.Tp {
+	switch ft.GetType() {
 	case mysql.TypeFloat:
 		datum.SetFloat32(float32(datum.GetFloat64()))
 		return datum, nil
 	case mysql.TypeVarchar, mysql.TypeString, mysql.TypeVarString, mysql.TypeTinyBlob,
 		mysql.TypeMediumBlob, mysql.TypeBlob, mysql.TypeLongBlob:
-		datum.SetString(datum.GetString(), ft.Collate)
+		datum.SetString(datum.GetString(), ft.GetCollate())
 	case mysql.TypeTiny, mysql.TypeShort, mysql.TypeYear, mysql.TypeInt24,
 		mysql.TypeLong, mysql.TypeLonglong, mysql.TypeDouble:
 		return datum, nil
 	case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeTimestamp:
-		t := types.NewTime(types.ZeroCoreTime, ft.Tp, ft.Decimal)
+		t := types.NewTime(types.ZeroCoreTime, ft.GetType(), ft.GetDecimal())
 		var err error
 		err = t.FromPackedUint(datum.GetUint64())
 		if err != nil {
 			return datum, cerror.WrapError(cerror.ErrDatumUnflatten, err)
 		}
-		if ft.Tp == mysql.TypeTimestamp && !t.IsZero() {
+		if ft.GetType() == mysql.TypeTimestamp && !t.IsZero() {
 			err = t.ConvertTimeZone(time.UTC, loc)
 			if err != nil {
 				return datum, cerror.WrapError(cerror.ErrDatumUnflatten, err)
@@ -274,27 +274,27 @@ func unflatten(datum types.Datum, ft *types.FieldType, loc *time.Location) (type
 		datum.SetMysqlTime(t)
 		return datum, nil
 	case mysql.TypeDuration: // duration should read fsp from column meta data
-		dur := types.Duration{Duration: time.Duration(datum.GetInt64()), Fsp: ft.Decimal}
+		dur := types.Duration{Duration: time.Duration(datum.GetInt64()), Fsp: ft.GetDecimal()}
 		datum.SetMysqlDuration(dur)
 		return datum, nil
 	case mysql.TypeEnum:
 		// ignore error deliberately, to read empty enum value.
-		enum, err := types.ParseEnumValue(ft.Elems, datum.GetUint64())
+		enum, err := types.ParseEnumValue(ft.GetElems(), datum.GetUint64())
 		if err != nil {
 			enum = types.Enum{}
 		}
-		datum.SetMysqlEnum(enum, ft.Collate)
+		datum.SetMysqlEnum(enum, ft.GetCollate())
 		return datum, nil
 	case mysql.TypeSet:
-		set, err := types.ParseSetValue(ft.Elems, datum.GetUint64())
+		set, err := types.ParseSetValue(ft.GetElems(), datum.GetUint64())
 		if err != nil {
 			return datum, cerror.WrapError(cerror.ErrDatumUnflatten, err)
 		}
-		datum.SetMysqlSet(set, ft.Collate)
+		datum.SetMysqlSet(set, ft.GetCollate())
 		return datum, nil
 	case mysql.TypeBit:
 		val := datum.GetUint64()
-		byteSize := (ft.Flen + 7) >> 3
+		byteSize := (ft.GetFlen() + 7) >> 3
 		datum.SetUint64(0)
 		datum.SetMysqlBit(types.NewBinaryLiteralFromUint(val, byteSize))
 	}

cdc/entry/mounter.go

@@ -280,8 +280,8 @@ func datum2Column(tableInfo *model.TableInfo, datums map[int64]types.Datum, fill
 		colSize += size
 		cols[tableInfo.RowColumnsOffset[colInfo.ID]] = &model.Column{
 			Name:    colName,
-			Type:    colInfo.Tp,
-			Charset: colInfo.Charset,
+			Type:    colInfo.GetType(),
+			Charset: colInfo.GetCharset(),
 			Value:   colValue,
 			Default: defaultValue,
 			Flag:    tableInfo.ColumnsFlag[colInfo.ID],
@@ -395,7 +395,7 @@ func formatColVal(datum types.Datum, col *timodel.ColumnInfo) (
 	if datum.IsNull() {
 		return nil, 0, "", nil
 	}
-	switch col.Tp {
+	switch col.GetType() {
 	case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeNewDate, mysql.TypeTimestamp:
 		v := datum.GetMysqlTime().String()
 		return v, sizeOfString(v), "", nil
@@ -472,24 +472,24 @@ func getDefaultOrZeroValue(col *timodel.ColumnInfo) (interface{}, int, string, e
 		return d.GetValue(), sizeOfDatum(d), "", nil
 	}
 
-	if !mysql.HasNotNullFlag(col.Flag) {
+	if !mysql.HasNotNullFlag(col.GetFlag()) {
 		// NOTICE: NotNullCheck need do after OriginDefaultValue check, as when TiDB meet "amend + add column default xxx",
 		// ref: https://github.com/pingcap/ticdc/issues/3929
 		// must use null if TiDB not write the column value when default value is null
 		// and the value is null, see https://github.com/pingcap/tidb/issues/9304
 		d = types.NewDatum(nil)
 	} else {
-		switch col.Tp {
+		switch col.GetType() {
 		case mysql.TypeEnum:
 			// For enum type, if no default value and not null is set,
 			// the default value is the first element of the enum list
-			d = types.NewDatum(col.FieldType.Elems[0])
+			d = types.NewDatum(col.FieldType.GetElem(0))
 		case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
 			return emptyBytes, sizeOfEmptyBytes, "", nil
 		default:
 			d = table.GetZeroValue(col)
 			if d.IsNull() {
-				log.Error("meet unsupported column type", zap.String("columnInfo", col.String()))
+				log.Error("meet unsupported column type", zap.String("columnInfo", col.FieldType.String()))
 			}
 		}
 	}