Support ANSI_QUOTES in MySQL/MariaDB DDL (#4452)

Parse out status vars from query event and pass the flag to the tidb
parser. Add an integration test that can work with binlog more freely as
we can't really observe query parsing from e2e (besides the 5.7-pos
configuration that doesn't have binlog metadata to fall back on, but we
need a maria test as well).

As a bonus, the integration test runs in 1.2 seconds.

Closes DBI-821

Author: ilidemi

.claude/skills/tilt/SKILL.md

@@ -86,7 +86,9 @@ All e2e tests also depend on core PeerDB services: `flow-api`, `flow-worker`, `c
 |----------|---------|----------------------|
 | `connector_postgres` | `connectors/postgres` | postgres |
 | `connector_mongo` | `connectors/mongo` | mongodb |
-| `connector_mysql` | `connectors/mysql` | mysql-gtid |
+| `connector_mysql-gtid` | `connectors/mysql` | mysql-gtid |
+| `connector_mysql-pos` | `connectors/mysql` | mysql-pos |
+| `connector_mariadb` | `connectors/mysql` | mariadb |
 | `connector_clickhouse` | `connectors/clickhouse` | clickhouse |
 
 Connector tests only depend on `catalog` + their specific DB.

.env.example

@@ -46,6 +46,7 @@ AWS_ACCESS_KEY_ID=_peerdb_minioadmin
 AWS_SECRET_ACCESS_KEY=_peerdb_minioadmin
 AWS_ENDPOINT_URL_S3=http://host.docker.internal:9001
 AWS_EC2_METADATA_DISABLED="true"
+FLOW_TESTS_RDS_IAM_AUTH_SKIP=true
 
 MINIO_ROOT_USER=_peerdb_minioadmin
 MINIO_ROOT_PASSWORD=_peerdb_minioadmin

Tiltfile

@@ -295,10 +295,10 @@ def e2e_test(name, test_run, extra_deps=[], vars_overrides={}):
         allow_parallel=True,
     )
 
-def connector_test(connector, extra_deps=[], vars_overrides={}):
+def connector_test(connector, extra_deps=[], vars_overrides={}, name=''):
     overrides_str = ' '.join(['%s=%s' % (var, value) for var, value in vars_overrides.items()])
     local_resource(
-        'connector_' + connector,
+        'connector_' + (name or connector),
         cmd='cd flow && %s go test -count=1 -v ./connectors/%s/...' % (overrides_str, connector),
         labels=['Test'],
         auto_init=False,
@@ -369,6 +369,11 @@ e2e_test('api-mongodb', 'TestApiMongo', ['provision-mongodb'])
 # Connectors tests
 
 connector_test('postgres', ['provision-postgres'])
+
+connector_test('mysql', ['provision-mysql-gtid'], vars_overrides=mysql_gtid_vars, name='mysql-gtid')
+connector_test('mysql', ['provision-mysql-pos'], vars_overrides=mysql_pos_vars, name='mysql-pos')
+connector_test('mysql', ['provision-mariadb'], vars_overrides=mariadb_vars, name='mariadb')
+
 connector_test('mongo', ['provision-mongodb'])
-connector_test('mysql', ['provision-mysql-gtid'])
+
 connector_test('clickhouse', ['provision-clickhouse'])

ancillary-docker-compose.yml

@@ -185,14 +185,15 @@ services:
       - 9902:9902
       - 9904:9904
       - 9903:9903
-      - 42001:42001
-      - 42002:42002
-      - 42003:42003
-      - 42004:42004
-      - 42005:42005
-      - 49001:49001
-      - 49002:49002
-      - 49003:49003
+      - 10001:10001
+      - 12001:12001
+      - 12002:12002
+      - 12003:12003
+      - 12004:12004
+      - 12005:12005
+      - 14001:14001
+      - 14002:14002
+      - 14003:14003
     extra_hosts:
       - "host.docker.internal:host-gateway"
     healthcheck:

flow/connectors/mysql/cdc.go

@@ -19,6 +19,7 @@ import (
 	"github.com/go-mysql-org/go-mysql/replication"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
+	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
 	_ "github.com/pingcap/tidb/pkg/types/parser_driver"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -44,6 +45,48 @@ const (
 	binlogStalenessMultiplier    = 3
 )
 
+const (
+	queryStatusVarFlags2  = 0
+	queryStatusVarSQLMode = 1
+)
+
+// sqlModeFromStatusVars extracts the sql_mode bitmask from a QueryEvent's status vars.
+// Both MySQL and MariaDB guarantee status vars are written in increasing order,
+// so we only need to parse 0 and 1 to get to 1
+// https://dev.mysql.com/doc/dev/mysql-server/latest/classmysql_1_1binlog_1_1event_1_1Query__event.html#Query_event_binary_format
+// https://github.com/MariaDB/server/blob/c3ec2dc368a8c7165cdbea58208eb828e76ebc57/sql/log_event_server.cc#L1083-L1087
+func sqlModeFromStatusVars(statusVars []byte) (uint64, bool) {
+	for pos := 0; pos < len(statusVars); {
+		code := statusVars[pos]
+		pos++
+
+		switch code {
+		case queryStatusVarFlags2:
+			pos += 4
+		case queryStatusVarSQLMode:
+			if pos+8 > len(statusVars) {
+				return 0, false
+			}
+			return binary.LittleEndian.Uint64(statusVars[pos : pos+8]), true
+		default:
+			return 0, false
+		}
+
+		if pos > len(statusVars) {
+			return 0, false
+		}
+	}
+	return 0, false
+}
+
+func setParserSQLModeFromStatusVars(mysqlParser *parser.Parser, statusVars []byte) {
+	var sqlMode tidbmysql.SQLMode
+	if mode, ok := sqlModeFromStatusVars(statusVars); ok && mode&uint64(tidbmysql.ModeANSIQuotes) != 0 {
+		sqlMode = tidbmysql.ModeANSIQuotes
+	}
+	mysqlParser.SetSQLMode(sqlMode)
+}
+
 func (c *MySqlConnector) binlogStalenessThreshold() time.Duration {
 	return binlogStalenessMultiplier * c.binlogHeartbeatPeriod
 }
@@ -533,6 +576,7 @@ func (c *MySqlConnector) PullRecords(
 			if mysqlParser == nil {
 				mysqlParser = parser.New()
 			}
+			setParserSQLModeFromStatusVars(mysqlParser, ev.StatusVars)
 			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))

flow/connectors/mysql/cdc_test.go

@@ -3,53 +3,119 @@ package connmysql
 import (
 	"context"
 	"fmt"
+	"slices"
+	"strings"
 	"testing"
+	"time"
 
+	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/google/uuid"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
+	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/stretchr/testify/require"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
 
-func newTestConnector(t *testing.T, ctx context.Context) *MySqlConnector {
+func startBinlogStream(t *testing.T, ctx context.Context, c *MySqlConnector) *replication.BinlogStreamer {
 	t.Helper()
-	flavor := protos.MySqlFlavor_MYSQL_MYSQL
-	if internal.MySQLTestVersionIsMaria() {
-		flavor = protos.MySqlFlavor_MYSQL_MARIA
+
+	var syncer *replication.BinlogSyncer
+	var stream *replication.BinlogStreamer
+	if internal.MySQLTestVersionIsMySQLPos() {
+		pos, err := c.GetMasterPos(ctx)
+		require.NoError(t, err)
+		syncer, stream, _, _, err = c.startCdcStreamingFilePos(ctx, pos, nil)
+		require.NoError(t, err)
+	} else {
+		gset, err := c.GetMasterGTIDSet(ctx)
+		require.NoError(t, err)
+		syncer, stream, _, _, err = c.startCdcStreamingGtid(ctx, gset, nil)
+		require.NoError(t, err)
 	}
-	connector, err := NewMySqlConnector(ctx, &protos.MySqlConfig{
-		Host:       internal.MySQLTestHost(),
-		Port:       internal.MySQLTestPort(),
-		User:       "root",
-		Password:   internal.MySQLTestRootPasswordWithFallback("cipass"),
-		Database:   "mysql",
-		DisableTls: true,
-		Flavor:     flavor,
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		if err := connector.Close(); err != nil {
-			t.Logf("connector close failed: %v", err)
-		}
-	})
-	return connector
+	t.Cleanup(syncer.Close)
+	return stream
 }
 
-func createTestDB(t *testing.T, ctx context.Context, c *MySqlConnector, dbName string) {
+func waitForQueryEventContaining(
+	t *testing.T, ctx context.Context, stream *replication.BinlogStreamer, marker string,
+) ([]byte, string) {
 	t.Helper()
-	quoted := "`" + dbName + "`"
-	_, err := c.Execute(ctx, "DROP DATABASE IF EXISTS "+quoted)
+
+	streamCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	for {
+		event, err := stream.GetEvent(streamCtx)
+		require.NoError(t, err, "did not observe QueryEvent containing %q in the binlog", marker)
+		if qe, ok := event.Event.(*replication.QueryEvent); ok && strings.Contains(string(qe.Query), marker) {
+			// status vars alias into the event buffer, which go-mysql may reuse
+			return slices.Clone(qe.StatusVars), string(qe.Query)
+		}
+	}
+}
+
+func TestANSIQuotesDDLParsedFromBinlog(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+	connector := newTestConnector(t, ctx)
+
+	dbName := testDBName("ansi_quotes_binlog")
+	createTestDB(t, ctx, connector, dbName)
+	_, err := connector.Execute(ctx, "CREATE TABLE `"+dbName+"`.`t` (id INT PRIMARY KEY)")
 	require.NoError(t, err)
-	_, err = c.Execute(ctx, "CREATE DATABASE "+quoted)
+
+	stream := startBinlogStream(t, ctx, connector)
+
+	// ANSI_QUOTES makes "..." an identifier quote, so the ALTER below only parses
+	// when the mode is passed to the parser as well
+	guid := uuid.NewString()
+	_, err = connector.Execute(ctx, `SET SESSION sql_mode='ANSI_QUOTES'`)
 	require.NoError(t, err)
-	t.Cleanup(func() {
-		if _, err := c.Execute(context.Background(), "DROP DATABASE IF EXISTS "+quoted); err != nil {
-			t.Logf("drop test db %s failed: %v", dbName, err)
-		}
-	})
+
+	ddl := fmt.Sprintf(`/* %s */ ALTER TABLE "%s"."t" ADD COLUMN "c1" INT`, guid, dbName)
+	_, err = connector.Execute(ctx, ddl)
+	require.NoError(t, err)
+
+	statusVars, query := waitForQueryEventContaining(t, ctx, stream, guid)
+
+	// Validate extraction
+	mode, ok := sqlModeFromStatusVars(statusVars)
+	require.True(t, ok)
+	require.NotZero(t, mode&uint64(tidbmysql.ModeANSIQuotes), "ANSI_QUOTES missing from binlog status vars")
+
+	// Validate parsing
+	mysqlParser := parser.New()
+	setParserSQLModeFromStatusVars(mysqlParser, statusVars)
+	stmts, _, err := mysqlParser.ParseSQL(query)
+	require.NoError(t, err)
+	require.Len(t, stmts, 1)
+	require.IsType(t, &ast.AlterTableStmt{}, stmts[0])
+
+	// Validate parsing fails without the mode
+	_, _, err = parser.New().ParseSQL(query)
+	require.Error(t, err)
+
+	// Status vars with codes > 1 are written after sql_mode (code 1). Every
+	// QueryEvent already carries catalog (2) and charset (4); force a few more —
+	// auto_increment (3), time_zone (5), lc_time_names (7) — to confirm a fuller
+	// status-var block still doesn't interfere with extracting sql_mode.
+	_, err = connector.Execute(ctx,
+		`SET SESSION auto_increment_increment=7, auto_increment_offset=3, time_zone='+05:00', lc_time_names='fr_FR'`)
+	require.NoError(t, err)
+
+	guid = uuid.NewString()
+	ddl = fmt.Sprintf(`/* %s */ ALTER TABLE "%s"."t" ADD COLUMN "c2" INT`, guid, dbName)
+	_, err = connector.Execute(ctx, ddl)
+	require.NoError(t, err)
+
+	statusVars, _ = waitForQueryEventContaining(t, ctx, stream, guid)
+
+	mode, ok = sqlModeFromStatusVars(statusVars)
+	require.True(t, ok)
+	require.NotZero(t, mode&uint64(tidbmysql.ModeANSIQuotes), "ANSI_QUOTES missing from binlog status vars with extra status vars present")
 }
 
 func TestParseSQLParsesTrailingNull(t *testing.T) {
@@ -101,8 +167,8 @@ func TestGetTableSchemaCaseSensitiveIdentifiers(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int64(0), lctn)
 
-	lowerDB := "cs"
-	upperDB := "CS_IDS_TEST"
+	lowerDB := testDBName("cs")
+	upperDB := testDBName("CS_IDS_TEST")
 	createTestDB(t, ctx, connector, lowerDB)
 	createTestDB(t, ctx, connector, upperDB)
 
@@ -151,7 +217,7 @@ func TestGetTableSchemaPrimaryKeyVariants(t *testing.T) {
 	ctx := t.Context()
 	connector := newTestConnector(t, ctx)
 
-	dbName := "pk_variants_test"
+	dbName := testDBName("pk_variants")
 	createTestDB(t, ctx, connector, dbName)
 
 	exec := func(sql string) {

flow/connectors/mysql/helpers_test.go

@@ -0,0 +1,45 @@
+package connmysql
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/pkg/common"
+)
+
+func newTestConnector(t *testing.T, ctx context.Context) *MySqlConnector {
+	t.Helper()
+	flavor, mechanism := internal.MySQLTestFlavorAndMechanism(t)
+	config := internal.GetMySQLConfigFromEnv(flavor, mechanism)
+	connector, err := NewMySqlConnector(ctx, config)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := connector.Close(); err != nil {
+			t.Logf("connector close failed: %v", err)
+		}
+	})
+	require.NoError(t, ConfigureReplication(t, connector, config))
+	return connector
+}
+
+func createTestDB(t *testing.T, ctx context.Context, c *MySqlConnector, dbName string) {
+	t.Helper()
+	quoted := "`" + dbName + "`"
+	_, err := c.Execute(ctx, "DROP DATABASE IF EXISTS "+quoted)
+	require.NoError(t, err)
+	_, err = c.Execute(ctx, "CREATE DATABASE "+quoted)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if _, err := c.Execute(context.Background(), "DROP DATABASE IF EXISTS "+quoted); err != nil {
+			t.Logf("drop test db %s failed: %v", dbName, err)
+		}
+	})
+}
+
+func testDBName(mnemonic string) string {
+	return mnemonic + "_" + strings.ToLower(common.RandomString(8))
+}

flow/connectors/mysql/rds_test.go

@@ -10,6 +10,7 @@ import (
 )
 
 func TestAwsRDSIAMAuthConnectForMYSQL(t *testing.T) {
+	internal.SkipRDSIAMAuthIfRequested(t)
 	internal.SetupRDSIAMAuthAWSCredentials(t)
 	conn := internal.RDSIAMAuthMySQLTestConnectionInfo(t)
 	mysqlConnector, err := NewMySqlConnector(t.Context(),
@@ -27,6 +28,7 @@ func TestAwsRDSIAMAuthConnectForMYSQL(t *testing.T) {
 }
 
 func TestAwsRDSIAMAuthConnectForMYSQLViaProxy(t *testing.T) {
+	internal.SkipRDSIAMAuthIfRequested(t)
 	internal.SetupRDSIAMAuthAWSCredentials(t)
 	conn := internal.RDSIAMAuthMySQLTestConnectionInfo(t)
 	mysqlConnector, err := NewMySqlConnector(t.Context(),