for cdc : inject some faults and fix some bugs

pkg/cdc/reader.go

@@ -215,6 +215,7 @@ var readTableWithTxn = func(
 func (reader *tableReader) readTable(ctx context.Context, ar *ActiveRoutine) (err error) {
 	//step1 : create an txnOp
 	txnOp, err := GetTxnOp(ctx, reader.cnEngine, reader.cnTxnClient, "readMultipleTables")
+
 	if err != nil {
 		return err
 	}
@@ -314,6 +315,7 @@ func (reader *tableReader) readTableWithTxn(
 
 	start := time.Now()
 	changes, err = CollectChanges(ctx, rel, fromTs, toTs, reader.mp)
+
 	v2.CdcReadDurationHistogram.Observe(time.Since(start).Seconds())
 	if err != nil {
 		return

pkg/cdc/sinker.go

@@ -91,12 +91,22 @@ var NewSinker = func(
 	ctx := context.Background()
 	padding := strings.Repeat(" ", sqlBufReserved)
 	// create db
-	_ = sink.Send(ctx, ar, []byte(padding+fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", dbTblInfo.SinkDbName)), false)
+	err = sink.Send(ctx, ar, []byte(padding+fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", dbTblInfo.SinkDbName)), false)
+	if err != nil {
+		return nil, err
+	}
 	// use db
-	_ = sink.Send(ctx, ar, []byte(padding+fmt.Sprintf("use `%s`", dbTblInfo.SinkDbName)), false)
+	err = sink.Send(ctx, ar, []byte(padding+fmt.Sprintf("use `%s`", dbTblInfo.SinkDbName)), false)
+	if err != nil {
+		return nil, err
+	}
 	// possibly need to drop table first
 	if dbTblInfo.IdChanged {
-		_ = sink.Send(ctx, ar, []byte(padding+fmt.Sprintf("DROP TABLE IF EXISTS `%s`", dbTblInfo.SinkTblName)), false)
+		err = sink.Send(ctx, ar, []byte(padding+fmt.Sprintf("DROP TABLE IF EXISTS `%s`", dbTblInfo.SinkTblName)), false)
+		if err != nil {
+			return nil, err
+		}
+		dbTblInfo.IdChanged = false
 	}
 	// create table
 	createSql := strings.TrimSpace(dbTblInfo.SourceCreateSql)
@@ -112,7 +122,10 @@ var NewSinker = func(
 		newTablePart = dbTblInfo.SinkTblName
 	}
 	createSql = createSql[:tableStart] + " " + newTablePart + createSql[tableEnd:]
-	_ = sink.Send(ctx, ar, []byte(padding+createSql), false)
+	err = sink.Send(ctx, ar, []byte(padding+createSql), false)
+	if err != nil {
+		return nil, err
+	}
 
 	return NewMysqlSinker(
 		sink,

pkg/cdc/sinker_test.go

@@ -19,6 +19,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"github.com/stretchr/testify/require"
 	"testing"
 	"time"
 
@@ -109,34 +110,42 @@ func TestNewSinker(t *testing.T) {
 		},
 	}
 
-	db, mock, err := sqlmock.New()
-	assert.NoError(t, err)
-	mock.ExpectExec(fakeSql).WillReturnResult(sqlmock.NewResult(1, 1))
-	mock.ExpectExec(fakeSql).WillReturnResult(sqlmock.NewResult(1, 1))
-	mock.ExpectExec(fakeSql).WillReturnResult(sqlmock.NewResult(1, 1))
-
-	sink := &mysqlSink{
-		user:          "root",
-		password:      "123456",
-		ip:            "127.0.0.1",
-		port:          3306,
-		retryTimes:    CDCDefaultRetryTimes,
-		retryDuration: CDCDefaultRetryDuration,
-		conn:          db,
-	}
-
-	sinkStub := gostub.Stub(&NewMysqlSink, func(_, _, _ string, _, _ int, _ time.Duration, _ string, _ bool) (Sink, error) {
-		return sink, nil
-	})
-	defer sinkStub.Reset()
-
-	sinkerStub := gostub.Stub(&NewMysqlSinker, func(Sink, uint64, string, *DbTableInfo, *CDCWatermarkUpdater, *plan.TableDef, *ActiveRoutine, uint64, bool) Sinker {
-		return nil
-	})
-	defer sinkerStub.Reset()
-
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			var sink Sink
+			var stub *gostub.Stubs
+
+			if tt.args.sinkUri.SinkTyp == CDCSinkType_MySQL {
+				db, mock, err := sqlmock.New()
+				require.NoError(t, err)
+				defer db.Close()
+
+				// create db + use + create table
+				expectCount := 3
+				if tt.args.dbTblInfo != nil && tt.args.dbTblInfo.IdChanged {
+					// create db + use + drop + create table
+					expectCount = 4
+				}
+
+				for i := 0; i < expectCount; i++ {
+					mock.ExpectExec(fakeSql).WillReturnResult(sqlmock.NewResult(1, 1))
+				}
+
+				sink = &mysqlSink{
+					conn: db,
+				}
+
+				stub = gostub.Stub(&NewMysqlSink, func(_, _, _ string, _, _ int, _ time.Duration, _ string, _ bool) (Sink, error) {
+					return sink, nil
+				})
+				defer stub.Reset()
+			}
+
+			sinkerStub := gostub.Stub(&NewMysqlSinker, func(Sink, uint64, string, *DbTableInfo, *CDCWatermarkUpdater, *plan.TableDef, *ActiveRoutine, uint64, bool) Sinker {
+				return nil
+			})
+			defer sinkerStub.Reset()
+
 			got, err := NewSinker(
 				tt.args.sinkUri,
 				tt.args.accountId,

pkg/cdc/table_scanner.go

@@ -17,6 +17,8 @@ package cdc
 import (
 	"context"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"slices"
 	"strings"
 	"sync"
@@ -46,7 +48,7 @@ var GetTableDetector = func(cnUUID string) *TableDetector {
 	once.Do(func() {
 		detector = &TableDetector{
 			Mp:                   make(map[uint32]TblMap),
-			Callbacks:            make(map[string]func(map[uint32]TblMap)),
+			Callbacks:            make(map[string]TableCallback),
 			exec:                 getSqlExecutor(cnUUID),
 			CallBackAccountId:    make(map[string]uint32),
 			SubscribedAccountIds: make(map[uint32][]string),
@@ -62,11 +64,13 @@ var GetTableDetector = func(cnUUID string) *TableDetector {
 // TblMap key is dbName.tableName, e.g. db1.t1
 type TblMap map[string]*DbTableInfo
 
+type TableCallback func(map[uint32]TblMap) error
+
 type TableDetector struct {
 	sync.Mutex
 
 	Mp        map[uint32]TblMap
-	Callbacks map[string]func(map[uint32]TblMap)
+	Callbacks map[string]TableCallback
 	exec      executor.SQLExecutor
 	cancel    context.CancelFunc
 
@@ -82,9 +86,14 @@ type TableDetector struct {
 	CallBackTableName map[string][]string
 	// tablename -> [taska, taskb ...]
 	SubscribedTableNames map[string][]string
+
+	// to make sure there is at most only one handleNewTables running, so the truncate info will not be lost
+	handling bool
+	lastMp   map[uint32]TblMap
+	mu       sync.Mutex
 }
 
-func (s *TableDetector) Register(id string, accountId uint32, dbs []string, tables []string, cb func(map[uint32]TblMap)) {
+func (s *TableDetector) Register(id string, accountId uint32, dbs []string, tables []string, cb TableCallback) {
 	s.Lock()
 	defer s.Unlock()
 
@@ -173,33 +182,83 @@ func (s *TableDetector) UnRegister(id string) {
 
 func (s *TableDetector) scanTableLoop(ctx context.Context) {
 	logutil.Info("CDC-TableDetector-Scan-Start")
-	defer func() {
-		logutil.Info("CDC-TableDetector-Scan-End")
-	}()
+	defer logutil.Info("CDC-TableDetector-Scan-End")
+
+	ticker := time.NewTicker(15 * time.Second)
+	defer ticker.Stop()
+
+	retryTicker := time.NewTicker(5 * time.Second)
+	defer retryTicker.Stop()
 
-	timeTick := time.Tick(15 * time.Second)
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-timeTick:
-			if err := s.scanTable(); err != nil {
-				logutil.Error(
-					"CDC-TableDetector-Scan-Error",
-					zap.Error(err),
-				)
+		case <-ticker.C:
+			s.mu.Lock()
+
+			if s.handling {
+				s.mu.Unlock()
+				continue
 			}
-			// do callbacks
-			s.Lock()
-			for _, cb := range s.Callbacks {
-				go cb(s.Mp)
+
+			s.mu.Unlock()
+
+			go s.scanAndProcess(ctx)
+		case <-retryTicker.C:
+			s.mu.Lock()
+			if s.handling || s.lastMp == nil {
+				s.mu.Unlock()
+				continue
 			}
-			s.Unlock()
+			s.mu.Unlock()
+
+			go s.processCallback(ctx, s.lastMp)
 		}
 	}
 }
 
+func (s *TableDetector) scanAndProcess(ctx context.Context) {
+	if err := s.scanTable(); err != nil {
+		logutil.Error("CDC-TableDetector-Scan-Error", zap.Error(err))
+		return
+	}
+
+	s.mu.Lock()
+	s.lastMp = s.Mp
+	s.mu.Unlock()
+
+	s.processCallback(ctx, s.lastMp)
+}
+
+func (s *TableDetector) processCallback(ctx context.Context, tables map[uint32]TblMap) {
+	s.mu.Lock()
+	s.handling = true
+	s.mu.Unlock()
+
+	var err error
+	for _, cb := range s.Callbacks {
+		err = cb(tables)
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if err != nil {
+		logutil.Warn("CDC-TableDetector-Callback-Failed", zap.Error(err))
+	} else {
+		logutil.Info("CDC-TableDetector-Callback-Success")
+		s.lastMp = nil
+	}
+
+	s.handling = false
+}
+
 func (s *TableDetector) scanTable() error {
+	if objectio.CDCScanTableErrInjected() {
+		return moerr.NewInternalError(context.Background(), "CDC_SCANTABLE_ERR")
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	var (
@@ -289,14 +348,14 @@ func (s *TableDetector) scanTable() error {
 			if !exists {
 				mp[accountId][key] = newInfo
 			} else {
-				mp[accountId][key] = &DbTableInfo{
-					SourceDbId:      dbId,
-					SourceDbName:    dbName,
-					SourceTblId:     tblId,
-					SourceTblName:   tblName,
-					SourceCreateSql: createSql,
-					IdChanged:       oldInfo.OnlyDiffinTblId(newInfo),
-				}
+				idChanged := oldInfo.OnlyDiffinTblId(newInfo)
+				oldInfo.SourceDbId = dbId
+				oldInfo.SourceDbName = dbName
+				oldInfo.SourceTblId = tblId
+				oldInfo.SourceTblName = tblName
+				oldInfo.SourceCreateSql = createSql
+				oldInfo.IdChanged = oldInfo.IdChanged || idChanged
+				mp[accountId][key] = oldInfo
 			}
 		}
 		return true