split commit and rollback DDLs

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -628,7 +628,7 @@ func (c *MySqlConnector) PullRecords(
 					}
 				case ddlKindRenameTable:
 					c.processRenameTableQuery(ctx, otelManager, req, renameStmt, string(ev.Schema))
-				case ddlKindCommit:
+				case ddlKindCommit, ddlKindRollback:
 					// Non-transactional engines (e.g. MyISAM) end a binlog group with a COMMIT/ROLLBACK
 					advanceCheckpoint(ev.GSet, event.Header.LogPos)
 					inTx = false

flow/connectors/mysql/query_parser.go

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

flow/connectors/mysql/query_parser_test.go

@@ -194,7 +194,7 @@ func TestClassifyParsedStatementTxControl(t *testing.T) {
 		want  ddlKind
 	}{
 		{name: "commit", query: "COMMIT", want: ddlKindCommit},
-		{name: "rollback", query: "ROLLBACK", want: ddlKindCommit},
+		{name: "rollback", query: "ROLLBACK", want: ddlKindRollback},
 		{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},