maria: require resync on partial row events

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,
 	}
@@ -1154,6 +1157,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if _, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok {
+		return ErrorNotifyBinlogPartialRowEventUnsupported, 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/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, "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{
+		Source: ErrorSourceMySQL,
+		Code:   "UNSUPPORTED_PARTIAL_ROW_EVENT",
+	}, errInfo)
+}

flow/connectors/mysql/cdc.go

@@ -12,6 +12,7 @@ import (
 	"log/slog"
 	"math/rand/v2"
 	"slices"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -45,6 +46,46 @@ const (
 	binlogStalenessMultiplier    = 3
 )
 
+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
+//
+//	[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, 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.
+func parsePartialRowEventTableID(data []byte) (uint64, uint32, bool) {
+	if len(data) < partialRowsHeaderLen {
+		return 0, 0, false
+	}
+	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
+		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
@@ -537,9 +578,22 @@ func (c *MySqlConnector) PullRecords(
 				time.Now().UTC().Sub(time.UnixMicro(int64(commitTs))).Microseconds())
 		}
 	}
+	recordUnsupportedEvent := func(ctx context.Context, event *replication.BinlogEvent, prefix string) {
+		otelManager.Metrics.UnsupportedBinlogEventCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
+			attribute.String(otel_metrics.BinlogEventTypeKey, fmt.Sprintf("%s_%d", prefix, int(event.Header.EventType))),
+		)))
+		if _, loaded := c.warnedUnsupportedEventTypes.LoadOrStore(event.Header.EventType, struct{}{}); !loaded {
+			c.logger.Warn("unsupported rows event", slog.Any("type", event.Header.EventType))
+		}
+	}
 
 	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:
@@ -566,12 +620,42 @@ 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:
+				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)), slog.String("table", sourceTableName))
+				return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent), schemaName, tableName)
+			default:
+				recordUnsupportedEvent(ctx, event, "Generic")
 			}
 		case *replication.QueryEvent:
 			if !inTx && gset == nil && event.Header.LogPos > pos.Pos {
@@ -603,6 +687,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
@@ -787,6 +873,8 @@ func (c *MySqlConnector) PullRecords(
 					}
 				case replication.WRITE_ROWS_EVENTv0, replication.UPDATE_ROWS_EVENTv0, replication.DELETE_ROWS_EVENTv0:
 					return fmt.Errorf("mysql v0 replication protocol not supported")
+				default:
+					recordUnsupportedEvent(ctx, event, "Rows")
 				}
 			}
 			if event.Header.Timestamp > 0 {
@@ -795,6 +883,8 @@ func (c *MySqlConnector) PullRecords(
 					int64(event.Header.Timestamp),
 				)
 			}
+		default:
+			recordUnsupportedEvent(ctx, event, "Untyped")
 		}
 
 		return nil

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

flow/connectors/mysql/mysql.go

@@ -31,18 +31,19 @@ import (
 
 type MySqlConnector struct {
 	*metadataStore.PostgresMetadata
-	config                *protos.MySqlConfig
-	ssh                   *utils.SSHTunnel
-	conn                  atomic.Pointer[client.Conn] // atomic used for internal concurrency, connector interface is not threadsafe
-	contexts              atomic.Pointer[chan context.Context]
-	logger                log.Logger
-	rdsAuth               *utils.RDSAuth
-	serverVersion         string
-	collationCharset      atomic.Pointer[map[uint64]string]
-	warnedCharsets        sync.Map
-	binlogHeartbeatPeriod time.Duration
-	totalBytesRead        atomic.Int64
-	deltaBytesRead        atomic.Int64
+	config                      *protos.MySqlConfig
+	ssh                         *utils.SSHTunnel
+	conn                        atomic.Pointer[client.Conn] // atomic used for internal concurrency, connector interface is not threadsafe
+	contexts                    atomic.Pointer[chan context.Context]
+	logger                      log.Logger
+	rdsAuth                     *utils.RDSAuth
+	serverVersion               string
+	collationCharset            atomic.Pointer[map[uint64]string]
+	warnedUnsupportedEventTypes sync.Map
+	warnedCharsets              sync.Map
+	binlogHeartbeatPeriod       time.Duration
+	totalBytesRead              atomic.Int64
+	deltaBytesRead              atomic.Int64
 }
 
 func NewMySqlConnector(ctx context.Context, config *protos.MySqlConfig) (*MySqlConnector, error) {