fix cdc sinker and handle new table (#22390)

fix cdc sinker and handle new table

Approved by: @ck89119, @daviszhen, @XuPeng-SH

Author: jiangxinmeng1

pkg/cdc/sinker.go

@@ -855,6 +855,10 @@ var NewMysqlSink = func(
 	return ret, err
 }
 
+func (s *mysqlSink) Reset() {
+	s.tx = nil
+}
+
 func (s *mysqlSink) recordTxnSQL(sqlBuf []byte) {
 	if !s.debugTxnRecorder.doRecord {
 		return
@@ -938,17 +942,11 @@ func (s *mysqlSink) SendBegin(ctx context.Context) (err error) {
 
 func (s *mysqlSink) SendCommit(_ context.Context) error {
 	s.resetRecordedTxn()
-	defer func() {
-		s.tx = nil
-	}()
 	return s.tx.Commit()
 }
 
 func (s *mysqlSink) SendRollback(_ context.Context) error {
 	s.resetRecordedTxn()
-	defer func() {
-		s.tx = nil
-	}()
 	return s.tx.Rollback()
 }
 

pkg/cdc/types.go

@@ -153,6 +153,7 @@ type Sink interface {
 	SendBegin(ctx context.Context) error
 	SendCommit(ctx context.Context) error
 	SendRollback(ctx context.Context) error
+	Reset()
 	Close()
 }
 

pkg/cdc/util.go

@@ -639,6 +639,8 @@ var GetTableDef = func(
 	cnEngine engine.Engine,
 	tblId uint64,
 ) (*plan.TableDef, error) {
+	ctx, cancel := context.WithTimeout(ctx, time.Minute*5)
+	defer cancel()
 	_, _, rel, err := cnEngine.GetRelationById(ctx, txnOp, tblId)
 	if err != nil {
 		return nil, err

pkg/frontend/cdc_exector.go

@@ -411,8 +411,6 @@ func (exec *CDCTaskExecutor) handleNewTables(allAccountTbls map[uint32]cdc.TblMa
 
 	accountId := uint32(exec.spec.Accounts[0].GetId())
 	ctx := defines.AttachAccountId(context.Background(), accountId)
-	ctx, cancel := context.WithTimeout(ctx, time.Minute*5)
-	defer cancel()
 
 	txnOp, err := cdc.GetTxnOp(ctx, exec.cnEngine, exec.cnTxnClient, "cdc-handleNewTables")
 	if err != nil {

pkg/vm/engine/test/cdc_sinker_test.go

@@ -0,0 +1,100 @@
+// Copyright 2024 Matrix Origin
+//
+// 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,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package test
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/cdc"
+	"github.com/matrixorigin/matrixone/pkg/defines"
+	catalog2 "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/test/testutil"
+	"github.com/prashantv/gostub"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCDC_Sinker1(t *testing.T) {
+	var mock sqlmock.Sqlmock
+	mockFn := func(_, _, _ string, _ int, _ string) (db *sql.DB, err error) {
+		db, mock, err = sqlmock.New()
+		return
+	}
+	stub := gostub.Stub(&cdc.OpenDbConn, mockFn)
+	defer stub.Reset()
+
+	sink, err := cdc.NewMysqlSink(
+		"root",
+		"123456",
+		"127.0.0.1",
+		3306,
+		3,
+		3*time.Second,
+		cdc.CDCDefaultSendSqlTimeout,
+		false,
+	)
+	require.NoError(t, err)
+	defer sink.Close()
+
+	ctx := context.Background()
+
+	mock.ExpectBegin()
+	err = sink.SendBegin(ctx)
+	require.NoError(t, err)
+	mock.ExpectCommit()
+	err = sink.SendCommit(ctx)
+	require.NoError(t, err)
+	mock.ExpectRollback()
+	err = sink.SendRollback(ctx)
+	require.Error(t, err)
+	sink.Reset()
+}
+
+func TestCDCUtil1(t *testing.T) {
+	catalog.SetupDefines("")
+
+	var (
+		accountId    = catalog.System_Account
+		tableName    = "test1"
+		databaseName = "db1"
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ctx = context.WithValue(ctx, defines.TenantIDKey{}, accountId)
+
+	disttaeEngine, taeHandler, rpcAgent, _ := testutil.CreateEngines(ctx, testutil.TestOptions{}, t)
+	defer func() {
+		disttaeEngine.Close(ctx)
+		taeHandler.Close(true)
+		rpcAgent.Close()
+	}()
+	schema := catalog2.MockSchemaAll(10, 0)
+	schema.Name = tableName
+	ctx, cancel = context.WithTimeout(ctx, time.Minute*5)
+	defer cancel()
+	_, rel, err := disttaeEngine.CreateDatabaseAndTable(ctx, databaseName, tableName, schema)
+	id := rel.GetTableID(context.Background())
+	require.NoError(t, err)
+
+	txn, err := disttaeEngine.NewTxnOperator(ctx, disttaeEngine.Now())
+	require.NoError(t, err)
+	_, err = cdc.GetTableDef(ctx, txn, disttaeEngine.Engine, id)
+	require.NoError(t, err)
+}