maria: require resync on partial row events

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{

flow/connectors/mysql/cdc.go

@@ -12,6 +12,7 @@ import (
 	"log/slog"
 	"math/rand/v2"
 	"slices"
+	"strconv"
 	"sync/atomic"
 	"time"
 
@@ -45,6 +46,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
@@ -566,12 +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)
+			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 {

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")

flow/e2e/mysql.go

@@ -3,7 +3,14 @@
 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,86 @@
 	return setupMyConnector(t, suffix, internal.GetMariaDBConfigFromEnv(flavor, replication), "mariadb")
 }
 
+// MySQLTestContainerConfig parameterizes a throwaway MySQL/MariaDB testcontainer source.
+type MySQLTestContainerConfig struct {
+	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).
+	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)

flow/otel_metrics/attributes.go

@@ -50,6 +50,7 @@ const (
 	TypeChangeToKey            = "to"
 	SourceEventTypeKey         = "sourceEventType"
 	OnlineSchemaMigrationTool  = "tool"
+	BinlogEventTypeKey         = "eventType"
 )
 
 const (

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 {

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)
+}