just for run tests

cdc/entry/metrics.go

@@ -13,11 +13,24 @@
 
 package entry
 
-import (
-	"github.com/prometheus/client_golang/prometheus"
-)
+import "github.com/prometheus/client_golang/prometheus"
 
 var (
+	tableMountedResolvedTsGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "entry",
+			Name:      "table_mounted_resolved_ts",
+			Help:      "real local resolved ts of processor",
+		}, []string{"changefeed", "capture", "table"})
+	ddlResolvedTsGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "entry",
+			Name:      "table_ddl_resolved_ts",
+			Help:      "real local resolved ts of processor",
+		}, []string{"changefeed", "capture"})
+
 	mounterOutputChanSizeGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
@@ -29,5 +42,7 @@ var (
 
 // InitMetrics registers all metrics in this file
 func InitMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(ddlResolvedTsGauge)
+	registry.MustRegister(tableMountedResolvedTsGauge)
 	registry.MustRegister(mounterOutputChanSizeGauge)
 }

cdc/entry/mounter.go

@@ -19,14 +19,19 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"math"
+	"strconv"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	timodel "github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/ticdc/cdc/model"
 	"github.com/pingcap/ticdc/pkg/util"
+	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
 	"go.uber.org/zap"
@@ -77,6 +82,7 @@ func (idx *indexKVEntry) unflatten(tableInfo *TableInfo) error {
 	}
 	return nil
 }
+
 func isDistinct(index *timodel.IndexInfo, indexValue []types.Datum) bool {
 	if index.Primary {
 		return true
@@ -102,6 +108,7 @@ type mounterImpl struct {
 	schemaStorage   *Storage
 	rawRowChangedCh <-chan *model.RawKVEntry
 	output          chan *model.RowChangedEvent
+	unmarshalRow    chan *model.RawKVEntry
 }
 
 // NewMounter creates a mounter
@@ -110,43 +117,124 @@ func NewMounter(rawRowChangedCh <-chan *model.RawKVEntry, schemaStorage *Storage
 		schemaStorage:   schemaStorage,
 		rawRowChangedCh: rawRowChangedCh,
 		output:          make(chan *model.RowChangedEvent, defaultOutputChanSize),
+		unmarshalRow:    make(chan *model.RawKVEntry, defaultOutputChanSize),
 	}
 }
 
 func (m *mounterImpl) Run(ctx context.Context) error {
-	go func() {
-		m.collectMetrics(ctx)
-	}()
+	captureID := util.CaptureIDFromCtx(ctx)
+	changefeedID := util.ChangefeedIDFromCtx(ctx)
+	tableID := util.TableIDFromCtx(ctx)
 
-	for {
-		var rawRow *model.RawKVEntry
-		select {
-		case <-ctx.Done():
-			return errors.Trace(ctx.Err())
-		case rawRow = <-m.rawRowChangedCh:
-		}
-		if rawRow == nil {
-			continue
-		}
+	errg, cctx := errgroup.WithContext(ctx)
 
-		if err := m.schemaStorage.HandlePreviousDDLJobIfNeed(rawRow.Ts); err != nil {
-			return errors.Trace(err)
-		}
+	errg.Go(func() error {
+		m.collectMetrics(cctx)
+		return nil
+	})
 
-		if rawRow.OpType == model.OpTypeResolved {
-			m.output <- &model.RowChangedEvent{Resolved: true, Ts: rawRow.Ts}
-			continue
-		}
+	errg.Go(func() error {
+		for {
+			var rawRow *model.RawKVEntry
+			select {
+			case <-cctx.Done():
+				return errors.Trace(cctx.Err())
+			case rawRow = <-m.rawRowChangedCh:
+			}
+			if rawRow == nil {
+				continue
+			}
+
+			if err := m.schemaStorage.HandlePreviousDDLJobIfNeed(rawRow.Ts); err != nil {
+				return errors.Trace(err)
+			}
+
+			if rawRow.OpType == model.OpTypeResolved {
+				tableMountedResolvedTsGauge.WithLabelValues(changefeedID, captureID, strconv.FormatInt(tableID, 10)).Set(float64(oracle.ExtractPhysical(rawRow.Ts)))
+			}
+			m.unmarshalRow <- rawRow
 
-		event, err := m.unmarshalAndMountRowChanged(rawRow)
-		if err != nil {
-			return errors.Trace(err)
-		}
-		if event == nil {
-			continue
 		}
-		m.output <- event
+	})
+
+	errg.Go(func() error {
+		return m.unmarshalWorker(cctx)
+	})
+	return errg.Wait()
+}
+
+const workerNum = 4
+
+func (m *mounterImpl) unmarshalWorker(ctx context.Context) error {
+	errg, cctx := errgroup.WithContext(ctx)
+	eventChs := make([]chan *model.RowChangedEvent, workerNum)
+
+	for i := 0; i < workerNum; i++ {
+		i := i
+		eventChs[i] = make(chan *model.RowChangedEvent, defaultOutputChanSize)
+		errg.Go(func() error {
+			var rawRow *model.RawKVEntry
+			for {
+				select {
+				case <-cctx.Done():
+					return errors.Trace(cctx.Err())
+				case rawRow = <-m.unmarshalRow:
+				}
+				if rawRow.OpType == model.OpTypeResolved {
+					for j := 0; j < workerNum; j++ {
+						eventChs[j] <- &model.RowChangedEvent{Ts: rawRow.Ts, Resolved: true}
+					}
+					continue
+				}
+				event, err := m.unmarshalAndMountRowChanged(rawRow)
+				if err != nil {
+					return errors.Trace(err)
+				}
+				if event == nil {
+					continue
+				}
+				eventChs[i] <- event
+			}
+		})
 	}
+
+	errg.Go(func() error {
+		defer func() {
+			for i := 0; i < workerNum; i++ {
+				close(eventChs[i])
+			}
+		}()
+		events := make([]*model.RowChangedEvent, workerNum)
+		var lastResolvedTs uint64
+		for {
+			minTs := uint64(math.MaxUint64)
+			var minChIndex int
+			for i := 0; i < workerNum; i++ {
+				if events[i] == nil {
+					select {
+					case events[i] = <-eventChs[i]:
+					case <-cctx.Done():
+						return errors.Trace(cctx.Err())
+					}
+				}
+				if minTs > events[i].Ts {
+					minTs = events[i].Ts
+					minChIndex = i
+				}
+			}
+			if events[minChIndex].Resolved {
+				if events[minChIndex].Ts != lastResolvedTs {
+					m.output <- events[minChIndex]
+					lastResolvedTs = events[minChIndex].Ts
+				}
+			} else {
+				m.output <- events[minChIndex]
+			}
+			events[minChIndex] = nil
+			minTs = uint64(math.MaxUint64)
+		}
+	})
+	return errg.Wait()
 }
 
 func (m *mounterImpl) Output() <-chan *model.RowChangedEvent {
@@ -167,6 +255,7 @@ func (m *mounterImpl) collectMetrics(ctx context.Context) {
 		}
 	}()
 }
+
 func (m *mounterImpl) unmarshalAndMountRowChanged(raw *model.RawKVEntry) (*model.RowChangedEvent, error) {
 	if !bytes.HasPrefix(raw.Key, tablePrefix) {
 		return nil, nil
@@ -404,7 +493,10 @@ func formatColVal(value interface{}, tp byte) (interface{}, error) {
 	}
 
 	switch tp {
-	case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeNewDate, mysql.TypeTimestamp, mysql.TypeDuration, mysql.TypeDecimal, mysql.TypeNewDecimal, mysql.TypeJSON:
+	case mysql.TypeDate, mysql.TypeDatetime,
+		mysql.TypeNewDate, mysql.TypeTimestamp,
+		mysql.TypeDuration, mysql.TypeDecimal,
+		mysql.TypeNewDecimal, mysql.TypeJSON:
 		value = fmt.Sprintf("%v", value)
 	case mysql.TypeEnum:
 		value = value.(types.Enum).Value

cdc/entry/schema_builder.go

@@ -7,6 +7,9 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/pingcap/ticdc/pkg/util"
+	"github.com/pingcap/tidb/store/tikv/oracle"
+
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	timodel "github.com/pingcap/parser/model"
@@ -128,6 +131,9 @@ func NewStorageBuilder(historyDDL []*timodel.Job, ddlEventCh <-chan *model.RawKV
 
 // Run runs the StorageBuilder
 func (b *StorageBuilder) Run(ctx context.Context) error {
+
+	captureID := util.CaptureIDFromCtx(ctx)
+	changefeedID := util.ChangefeedIDFromCtx(ctx)
 	for {
 		var rawKV *model.RawKVEntry
 		select {
@@ -143,6 +149,8 @@ func (b *StorageBuilder) Run(ctx context.Context) error {
 		}
 
 		if rawKV.OpType == model.OpTypeResolved {
+			ddlResolvedTsGauge.WithLabelValues(changefeedID, captureID).Set(float64(oracle.ExtractPhysical(rawKV.Ts)))
+
 			atomic.StoreUint64(&b.resolvedTs, rawKV.Ts)
 			continue
 		}

cdc/entry/schema_storage.go

@@ -106,6 +106,7 @@ func WrapTableInfo(info *timodel.TableInfo) *TableInfo {
 			if mysql.HasPriKeyFlag(col.Flag) {
 				// Prepend to make sure the primary key ends up at the front
 				ti.UniqueColumns[col.ID] = struct{}{}
+				ti.IndieMarkCol = col.Name.O
 				break
 			}
 		}

cdc/kv/client.go

@@ -155,7 +155,7 @@ func (a *connArray) Init(ctx context.Context) error {
 				Timeout:             3 * time.Second,
 				PermitWithoutStream: true,
 			}),
-			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(32*1024*1024)),
+			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(128*1024*1024)),
 		)
 		cancel()
 

cdc/owner_operator.go

@@ -41,7 +41,7 @@ type ddlHandler struct {
 func newDDLHandler(pdCli pd.Client, checkpointTS uint64) *ddlHandler {
 	// The key in DDL kv pair returned from TiKV is already memcompariable encoded,
 	// so we set `needEncode` to false.
-	puller := puller.NewPuller(pdCli, checkpointTS, []util.Span{util.GetDDLSpan(), util.GetAddIndexDDLSpan()}, false, nil)
+	puller := puller.NewPuller(pdCli, checkpointTS, 0, []util.Span{util.GetDDLSpan(), util.GetAddIndexDDLSpan()}, false, nil)
 	ctx, cancel := context.WithCancel(context.Background())
 	h := &ddlHandler{
 		puller: puller,

cdc/processor.go

@@ -131,7 +131,7 @@ func newProcessor(
 
 	// The key in DDL kv pair returned from TiKV is already memcompariable encoded,
 	// so we set `needEncode` to false.
-	ddlPuller := puller.NewPuller(pdCli, checkpointTs, []util.Span{util.GetDDLSpan(), util.GetAddIndexDDLSpan()}, false, limitter)
+	ddlPuller := puller.NewPuller(pdCli, checkpointTs, 0, []util.Span{util.GetDDLSpan(), util.GetAddIndexDDLSpan()}, false, limitter)
 	ddlEventCh := ddlPuller.SortedOutput(ctx)
 	schemaBuilder, err := createSchemaBuilder(endpoints, ddlEventCh)
 	if err != nil {
@@ -565,6 +565,7 @@ func createTsRWriter(cli kv.CDCEtcdClient, changefeedID, captureID string) (stor
 func (p *processor) addTable(ctx context.Context, tableID int64, startTs uint64) {
 	p.tablesMu.Lock()
 	defer p.tablesMu.Unlock()
+	ctx = util.PutTableIDInCtx(ctx, tableID)
 
 	log.Debug("Add table", zap.Int64("tableID", tableID))
 	if _, ok := p.tables[tableID]; ok {
@@ -582,7 +583,7 @@ func (p *processor) addTable(ctx context.Context, tableID int64, startTs uint64)
 	// The key in DML kv pair returned from TiKV is not memcompariable encoded,
 	// so we set `needEncode` to true.
 	span := util.GetTableSpan(tableID, true)
-	puller := puller.NewPuller(p.pdCli, startTs, []util.Span{span}, true, p.limitter)
+	puller := puller.NewPuller(p.pdCli, startTs, tableID, []util.Span{span}, true, p.limitter)
 	go func() {
 		err := puller.Run(ctx)
 		if errors.Cause(err) != context.Canceled {