From a98a69660a4052da0489b76ebed371bd8cd6d703 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 2 Jul 2026 13:23:40 +0200 Subject: [PATCH 01/10] fail loudly on partial row maria event --- flow/connectors/mysql/cdc.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 8a3132b7a..cdba18b19 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -45,6 +45,13 @@ const ( binlogStalenessMultiplier = 3 ) +// mariadbPartialRowDataEvent is MariaDB 12.3+ PARTIAL_ROW_DATA_EVENT (Log_event_type = 172). +// It fragments an oversized rows event across several binlog events, which a consumer must buffer +// and reassemble before decoding. go-mysql lib has no constant or parser for it, so it +// surfaces as a GenericEvent; we don't reassemble fragments yet, so we fail loudly rather than +// silently dropping the row change. +const mariadbPartialRowDataEvent replication.EventType = 172 + const ( queryStatusVarFlags2 = 0 queryStatusVarSQLMode = 1 @@ -573,6 +580,11 @@ func (c *MySqlConnector) PullRecords( slog.Uint64("incident", uint64(incident)), slog.String("message", message)) return exceptions.NewMySQLBinlogIncidentError(incident, message) } + if event.Header.EventType == mariadbPartialRowDataEvent { + return fmt.Errorf( + "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d): fragmented oversized row events are not supported", + uint64(mariadbPartialRowDataEvent)) + } case *replication.QueryEvent: if !inTx && gset == nil && event.Header.LogPos > pos.Pos { pos.Pos = event.Header.LogPos From ce8eca66b0e7395c4c975fd2480423a8b03b8c85 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 2 Jul 2026 13:57:46 +0200 Subject: [PATCH 02/10] - classify error - add metric to count unsupported event types --- flow/alerting/classifier.go | 7 +++++++ flow/connectors/mysql/cdc.go | 22 +++++++++++++++------- flow/otel_metrics/attributes.go | 1 + flow/otel_metrics/otel_manager.go | 10 ++++++++++ flow/shared/exceptions/mysql.go | 14 ++++++++++++++ 5 files changed, 47 insertions(+), 7 deletions(-) diff --git a/flow/alerting/classifier.go b/flow/alerting/classifier.go index 5ad5ba3d2..3ceabc5a2 100644 --- a/flow/alerting/classifier.go +++ b/flow/alerting/classifier.go @@ -1154,6 +1154,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) { } } + if _, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok { + return ErrorNotifyBinlogInvalid, ErrorInfo{ + Source: ErrorSourceMySQL, + Code: "UNSUPPORTED_PARTIAL_ROW_EVENT", + } + } + if mysqlGeometryParseError, ok := errors.AsType[*exceptions.MySQLGeometryParseError](err); ok && strings.Contains(mysqlGeometryParseError.Error(), mysqlGeometryLinearRingNotClosedError) { return ErrorUnsupportedDatatype, ErrorInfo{ diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index cdba18b19..f1baf6bd8 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -12,6 +12,7 @@ import ( "log/slog" "math/rand/v2" "slices" + "strconv" "sync/atomic" "time" @@ -573,17 +574,24 @@ func (c *MySqlConnector) PullRecords( c.logger.Info("rotate", slog.String("name", pos.Name), slog.Uint64("pos", uint64(pos.Pos))) } case *replication.GenericEvent: - // INCIDENT_EVENT (LOST_EVENTS) - fail and require resync - if event.Header.EventType == replication.INCIDENT_EVENT { + switch event.Header.EventType { + case replication.INCIDENT_EVENT: + // INCIDENT_EVENT (LOST_EVENTS) - fail and require resync incident, message := parseIncidentEvent(ev.Data) c.logger.Error("[mysql] received binlog incident event, resync required", slog.Uint64("incident", uint64(incident)), slog.String("message", message)) return exceptions.NewMySQLBinlogIncidentError(incident, message) - } - if event.Header.EventType == mariadbPartialRowDataEvent { - return fmt.Errorf( - "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d): fragmented oversized row events are not supported", - uint64(mariadbPartialRowDataEvent)) + case mariadbPartialRowDataEvent: + // MariaDB 12.3+ fragments an oversized rows event across multiple events; we don't + // reassemble fragments yet, so the row change is unrecoverable and a resync is required. + c.logger.Error("[mysql] received MariaDB partial row data event, resync required", + slog.Uint64("eventType", uint64(mariadbPartialRowDataEvent))) + return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent)) + default: + c.logger.Warn("unknown generic event", slog.Any("type", event.Header.EventType)) + otelManager.Metrics.UnsupportedBinlogEventCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet( + attribute.String(otel_metrics.BinlogEventTypeKey, strconv.Itoa(int(event.Header.EventType))), + ))) } case *replication.QueryEvent: if !inTx && gset == nil && event.Header.LogPos > pos.Pos { diff --git a/flow/otel_metrics/attributes.go b/flow/otel_metrics/attributes.go index 67af98133..46827ae52 100644 --- a/flow/otel_metrics/attributes.go +++ b/flow/otel_metrics/attributes.go @@ -50,6 +50,7 @@ const ( TypeChangeToKey = "to" SourceEventTypeKey = "sourceEventType" OnlineSchemaMigrationTool = "tool" + BinlogEventTypeKey = "eventType" ) const ( diff --git a/flow/otel_metrics/otel_manager.go b/flow/otel_metrics/otel_manager.go index 1753d0f36..6f14033fc 100644 --- a/flow/otel_metrics/otel_manager.go +++ b/flow/otel_metrics/otel_manager.go @@ -83,6 +83,7 @@ const ( UsedMySQLCharsetsName = "used_mysql_charsets" ColumnTypeChangesName = "column_type_changes" OnlineSchemaMigrationsName = "online_schema_migrations" + UnsupportedBinlogEventName = "unsupported_binlog_event" ) type Metrics struct { @@ -140,6 +141,7 @@ type Metrics struct { UsedMySQLCharsetsCounter metric.Int64Counter ColumnTypeChangesCounter metric.Int64Counter OnlineSchemaMigrationsCounter metric.Int64Counter + UnsupportedBinlogEventCounter metric.Int64Counter } type SlotMetricGauges struct { @@ -603,6 +605,14 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error { return err } + if om.Metrics.UnsupportedBinlogEventCounter, err = om.GetOrInitInt64Counter(BuildMetricName(UnsupportedBinlogEventName), + metric.WithDescription( + "Counter of unsupported binlog generic events seen on the CDC path, with `eventType` label "+ + "holding the numeric binlog event type"), + ); err != nil { + return err + } + if CodeNotificationCounter, err = om.GetOrInitInt64Counter(BuildMetricName(CodeNotificationCounterName), metric.WithDescription("One-off notifications with unique `message` attribute, triggers generic non-paging alert"), ); err != nil { diff --git a/flow/shared/exceptions/mysql.go b/flow/shared/exceptions/mysql.go index 907b6ee31..e6a68c102 100644 --- a/flow/shared/exceptions/mysql.go +++ b/flow/shared/exceptions/mysql.go @@ -153,3 +153,17 @@ func (e *MySQLBinlogIncidentError) Error() string { } return fmt.Sprintf("MySQL binlog incident event received (incident=%d); a resync is required", e.Incident) } + +type MySQLUnsupportedPartialRowEventError struct { + EventType byte +} + +func NewMySQLUnsupportedPartialRowEventError(eventType byte) *MySQLUnsupportedPartialRowEventError { + return &MySQLUnsupportedPartialRowEventError{EventType: eventType} +} + +func (e *MySQLUnsupportedPartialRowEventError) Error() string { + return fmt.Sprintf( + "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d): fragmented oversized row events cannot be processed; a resync is required", + e.EventType) +} From 50ae9f74dda06f6a51d612f0779481d12f1f2b38 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 2 Jul 2026 14:11:53 +0200 Subject: [PATCH 03/10] e2e test for partial row event + test container helper for e2e tests --- flow/e2e/clickhouse_mysql_test.go | 129 +++++++++++++++++++----------- flow/e2e/mysql.go | 91 +++++++++++++++++++++ 2 files changed, 172 insertions(+), 48 deletions(-) diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index 7d7a45bc8..b2a7cf6e8 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -13,15 +13,12 @@ import ( "github.com/go-mysql-org/go-mysql/mysql" "github.com/google/uuid" "github.com/stretchr/testify/require" - "github.com/testcontainers/testcontainers-go" - "github.com/testcontainers/testcontainers-go/wait" "github.com/PeerDB-io/peerdb/flow/connectors" connclickhouse "github.com/PeerDB-io/peerdb/flow/connectors/clickhouse" "github.com/PeerDB-io/peerdb/flow/generated/protos" "github.com/PeerDB-io/peerdb/flow/internal" "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse" - "github.com/PeerDB-io/peerdb/flow/pkg/common" mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql" "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/types" @@ -1625,57 +1622,18 @@ func (s ClickHouseSuite) Test_MySQL_BinlogIncident() { s.t.Skip("binlog incident injection requires a MySQL debug build; not available for MariaDB") } - req := testcontainers.ContainerRequest{ - Image: "ghcr.io/peerdb-io/mysql-debug:8.0.46", - Env: map[string]string{ - "MYSQL_ROOT_PASSWORD": internal.MySQLTestRootPasswordWithFallback("cipass"), - "MYSQL_ROOT_HOST": "%", - }, - // Keep the debug server small for the one-off testcontainers - Cmd: []string{ - "mysqld", - "--server-id=1", - "--log-bin=mysql-bin", - "--binlog-format=ROW", - "--innodb-buffer-pool-size=64M", - "--performance-schema=OFF", + // mysql-debug is a MySQL image; disable mysqlx and trim caches to keep the one-off server small. + src, suffix := SetupMySQLTestContainerSource(s.t, "mydbginc", MySQLTestContainerConfig{ + Image: "ghcr.io/peerdb-io/mysql-debug:8.0.46", + Flavor: protos.MySqlFlavor_MYSQL_MYSQL, + ReplicationMechanism: mySource.Config.ReplicationMechanism, + ExtraServerFlags: []string{ "--mysqlx=0", - "--max-connections=20", "--table-open-cache=64", "--table-definition-cache=128", "--innodb-log-buffer-size=8M", }, - ExposedPorts: []string{"3306/tcp"}, - WaitingFor: wait.ForListeningPort("3306/tcp").WithStartupTimeout(3 * time.Minute), - } - - ctr, err := testcontainers.GenericContainer(s.t.Context(), testcontainers.GenericContainerRequest{ - ContainerRequest: req, - Started: true, }) - testcontainers.CleanupContainer(s.t, ctr, testcontainers.StopTimeout(30*time.Second)) - require.NoError(s.t, err) - - mapped, err := ctr.MappedPort(s.t.Context(), "3306/tcp") - require.NoError(s.t, err) - port, err := strconv.Atoi(mapped.Port()) - require.NoError(s.t, err) - - suffix := "mydbginc_" + strings.ToLower(common.RandomString(8)) - config := &protos.MySqlConfig{ - // host.docker.internal resolves both from the test process (to the published port) and - // from the flow worker container (via host-gateway), unlike the container's own host. - Host: internal.MySQLTestHost(), - Port: uint32(port), - User: "root", - Password: internal.MySQLTestRootPasswordWithFallback("cipass"), - DisableTls: true, - Flavor: protos.MySqlFlavor_MYSQL_MYSQL, - ReplicationMechanism: mySource.Config.ReplicationMechanism, - } - src, err := setupMyConnector(s.t, suffix, config, "mysql_debug_"+suffix) - require.NoError(s.t, err) - s.t.Cleanup(func() { src.Teardown(s.t, context.Background(), suffix) }) srcTableName := "incident" srcFullName := fmt.Sprintf("e2e_test_%s.%s", suffix, srcTableName) @@ -1724,6 +1682,81 @@ func (s ClickHouseSuite) Test_MySQL_BinlogIncident() { RequireEnvCanceled(s.t, env) } +// Test_MariaDB_PartialRowEvent verifies that a MariaDB PARTIAL_ROW_DATA_EVENT (MariaDB 12.3+) - an +// oversized rows event fragmented across multiple binlog events - surfaces as a resync-required +// error on the mirror. It runs its own throwaway MariaDB because it must lower +// binlog_row_event_fragment_threshold to its minimum to force fragmentation, and the event type +// only exists on MariaDB 12.3+. +func (s ClickHouseSuite) Test_MariaDB_PartialRowEvent() { + if s.cluster { + s.t.Skip("source-side partial row event coverage does not need to run against ClickHouse cluster") + } + mySource, ok := s.source.(*MySqlSource) + if !ok { + s.t.Skip("only applies to mysql") + } + // We spin up our own MariaDB 12.3 below regardless of the suite source, so only run this once - + // under the MySQL suite variant - to avoid launching a second throwaway container needlessly. + if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MARIA { + s.t.Skip("partial row event test spins up its own MariaDB; skip duplicate run under MariaDB suite") + } + + // binlog_row_event_fragment_threshold is pinned to its 1024-byte minimum so a modest row + // overflows a single rows event and MariaDB splits it into PARTIAL_ROW_DATA_EVENT fragments. + src, suffix := SetupMySQLTestContainerSource(s.t, "mdbpart", MySQLTestContainerConfig{ + Image: "mariadb:12.3", + Flavor: protos.MySqlFlavor_MYSQL_MARIA, + ReplicationMechanism: protos.MySqlReplicationMechanism_MYSQL_GTID, + ExtraServerFlags: []string{ + "--binlog-row-metadata=FULL", + "--binlog-row-event-fragment-threshold=1024", + }, + }) + + srcTableName := "partial_rows" + srcFullName := fmt.Sprintf("e2e_test_%s.%s", suffix, srcTableName) + dstTableName := "partial_rows_dst" + + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `CREATE TABLE %s (id INT PRIMARY KEY, payload LONGTEXT)`, srcFullName))) + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: "test_mariadb_partial_row_" + suffix, + TableMappings: []*protos.TableMapping{{ + SourceTableIdentifier: srcFullName, + DestinationTableIdentifier: dstTableName, + ShardingKey: "id", + }}, + Destination: s.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + // The suite source is the shared CI MySQL; point the mirror at our MariaDB source instead. + flowConnConfig.SourceName = src.GeneratePeer(s.t).Name + + tc := NewTemporalClient(s.t) + env := ExecutePeerflow(s.t, tc, flowConnConfig) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + + // Insert a row larger than the 1024-byte fragment threshold so the CDC rows event is fragmented + // into PARTIAL_ROW_DATA_EVENT, which we don't reassemble and must fail loudly on. + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (1, REPEAT('x', 8192))`, srcFullName))) + + catalogPool, err := internal.GetCatalogConnectionPoolFromEnv(s.t.Context()) + require.NoError(s.t, err) + EnvWaitFor(s.t, env, 3*time.Minute, "waiting for partial row event error", func() bool { + count, err := GetLogCount(s.t.Context(), catalogPool, flowConnConfig.FlowJobName, "error", "fragmented oversized row events") + if err != nil { + s.t.Log("Error querying flow_errors:", err) + return false + } + return count > 0 + }) + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + func (s ClickHouseSuite) Test_MySQL_Default_Partition_Key_Parallel_Snapshot() { if _, ok := s.source.(*MySqlSource); !ok { s.t.Skip("only applies to mysql") diff --git a/flow/e2e/mysql.go b/flow/e2e/mysql.go index b655c4134..746bab92d 100644 --- a/flow/e2e/mysql.go +++ b/flow/e2e/mysql.go @@ -3,7 +3,14 @@ package e2e import ( "context" "fmt" + "strconv" + "strings" "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" "github.com/PeerDB-io/peerdb/flow/connectors" connmysql "github.com/PeerDB-io/peerdb/flow/connectors/mysql" @@ -34,6 +41,90 @@ func SetupMariaDB(t *testing.T, suffix string) (*MySqlSource, error) { return setupMyConnector(t, suffix, internal.GetMariaDBConfigFromEnv(flavor, replication), "mariadb") } +// MySQLTestContainerConfig parameterizes a throwaway MySQL/MariaDB testcontainer source. +type MySQLTestContainerConfig struct { + // Image is the container image, e.g. "mariadb:12.3" or "ghcr.io/peerdb-io/mysql-debug:8.0.46". + Image string + // Flavor selects MySQL vs MariaDB; it drives both the root-password env var names and the + // connector flavor. + Flavor protos.MySqlFlavor + // ReplicationMechanism is the CDC mechanism the mirror should use against this source. + ReplicationMechanism protos.MySqlReplicationMechanism + // ExtraServerFlags are appended to a common small-footprint server flag base, e.g. + // "--binlog-row-event-fragment-threshold=1024" (MariaDB) or "--mysqlx=0" (MySQL). + ExtraServerFlags []string +} + +// SetupMySQLTestContainerSource starts a throwaway MySQL/MariaDB server in a testcontainer and +// returns a MySqlSource pointed at it, plus the generated suffix used for its e2e database. It +// registers container and database cleanup on t. Use it to replace the shared CI source with an +// isolated server a test can reconfigure (custom image, extra server flags) - typically by assigning +// the returned source's peer to flowConnConfig.SourceName. +func SetupMySQLTestContainerSource( + t *testing.T, namePrefix string, cfg MySQLTestContainerConfig, +) (*MySqlSource, string) { + t.Helper() + + rootPassword := internal.MySQLTestRootPasswordWithFallback("cipass") + env := map[string]string{} + if cfg.Flavor == protos.MySqlFlavor_MYSQL_MARIA { + env["MARIADB_ROOT_PASSWORD"] = rootPassword + env["MARIADB_ROOT_HOST"] = "%" + } else { + env["MYSQL_ROOT_PASSWORD"] = rootPassword + env["MYSQL_ROOT_HOST"] = "%" + } + + // Common small-footprint server flags valid on both MySQL 8 and MariaDB. Both official + // entrypoints prepend the server binary when the first arg starts with "-", so none is needed. + cmd := append([]string{ + "--server-id=1", + "--log-bin=mysql-bin", + "--binlog-format=ROW", + "--innodb-buffer-pool-size=64M", + "--performance-schema=OFF", + "--max-connections=20", + }, cfg.ExtraServerFlags...) + + req := testcontainers.ContainerRequest{ + Image: cfg.Image, + Env: env, + Cmd: cmd, + ExposedPorts: []string{"3306/tcp"}, + WaitingFor: wait.ForListeningPort("3306/tcp").WithStartupTimeout(3 * time.Minute), + } + + ctr, err := testcontainers.GenericContainer(t.Context(), testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + testcontainers.CleanupContainer(t, ctr, testcontainers.StopTimeout(30*time.Second)) + require.NoError(t, err) + + mapped, err := ctr.MappedPort(t.Context(), "3306/tcp") + require.NoError(t, err) + port, err := strconv.Atoi(mapped.Port()) + require.NoError(t, err) + + suffix := namePrefix + "_" + strings.ToLower(common.RandomString(8)) + config := &protos.MySqlConfig{ + // host.docker.internal resolves both from the test process (to the published port) and from + // the flow worker container (via host-gateway), unlike the container's own hostname. + Host: internal.MySQLTestHost(), + Port: uint32(port), + User: "root", + Password: rootPassword, + DisableTls: true, + Flavor: cfg.Flavor, + ReplicationMechanism: cfg.ReplicationMechanism, + } + src, err := setupMyConnector(t, suffix, config, suffix) + require.NoError(t, err) + t.Cleanup(func() { src.Teardown(t, context.Background(), suffix) }) + + return src, suffix +} + func setupMyConnector(t *testing.T, suffix string, config *protos.MySqlConfig, peerName string) (*MySqlSource, error) { t.Helper() connector, err := connmysql.NewMySqlConnector(t.Context(), config) From 5e96c0fa060c79e1e083c0a791f46b7cecd60f84 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 2 Jul 2026 14:15:16 +0200 Subject: [PATCH 04/10] cleanup comments --- flow/e2e/mysql.go | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/flow/e2e/mysql.go b/flow/e2e/mysql.go index 746bab92d..5ef174197 100644 --- a/flow/e2e/mysql.go +++ b/flow/e2e/mysql.go @@ -43,12 +43,8 @@ func SetupMariaDB(t *testing.T, suffix string) (*MySqlSource, error) { // MySQLTestContainerConfig parameterizes a throwaway MySQL/MariaDB testcontainer source. type MySQLTestContainerConfig struct { - // Image is the container image, e.g. "mariadb:12.3" or "ghcr.io/peerdb-io/mysql-debug:8.0.46". - Image string - // Flavor selects MySQL vs MariaDB; it drives both the root-password env var names and the - // connector flavor. - Flavor protos.MySqlFlavor - // ReplicationMechanism is the CDC mechanism the mirror should use against this source. + Image string + Flavor protos.MySqlFlavor ReplicationMechanism protos.MySqlReplicationMechanism // ExtraServerFlags are appended to a common small-footprint server flag base, e.g. // "--binlog-row-event-fragment-threshold=1024" (MariaDB) or "--mysqlx=0" (MySQL). From 7ef6ee34cb0be5bc77cab7096969ac6e0899268e Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 2 Jul 2026 14:31:23 +0200 Subject: [PATCH 05/10] fix lint errs --- flow/e2e/mysql.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/flow/e2e/mysql.go b/flow/e2e/mysql.go index 5ef174197..8a27ef877 100644 --- a/flow/e2e/mysql.go +++ b/flow/e2e/mysql.go @@ -43,12 +43,12 @@ func SetupMariaDB(t *testing.T, suffix string) (*MySqlSource, error) { // MySQLTestContainerConfig parameterizes a throwaway MySQL/MariaDB testcontainer source. type MySQLTestContainerConfig struct { - Image string - Flavor protos.MySqlFlavor - ReplicationMechanism protos.MySqlReplicationMechanism + Image string // ExtraServerFlags are appended to a common small-footprint server flag base, e.g. // "--binlog-row-event-fragment-threshold=1024" (MariaDB) or "--mysqlx=0" (MySQL). - ExtraServerFlags []string + ExtraServerFlags []string + Flavor protos.MySqlFlavor + ReplicationMechanism protos.MySqlReplicationMechanism } // SetupMySQLTestContainerSource starts a throwaway MySQL/MariaDB server in a testcontainer and @@ -99,7 +99,7 @@ func SetupMySQLTestContainerSource( mapped, err := ctr.MappedPort(t.Context(), "3306/tcp") require.NoError(t, err) - port, err := strconv.Atoi(mapped.Port()) + port, err := strconv.ParseUint(mapped.Port(), 10, 32) require.NoError(t, err) suffix := namePrefix + "_" + strings.ToLower(common.RandomString(8)) From 1d13df1ba9f326d21345294bc340cfb1f1ecfbfc Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Fri, 3 Jul 2026 14:25:36 +0200 Subject: [PATCH 06/10] classify "NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET" as distinct error class --- flow/alerting/classifier.go | 5 ++++- flow/alerting/classifier_test.go | 12 ++++++++++++ flow/connectors/mysql/cdc.go | 5 ----- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/flow/alerting/classifier.go b/flow/alerting/classifier.go index 3ceabc5a2..0b0dca9b5 100644 --- a/flow/alerting/classifier.go +++ b/flow/alerting/classifier.go @@ -145,6 +145,9 @@ var ( ErrorNotifyBinlogEventExceededMaxAllowedPacket = ErrorClass{ Class: "NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET", action: NotifyUser, } + ErrorNotifyBinlogPartialRowEventUnsupported = ErrorClass{ + Class: "NOTIFY_BINLOG_PARTIAL_ROW_EVENT_UNSUPPORTED", action: NotifyUser, + } ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{ Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser, } @@ -1155,7 +1158,7 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) { } if _, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok { - return ErrorNotifyBinlogInvalid, ErrorInfo{ + return ErrorNotifyBinlogPartialRowEventUnsupported, ErrorInfo{ Source: ErrorSourceMySQL, Code: "UNSUPPORTED_PARTIAL_ROW_EVENT", } diff --git a/flow/alerting/classifier_test.go b/flow/alerting/classifier_test.go index 0c1dcb7f0..aca895710 100644 --- a/flow/alerting/classifier_test.go +++ b/flow/alerting/classifier_test.go @@ -1243,3 +1243,15 @@ func TestMySQLBinlogIncidentErrorShouldBeNotifyBinlogInvalid(t *testing.T) { Code: "BINLOG_INCIDENT", }, errInfo) } + +func TestMySQLUnsupportedPartialRowEventShouldBeNotifyPartialRowEventUnsupported(t *testing.T) { + t.Parallel() + + err := exceptions.NewMySQLUnsupportedPartialRowEventError(172) + errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err)) + assert.Equal(t, ErrorNotifyBinlogPartialRowEventUnsupported, errorClass) + assert.Equal(t, ErrorInfo{ + Source: ErrorSourceMySQL, + Code: "UNSUPPORTED_PARTIAL_ROW_EVENT", + }, errInfo) +} diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index f1baf6bd8..9dad9319a 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -46,11 +46,6 @@ const ( binlogStalenessMultiplier = 3 ) -// mariadbPartialRowDataEvent is MariaDB 12.3+ PARTIAL_ROW_DATA_EVENT (Log_event_type = 172). -// It fragments an oversized rows event across several binlog events, which a consumer must buffer -// and reassemble before decoding. go-mysql lib has no constant or parser for it, so it -// surfaces as a GenericEvent; we don't reassemble fragments yet, so we fail loudly rather than -// silently dropping the row change. const mariadbPartialRowDataEvent replication.EventType = 172 const ( From d91b0b5e96858e4b3cba0c2d478b8e6c73397217 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Fri, 3 Jul 2026 14:51:32 +0200 Subject: [PATCH 07/10] event type label as integer --- flow/connectors/mysql/cdc.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 9dad9319a..8c8feb204 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -12,7 +12,6 @@ import ( "log/slog" "math/rand/v2" "slices" - "strconv" "sync/atomic" "time" @@ -585,7 +584,7 @@ func (c *MySqlConnector) PullRecords( default: c.logger.Warn("unknown generic event", slog.Any("type", event.Header.EventType)) otelManager.Metrics.UnsupportedBinlogEventCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet( - attribute.String(otel_metrics.BinlogEventTypeKey, strconv.Itoa(int(event.Header.EventType))), + attribute.Int(otel_metrics.BinlogEventTypeKey, int(event.Header.EventType)), ))) } case *replication.QueryEvent: From 110681c2130343ac9b74fe03e46c6cda7abdc063 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Fri, 3 Jul 2026 15:32:30 +0200 Subject: [PATCH 08/10] don't fail if partial row event belong to out of pipe table --- flow/alerting/classifier_test.go | 2 +- flow/connectors/mysql/cdc.go | 76 +++++++++++++++++++++++++++++-- flow/e2e/clickhouse_mysql_test.go | 34 ++++++++------ flow/shared/exceptions/mysql.go | 12 ++++- 4 files changed, 103 insertions(+), 21 deletions(-) diff --git a/flow/alerting/classifier_test.go b/flow/alerting/classifier_test.go index aca895710..fa57f9b2f 100644 --- a/flow/alerting/classifier_test.go +++ b/flow/alerting/classifier_test.go @@ -1247,7 +1247,7 @@ func TestMySQLBinlogIncidentErrorShouldBeNotifyBinlogInvalid(t *testing.T) { func TestMySQLUnsupportedPartialRowEventShouldBeNotifyPartialRowEventUnsupported(t *testing.T) { t.Parallel() - err := exceptions.NewMySQLUnsupportedPartialRowEventError(172) + err := exceptions.NewMySQLUnsupportedPartialRowEventError(172, "e2e_test", "partial_rows") errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err)) assert.Equal(t, ErrorNotifyBinlogPartialRowEventUnsupported, errorClass) assert.Equal(t, ErrorInfo{ diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 8c8feb204..cc200e9ee 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -12,6 +12,7 @@ import ( "log/slog" "math/rand/v2" "slices" + "strings" "sync/atomic" "time" @@ -47,6 +48,45 @@ const ( const mariadbPartialRowDataEvent replication.EventType = 172 +const ( + partialRowsHeaderLen = 4 + 4 + 1 + partialRowFlagOrigEventSize = 0x01 + binlogCommonHeaderLen = 19 + rowsEventTableIDSize = 6 +) + +// parsePartialRowEventTableID extracts the source table id from a MariaDB PARTIAL_ROW_DATA_EVENT +// Layout of data: +// +// [0:4] seq_no (uint32 LE, 1-based) +// [4:8] total_fragments (uint32 LE) +// [8] flags (bit 0 = FL_ORIG_EVENT_SIZE) +// [9:17] original size (present only when FL_ORIG_EVENT_SIZE is set) +// then content (first fragment: embedded rows event = 19-byte header + post-header) +// +// The embedded rows-event post-header begins with a 6-byte little-endian table id. +func parsePartialRowEventTableID(data []byte) (uint64, uint32, bool) { + if len(data) < partialRowsHeaderLen { + return 0, 0, false + } + seqNo := binary.LittleEndian.Uint32(data[0:4]) + totalFragments := binary.LittleEndian.Uint32(data[4:8]) + flags := data[8] + if seqNo != 1 { + // continuation fragment: carries only raw row data, no embedded header, so no table id + return 0, totalFragments, false + } + contentStart := partialRowsHeaderLen + if flags&partialRowFlagOrigEventSize != 0 { + contentStart += 8 + } + tableIDStart := contentStart + binlogCommonHeaderLen + if len(data) < tableIDStart+rowsEventTableIDSize { + return 0, totalFragments, false + } + return mysql.FixedLengthInt(data[tableIDStart : tableIDStart+rowsEventTableIDSize]), totalFragments, true +} + const ( queryStatusVarFlags2 = 0 queryStatusVarSQLMode = 1 @@ -542,6 +582,11 @@ func (c *MySqlConnector) PullRecords( lastEventAt := time.Now() var mysqlParser *parser.Parser + // table id -> "schema.table", maintained from TABLE_MAP_EVENTs + tableIdToName := make(map[uint64]string) + // When we ignore an out-of-pipe fragmented rows event, its continuation fragments carry no table + // id; skip this many remaining PARTIAL_ROW_DATA_EVENTs before resolving table ids again. + var partialRowSkipFragments uint32 processEvent := func(event *replication.BinlogEvent) error { switch ev := event.Event.(type) { case *replication.GTIDEvent: @@ -576,11 +621,32 @@ func (c *MySqlConnector) PullRecords( slog.Uint64("incident", uint64(incident)), slog.String("message", message)) return exceptions.NewMySQLBinlogIncidentError(incident, message) case mariadbPartialRowDataEvent: - // MariaDB 12.3+ fragments an oversized rows event across multiple events; we don't - // reassemble fragments yet, so the row change is unrecoverable and a resync is required. + if partialRowSkipFragments > 0 { + // continuation fragments of an out-of-pipe group we already decided to ignore + partialRowSkipFragments-- + break + } + tableID, totalFragments, ok := parsePartialRowEventTableID(ev.Data) + sourceTableName, known := tableIdToName[tableID] + if !ok || !known { + // couldn't recover/resolve the table, fail loudly + c.logger.Error("[mysql] received unresolvable MariaDB partial row data event, resync required", + slog.Uint64("eventType", uint64(mariadbPartialRowDataEvent)), slog.Bool("parsed", ok)) + return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent), "", "") + } + if req.TableNameSchemaMapping[req.TableNameMapping[sourceTableName].Name] == nil { + // table not in the pipe - ignore this fragment group and its continuation fragments + if totalFragments > 1 { + partialRowSkipFragments = totalFragments - 1 + } + c.logger.Warn("[mysql] ignoring MariaDB partial row data event for table outside the pipe", + slog.String("table", sourceTableName), slog.Uint64("totalFragments", uint64(totalFragments))) + break + } + schemaName, tableName, _ := strings.Cut(sourceTableName, ".") c.logger.Error("[mysql] received MariaDB partial row data event, resync required", - slog.Uint64("eventType", uint64(mariadbPartialRowDataEvent))) - return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent)) + slog.Uint64("eventType", uint64(mariadbPartialRowDataEvent)), slog.String("table", sourceTableName)) + return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent), schemaName, tableName) default: c.logger.Warn("unknown generic event", slog.Any("type", event.Header.EventType)) otelManager.Metrics.UnsupportedBinlogEventCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet( @@ -617,6 +683,8 @@ func (c *MySqlConnector) PullRecords( c.processRenameTableQuery(ctx, otelManager, req, s, string(ev.Schema)) } } + case *replication.TableMapEvent: + tableIdToName[ev.TableID] = string(ev.Schema) + "." + string(ev.Table) case *replication.RowsEvent: sourceTableName := string(ev.Table.Schema) + "." + string(ev.Table.Table) // TODO this is fragile destinationTableName := req.TableNameMapping[sourceTableName].Name diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index b2a7cf6e8..753c33a70 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -1682,11 +1682,8 @@ func (s ClickHouseSuite) Test_MySQL_BinlogIncident() { RequireEnvCanceled(s.t, env) } -// Test_MariaDB_PartialRowEvent verifies that a MariaDB PARTIAL_ROW_DATA_EVENT (MariaDB 12.3+) - an -// oversized rows event fragmented across multiple binlog events - surfaces as a resync-required -// error on the mirror. It runs its own throwaway MariaDB because it must lower -// binlog_row_event_fragment_threshold to its minimum to force fragmentation, and the event type -// only exists on MariaDB 12.3+. +// Test_MariaDB_PartialRowEvent verifies how the mirror handles a MariaDB PARTIAL_ROW_DATA_EVENT +// (MariaDB 12.3+) - an oversized rows event fragmented across multiple binlog events. func (s ClickHouseSuite) Test_MariaDB_PartialRowEvent() { if s.cluster { s.t.Skip("source-side partial row event coverage does not need to run against ClickHouse cluster") @@ -1695,10 +1692,8 @@ func (s ClickHouseSuite) Test_MariaDB_PartialRowEvent() { if !ok { s.t.Skip("only applies to mysql") } - // We spin up our own MariaDB 12.3 below regardless of the suite source, so only run this once - - // under the MySQL suite variant - to avoid launching a second throwaway container needlessly. - if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MARIA { - s.t.Skip("partial row event test spins up its own MariaDB; skip duplicate run under MariaDB suite") + if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MYSQL { + s.t.Skip("only applies to maria flavor") } // binlog_row_event_fragment_threshold is pinned to its 1024-byte minimum so a modest row @@ -1737,13 +1732,24 @@ func (s ClickHouseSuite) Test_MariaDB_PartialRowEvent() { env := ExecutePeerflow(s.t, tc, flowConnConfig) SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) - // Insert a row larger than the 1024-byte fragment threshold so the CDC rows event is fragmented - // into PARTIAL_ROW_DATA_EVENT, which we don't reassemble and must fail loudly on. - require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( - `INSERT INTO %s (id, payload) VALUES (1, REPEAT('x', 8192))`, srcFullName))) - catalogPool, err := internal.GetCatalogConnectionPoolFromEnv(s.t.Context()) require.NoError(s.t, err) + + // A fragmented rows event on a table OUTSIDE the pipe must NOT fail the mirror + ignoredFullName := fmt.Sprintf("e2e_test_%s.partial_rows_ignored", suffix) + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `CREATE TABLE %s (id INT PRIMARY KEY, payload LONGTEXT)`, ignoredFullName))) + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (1, REPEAT('y', 8192))`, ignoredFullName))) + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (1, 'small')`, srcFullName))) + EnvWaitForCount(env, s, "waiting for in-pipe row past out-of-pipe partial row event", + dstTableName, "id,payload", 1) + + // An oversized row larger than the 1024-byte fragment threshold on the mirrored table itself is + // fragmented into PARTIAL_ROW_DATA_EVENTs, which we don't reassemble and must fail loudly on. + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (2, REPEAT('x', 8192))`, srcFullName))) EnvWaitFor(s.t, env, 3*time.Minute, "waiting for partial row event error", func() bool { count, err := GetLogCount(s.t.Context(), catalogPool, flowConnConfig.FlowJobName, "error", "fragmented oversized row events") if err != nil { diff --git a/flow/shared/exceptions/mysql.go b/flow/shared/exceptions/mysql.go index e6a68c102..af3a38487 100644 --- a/flow/shared/exceptions/mysql.go +++ b/flow/shared/exceptions/mysql.go @@ -155,14 +155,22 @@ func (e *MySQLBinlogIncidentError) Error() string { } type MySQLUnsupportedPartialRowEventError struct { + Schema string + Table string EventType byte } -func NewMySQLUnsupportedPartialRowEventError(eventType byte) *MySQLUnsupportedPartialRowEventError { - return &MySQLUnsupportedPartialRowEventError{EventType: eventType} +func NewMySQLUnsupportedPartialRowEventError(eventType byte, schema string, table string) *MySQLUnsupportedPartialRowEventError { + return &MySQLUnsupportedPartialRowEventError{EventType: eventType, Schema: schema, Table: table} } func (e *MySQLUnsupportedPartialRowEventError) Error() string { + if e.Schema != "" || e.Table != "" { + return fmt.Sprintf( + "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d) on table %s.%s: "+ + "fragmented oversized row events cannot be processed; a resync is required", + e.EventType, e.Schema, e.Table) + } return fmt.Sprintf( "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d): fragmented oversized row events cannot be processed; a resync is required", e.EventType) From 48bf84deac73e4922e326b1270acaf3a30cc35c3 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Fri, 3 Jul 2026 16:02:53 +0200 Subject: [PATCH 09/10] fix parsePartialRowEventTableID --- flow/connectors/mysql/cdc.go | 11 +++++------ flow/connectors/mysql/cdc_test.go | 33 +++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 6 deletions(-) diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index cc200e9ee..08f101acb 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -56,12 +56,11 @@ const ( ) // parsePartialRowEventTableID extracts the source table id from a MariaDB PARTIAL_ROW_DATA_EVENT -// Layout of data: // -// [0:4] seq_no (uint32 LE, 1-based) -// [4:8] total_fragments (uint32 LE) +// [0:4] total_fragments (uint32 LE) +// [4:8] seq_no (uint32 LE, 1-based) // [8] flags (bit 0 = FL_ORIG_EVENT_SIZE) -// [9:17] original size (present only when FL_ORIG_EVENT_SIZE is set) +// [9:17] original size (present only when FL_ORIG_EVENT_SIZE is set, i.e. on the first fragment) // then content (first fragment: embedded rows event = 19-byte header + post-header) // // The embedded rows-event post-header begins with a 6-byte little-endian table id. @@ -69,8 +68,8 @@ func parsePartialRowEventTableID(data []byte) (uint64, uint32, bool) { if len(data) < partialRowsHeaderLen { return 0, 0, false } - seqNo := binary.LittleEndian.Uint32(data[0:4]) - totalFragments := binary.LittleEndian.Uint32(data[4:8]) + totalFragments := binary.LittleEndian.Uint32(data[0:4]) + seqNo := binary.LittleEndian.Uint32(data[4:8]) flags := data[8] if seqNo != 1 { // continuation fragment: carries only raw row data, no embedded header, so no table id diff --git a/flow/connectors/mysql/cdc_test.go b/flow/connectors/mysql/cdc_test.go index 7ee420420..a163302bf 100644 --- a/flow/connectors/mysql/cdc_test.go +++ b/flow/connectors/mysql/cdc_test.go @@ -2,6 +2,7 @@ package connmysql import ( "context" + "encoding/hex" "fmt" "log/slog" "slices" @@ -438,3 +439,35 @@ func TestIntegrationGetTableSchemaPrimaryKeyVariants(t *testing.T) { }) } } + +func TestParsePartialRowEventTableID(t *testing.T) { + t.Parallel() + + // Real PARTIAL_ROW_DATA_EVENT payloads captured from MariaDB 12.3 + decode := func(s string) []byte { + b, err := hex.DecodeString(s) + require.NoError(t, err) + return b + } + // first fragment: total=9, seq=1, flags=FL_ORIG_EVENT_SIZE, then origSize + embedded rows event + firstFragment := decode("09000000010000000126200000000000002cc0476a" + + "17010000002a20000000000000000012000000000001000203fc0100000000200000") + // a continuation fragment: total=9, seq=2, no flags, raw row data + continuationFragment := decode("09000000020000000079797979797979797979") + + tableID, total, ok := parsePartialRowEventTableID(firstFragment) + require.True(t, ok, "first fragment table id should be recoverable") + require.Equal(t, uint64(18), tableID) + require.Equal(t, uint32(9), total) + + _, total, ok = parsePartialRowEventTableID(continuationFragment) + require.False(t, ok, "continuation fragment carries no table id") + require.Equal(t, uint32(9), total, "total_fragments is still readable on continuation fragments") + + // too short for even the post-header + _, _, ok = parsePartialRowEventTableID([]byte{0x09, 0x00, 0x00}) + require.False(t, ok) + // post-header present but content truncated before the embedded table id + _, _, ok = parsePartialRowEventTableID(firstFragment[:partialRowsHeaderLen+8+binlogCommonHeaderLen+2]) + require.False(t, ok) +} From 730295cc3aa90862f2193dafc1ed56a275e8d23d Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Fri, 3 Jul 2026 16:23:02 +0200 Subject: [PATCH 10/10] common resource flags for mysql/maria test container --- flow/e2e/clickhouse_mysql_test.go | 7 ------- flow/e2e/mysql.go | 20 ++++++++++++++++---- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index 753c33a70..7f76d67a9 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -1622,17 +1622,10 @@ func (s ClickHouseSuite) Test_MySQL_BinlogIncident() { s.t.Skip("binlog incident injection requires a MySQL debug build; not available for MariaDB") } - // mysql-debug is a MySQL image; disable mysqlx and trim caches to keep the one-off server small. src, suffix := SetupMySQLTestContainerSource(s.t, "mydbginc", MySQLTestContainerConfig{ Image: "ghcr.io/peerdb-io/mysql-debug:8.0.46", Flavor: protos.MySqlFlavor_MYSQL_MYSQL, ReplicationMechanism: mySource.Config.ReplicationMechanism, - ExtraServerFlags: []string{ - "--mysqlx=0", - "--table-open-cache=64", - "--table-definition-cache=128", - "--innodb-log-buffer-size=8M", - }, }) srcTableName := "incident" diff --git a/flow/e2e/mysql.go b/flow/e2e/mysql.go index 8a27ef877..752296410 100644 --- a/flow/e2e/mysql.go +++ b/flow/e2e/mysql.go @@ -44,8 +44,8 @@ func SetupMariaDB(t *testing.T, suffix string) (*MySqlSource, error) { // MySQLTestContainerConfig parameterizes a throwaway MySQL/MariaDB testcontainer source. type MySQLTestContainerConfig struct { Image string - // ExtraServerFlags are appended to a common small-footprint server flag base, e.g. - // "--binlog-row-event-fragment-threshold=1024" (MariaDB) or "--mysqlx=0" (MySQL). + // ExtraServerFlags are appended to a common small-footprint server flag base (which already + // includes flavor-aware low-resource flags), e.g. "--binlog-row-event-fragment-threshold=1024". ExtraServerFlags []string Flavor protos.MySqlFlavor ReplicationMechanism protos.MySqlReplicationMechanism @@ -73,14 +73,26 @@ func SetupMySQLTestContainerSource( // Common small-footprint server flags valid on both MySQL 8 and MariaDB. Both official // entrypoints prepend the server binary when the first arg starts with "-", so none is needed. - cmd := append([]string{ + // These throwaway servers run alongside every other container on the machine, so keep their + // resource usage low. + cmd := []string{ "--server-id=1", "--log-bin=mysql-bin", "--binlog-format=ROW", "--innodb-buffer-pool-size=64M", + "--innodb-log-buffer-size=8M", "--performance-schema=OFF", "--max-connections=20", - }, cfg.ExtraServerFlags...) + "--table-open-cache=64", + "--table-definition-cache=128", + } + // Flavor-specific low-resource flags for options that only exist on one server. + if cfg.Flavor == protos.MySqlFlavor_MYSQL_MARIA { + cmd = append(cmd, "--aria-pagecache-buffer-size=8M") + } else { + cmd = append(cmd, "--mysqlx=0") + } + cmd = append(cmd, cfg.ExtraServerFlags...) req := testcontainers.ContainerRequest{ Image: cfg.Image,