Support null-terminated strings in MySQL DDL (#4459)

Trailing `\x00` is the most common case of parse failures and also the
most silly one.

Closes DBI-859

Author: ilidemi

flow/connectors/mysql/cdc.go

@@ -1,6 +1,7 @@
 package connmysql
 
 import (
+	"bytes"
 	"cmp"
 	"context"
 	"crypto/tls"
@@ -46,6 +47,12 @@ func (c *MySqlConnector) binlogStalenessThreshold() time.Duration {
 	return binlogStalenessMultiplier * c.binlogHeartbeatPeriod
 }
 
+func parseSQL(parser *parser.Parser, query []byte) ([]ast.StmtNode, []error, error) {
+	// TIDB parser errors on null-terminated strings
+	trimmedQuery := shared.UnsafeFastReadOnlyBytesToString(bytes.TrimRight(query, "\x00"))
+	return parser.ParseSQL(trimmedQuery)
+}
+
 func (c *MySqlConnector) GetTableSchema(
 	ctx context.Context,
 	env map[string]string,
@@ -571,7 +578,7 @@ func (c *MySqlConnector) PullRecords(
 			if mysqlParser == nil {
 				mysqlParser = parser.New()
 			}
-			stmts, warns, err := mysqlParser.ParseSQL(shared.UnsafeFastReadOnlyBytesToString(ev.Query))
+			stmts, warns, err := parseSQL(mysqlParser, ev.Query)
 			if err != nil {
 				c.logger.Warn("failed to parse QueryEvent", slog.String("query", string(ev.Query)), slog.Any("error", err))
 				break

flow/connectors/mysql/cdc_test.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/pingcap/tidb/pkg/parser"
+	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/stretchr/testify/require"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -50,6 +52,43 @@ func createTestDB(t *testing.T, ctx context.Context, c *MySqlConnector, dbName s
 	})
 }
 
+func TestParseSQLParsesTrailingNull(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		query  []byte
+		assert func(*testing.T, ast.StmtNode)
+	}{
+		{
+			name:  "alter table",
+			query: []byte("ALTER TABLE t ADD COLUMN c INT\x00"),
+			assert: func(t *testing.T, stmt ast.StmtNode) {
+				t.Helper()
+				require.IsType(t, &ast.AlterTableStmt{}, stmt)
+			},
+		},
+		{
+			name:  "commit",
+			query: []byte("COMMIT\x00"),
+			assert: func(t *testing.T, stmt ast.StmtNode) {
+				t.Helper()
+				require.IsType(t, &ast.CommitStmt{}, stmt)
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			mysqlParser := parser.New()
+			_, _, err := mysqlParser.ParseSQL(string(tc.query))
+			require.Error(t, err)
+
+			stmts, warns, err := parseSQL(mysqlParser, tc.query)
+			require.NoError(t, err)
+			require.Empty(t, warns)
+			require.Len(t, stmts, 1)
+			tc.assert(t, stmts[0])
+		})
+	}
+}
+
 func TestGetTableSchemaCaseSensitiveIdentifiers(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()