handle commit for non transactional mysql engines

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -550,6 +550,18 @@ func (c *MySqlConnector) PullRecords(
 		}
 	}
 
+	advanceCheckpoint := func(evGSet mysql.GTIDSet, logPos uint32) {
+		if gset != nil {
+			gset = evGSet
+			updatedOffset = gset.String()
+			req.RecordStream.UpdateLatestCheckpointText(updatedOffset)
+		} else if logPos > pos.Pos {
+			pos.Pos = logPos
+			updatedOffset = posToOffsetText(pos)
+			req.RecordStream.UpdateLatestCheckpointText(updatedOffset)
+		}
+	}
+
 	lastEventAt := time.Now()
 	var mysqlParser *parser.Parser
 	processEvent := func(event *replication.BinlogEvent) error {
@@ -559,15 +571,7 @@ func (c *MySqlConnector) PullRecords(
 		case *replication.GtidTaggedLogEvent: // MySQL 8.4+ tagged GTIDs
 			recordCommitLagMetric(ctx, ev.ImmediateCommitTimestamp)
 		case *replication.XIDEvent:
-			if gset != nil {
-				gset = ev.GSet
-				updatedOffset = gset.String() // accumulated from GTID events above
-				req.RecordStream.UpdateLatestCheckpointText(updatedOffset)
-			} else if event.Header.LogPos > pos.Pos {
-				pos.Pos = event.Header.LogPos
-				updatedOffset = posToOffsetText(pos)
-				req.RecordStream.UpdateLatestCheckpointText(updatedOffset)
-			}
+			advanceCheckpoint(ev.GSet, event.Header.LogPos)
 			inTx = false
 		case *replication.RotateEvent:
 			if gset == nil && (event.Header.Timestamp != 0 || string(ev.NextLogName) != pos.Name) {
@@ -624,6 +628,10 @@ func (c *MySqlConnector) PullRecords(
 					}
 				case ddlKindRenameTable:
 					c.processRenameTableQuery(ctx, otelManager, req, renameStmt, string(ev.Schema))
+				case ddlKindCommit:
+					// Non-transactional engines (e.g. MyISAM) end a binlog group with a COMMIT/ROLLBACK
+					advanceCheckpoint(ev.GSet, event.Header.LogPos)
+					inTx = false
 				case ddlKindIgnored:
 				default:
 					return fmt.Errorf("unknown stmt kind: %v", kind)

flow/connectors/mysql/query_parser.go

@@ -222,6 +222,7 @@ const (
 	ddlKindIgnored ddlKind = iota
 	ddlKindAlterTable
 	ddlKindRenameTable
+	ddlKindCommit
 )
 
 // classifyParsedStatement maps a successfully parsed statement to the handler that
@@ -235,6 +236,8 @@ func classifyParsedStatement(stmt ast.StmtNode) (ddlKind, *ast.AlterTableStmt, *
 		return ddlKindAlterTable, s, nil
 	case *ast.RenameTableStmt:
 		return ddlKindRenameTable, nil, s
+	case *ast.CommitStmt, *ast.RollbackStmt:
+		return ddlKindCommit, nil, nil
 	default:
 		return ddlKindIgnored, nil, nil
 	}

flow/connectors/mysql/query_parser_test.go

@@ -3,6 +3,7 @@ package connmysql
 import (
 	"testing"
 
+	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/stretchr/testify/require"
 )
 
@@ -185,3 +186,25 @@ func TestIsBenignUnparsedStatement(t *testing.T) {
 		})
 	}
 }
+
+func TestClassifyParsedStatementTxControl(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		query string
+		want  ddlKind
+	}{
+		{name: "commit", query: "COMMIT", want: ddlKindCommit},
+		{name: "rollback", query: "ROLLBACK", want: ddlKindCommit},
+		{name: "begin", query: "BEGIN", want: ddlKindIgnored},
+		{name: "alter table add column", query: "ALTER TABLE t ADD COLUMN c INT", want: ddlKindAlterTable},
+		{name: "rename table", query: "RENAME TABLE a TO b", want: ddlKindRenameTable},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			stmts, _, err := parseSQL(parser.New(), []byte(tc.query))
+			require.NoError(t, err)
+			require.Len(t, stmts, 1)
+			kind, _, _ := classifyParsedStatement(stmts[0])
+			require.Equal(t, tc.want, kind)
+		})
+	}
+}

flow/e2e/clickhouse_mysql_test.go

@@ -81,6 +81,50 @@ func (s ClickHouseSuite) Test_UnsignedMySQL() {
 	RequireEnvCanceled(s.t, env)
 }
 
+func (s ClickHouseSuite) Test_MySQL_MyISAM_NonTransactional() {
+	if _, ok := s.source.(*MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	srcTableName := "test_myisam"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_myisam"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`CREATE TABLE %s (
+		id int primary key,
+		val text
+	) ENGINE=MyISAM`, srcFullName)))
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id, val) VALUES (1, 'snapshot')`, srcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      srcFullName,
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial snapshot", srcTableName, dstTableName, "id,val")
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`INSERT INTO %s (id, val) VALUES (2, 'cdc-insert')`, srcFullName)))
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`UPDATE %s SET val = 'cdc-update' WHERE id = 1`, srcFullName)))
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf(`DELETE FROM %s WHERE id = 2`, srcFullName)))
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc changes", srcTableName, dstTableName, "id,val")
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_MySQL_Time() {
 	if _, ok := s.source.(*MySqlSource); !ok {
 		s.t.Skip("only applies to mysql")