fix(mysql): handle transaction payload event (#4433)

- DBI-809 - handle transaction payload event (MySQL can send it if
binlog_transaction_compression=ON)
- handle missing GTID events (MariadbGTIDEvent, GtidTaggedLogEvent)

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -482,72 +482,25 @@ func (c *MySqlConnector) PullRecords(
 		return nil
 	}
 
-	lastEventAt := time.Now()
-	var mysqlParser *parser.Parser
-	for inTx || (!overtime && recordCount < req.MaxBatchSize) {
-		var event *replication.BinlogEvent
-		// don't gamble on closed timeoutCtx.Done() being prioritized over event backlog channel
-		err := timeoutCtx.Err()
-		if err == nil {
-			event, err = mystream.GetEvent(timeoutCtx)
-		}
-		if err != nil {
-			if ctxErr := ctx.Err(); ctxErr != nil {
-				c.logger.Info("[mysql] PullRecords context canceled, stopping streaming", slog.Any("error", err))
-				return ctxErr
-			}
-
-			if errors.Is(err, context.DeadlineExceeded) {
-				if recordCount == 0 || inTx {
-					if since := time.Since(lastEventAt); since > c.binlogStalenessThreshold() {
-						return exceptions.NewMySQLStaleConnectionError(since, c.binlogHeartbeatPeriod)
-					}
-
-					if recordCount == 0 {
-						// progress offset while no records read to avoid falling behind when all tables inactive
-						if updatedOffset != "" {
-							c.logger.Info("[mysql] updating inactive offset", slog.Any("offset", updatedOffset))
-							if err := c.SetLastOffset(ctx, req.FlowJobName, model.CdcCheckpoint{Text: updatedOffset}); err != nil {
-								c.logger.Error("[mysql] failed to update offset, ignoring", slog.Any("error", err))
-							} else {
-								updatedOffset = ""
-							}
-						}
-						resetTimeout(c.binlogStalenessThreshold())
-					} else {
-						c.logger.Info("[mysql] timeout reached, but still in transaction, waiting for inTx false",
-							slog.Uint64("records", uint64(recordCount)),
-							slog.Int64("bytes", totalFetchedBytes.Load()),
-							slog.Int("channelLen", req.RecordStream.ChannelLen()),
-							slog.Float64("elapsedMinutes", time.Since(pullStart).Minutes()))
-						resetTimeout(time.Minute)
-						overtime = true
-					}
-
-					continue
-				}
-
-				return nil
-			}
-
-			c.logger.Error("[mysql] PullRecords failed to get event", slog.Any("error", err))
-			return exceptions.NewMySQLExecuteError(err)
+	recordCommitLagMetric := func(ctx context.Context, commitTs uint64) {
+		if commitTs > 0 {
+			otelManager.Metrics.CommitLagGauge.Record(ctx,
+				time.Now().UTC().Sub(time.UnixMicro(int64(commitTs))).Microseconds())
 		}
+	}
 
-		lastEventAt = time.Now()
-
-		allFetchedBytes.Add(int64(len(event.RawData)))
-
+	lastEventAt := time.Now()
+	var mysqlParser *parser.Parser
+	processEvent := func(event *replication.BinlogEvent) error {
 		switch ev := event.Event.(type) {
 		case *replication.GTIDEvent:
-			if ev.ImmediateCommitTimestamp > 0 {
-				otelManager.Metrics.CommitLagGauge.Record(ctx,
-					time.Now().UTC().Sub(time.UnixMicro(int64(ev.ImmediateCommitTimestamp))).Microseconds())
-			}
+			recordCommitLagMetric(ctx, ev.ImmediateCommitTimestamp)
+		case *replication.GtidTaggedLogEvent: // MySQL 8.4+ tagged GTIDs
+			recordCommitLagMetric(ctx, ev.ImmediateCommitTimestamp)
 		case *replication.XIDEvent:
 			if gset != nil {
 				gset = ev.GSet
-				updatedOffset = gset.String()
+				updatedOffset = gset.String() // accumulated from GTID events above
 				req.RecordStream.UpdateLatestCheckpointText(updatedOffset)
 			} else if event.Header.LogPos > pos.Pos {
 				pos.Pos = event.Header.LogPos
@@ -825,6 +778,78 @@ func (c *MySqlConnector) PullRecords(
 				)
 			}
 		}
+
+		return nil
+	}
+
+	for inTx || (!overtime && recordCount < req.MaxBatchSize) {
+		var event *replication.BinlogEvent
+		// don't gamble on closed timeoutCtx.Done() being prioritized over event backlog channel
+		err := timeoutCtx.Err()
+		if err == nil {
+			event, err = mystream.GetEvent(timeoutCtx)
+		}
+		if err != nil {
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				c.logger.Info("[mysql] PullRecords context canceled, stopping streaming", slog.Any("error", err))
+				return ctxErr
+			}
+
+			if errors.Is(err, context.DeadlineExceeded) {
+				if recordCount == 0 || inTx {
+					if since := time.Since(lastEventAt); since > c.binlogStalenessThreshold() {
+						return exceptions.NewMySQLStaleConnectionError(since, c.binlogHeartbeatPeriod)
+					}
+
+					if recordCount == 0 {
+						// progress offset while no records read to avoid falling behind when all tables inactive
+						if updatedOffset != "" {
+							c.logger.Info("[mysql] updating inactive offset", slog.Any("offset", updatedOffset))
+							if err := c.SetLastOffset(ctx, req.FlowJobName, model.CdcCheckpoint{Text: updatedOffset}); err != nil {
+								c.logger.Error("[mysql] failed to update offset, ignoring", slog.Any("error", err))
+							} else {
+								updatedOffset = ""
+							}
+						}
+						resetTimeout(c.binlogStalenessThreshold())
+					} else {
+						c.logger.Info("[mysql] timeout reached, but still in transaction, waiting for inTx false",
+							slog.Uint64("records", uint64(recordCount)),
+							slog.Int64("bytes", totalFetchedBytes.Load()),
+							slog.Int("channelLen", req.RecordStream.ChannelLen()),
+							slog.Float64("elapsedMinutes", time.Since(pullStart).Minutes()))
+						resetTimeout(time.Minute)
+						overtime = true
+					}
+
+					continue
+				}
+
+				return nil
+			}
+
+			c.logger.Error("[mysql] PullRecords failed to get event", slog.Any("error", err))
+			return exceptions.NewMySQLExecuteError(err)
+		}
+
+		lastEventAt = time.Now()
+
+		allFetchedBytes.Add(int64(len(event.RawData)))
+
+		switch ev := event.Event.(type) {
+		case *replication.TransactionPayloadEvent:
+			for _, inner := range ev.Events {
+				err = processEvent(inner)
+				if err != nil {
+					return err
+				}
+			}
+		default:
+			err = processEvent(event)
+			if err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }

flow/e2e/clickhouse_mysql_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"github.com/testcontainers/testcontainers-go"
@@ -235,6 +236,117 @@ func (s ClickHouseSuite) Test_MySQL_Blobs() {
 	RequireEnvCanceled(s.t, env)
 }
 
+func (s ClickHouseSuite) Test_MySQL_TransactionPayloadCompression() {
+	mySource, ok := s.source.(*MySqlSource)
+	if !ok {
+		s.t.Skip("only applies to mysql")
+	}
+	if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
+		s.t.Skip("binlog_transaction_compression is not supported by MariaDB")
+	}
+	cmp, err := mySource.CompareServerVersion(s.t.Context(), mysql_validation.MySQLMinVersionForBinlogTransactionCompression)
+	require.NoError(s.t, err)
+	if cmp < 0 {
+		s.t.Skip("only applies to mysql versions with binlog_transaction_compression")
+	}
+
+	srcTableName := "test_txn_payload"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_txn_payload_dst"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id BIGINT PRIMARY KEY,
+			val TEXT NOT NULL
+		)
+	`, srcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      s.attachSuffix(srcTableName),
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	// Compress this session's transactions so the upcoming DML lands in the binlog as a single
+	// TRANSACTION_PAYLOAD_EVENT, exercising the compressed-payload decode path in CDC.
+	require.NoError(s.t, s.source.Exec(s.t.Context(), "SET SESSION binlog_transaction_compression = ON"))
+
+	// Capture the binlog position, then write a large, highly compressible transaction as a single
+	// multi-row INSERT so MySQL compresses it (compressed < uncompressed) into one payload event.
+	posBefore, err := mySource.GetMasterPos(s.t.Context())
+	require.NoError(s.t, err)
+
+	const rowCount = 200
+	rowVal := strings.Repeat("peerdb-compressible-", 100) // ~2KB highly repetitive per row
+	var insert strings.Builder
+	fmt.Fprintf(&insert, "INSERT INTO %s (id, val) VALUES ", srcFullName)
+	for i := 1; i <= rowCount; i++ {
+		if i > 1 {
+			insert.WriteByte(',')
+		}
+		fmt.Fprintf(&insert, "(%d, '%s')", i, rowVal)
+	}
+	require.NoError(s.t, s.source.Exec(s.t.Context(), insert.String()))
+
+	findPayloadEndPos := func(ctx context.Context, from mysql.Position) (bool, uint32, error) {
+		rs, err := mySource.Execute(ctx, fmt.Sprintf("SHOW BINLOG EVENTS IN '%s' FROM %d", from.Name, from.Pos))
+		if err != nil {
+			return false, 0, err
+		}
+		for i := range rs.Values {
+			// SHOW BINLOG EVENTS columns: Log_name, Pos, Event_type, Server_id, End_log_pos, Info
+			eventType, _ := rs.GetString(i, 2)
+			if strings.Contains(strings.ToLower(eventType), "payload") {
+				pos, _ := rs.GetInt(i, 4)
+				return true, uint32(pos), nil
+			}
+		}
+		return false, 0, nil
+	}
+	hasPayload, payloadEndPos, err := findPayloadEndPos(s.t.Context(), posBefore)
+	require.NoError(s.t, err)
+	require.True(s.t, hasPayload, "expected a TRANSACTION_PAYLOAD_EVENT in the binlog")
+
+	// for GTID "SHOW BINLOG EVENTS" returns really complex format and it's different
+	// across versions/flavors.
+	// Therefore, we just get the latest GTID from a server instead of reading it from a binglog.
+	gtidAfter, err := mySource.GetMasterGTIDSet(s.t.Context())
+	require.NoError(s.t, err)
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc", srcTableName, dstTableName, "id,val")
+
+	// Assert the payload transaction was checkpointed
+	pool, err := catalogTestAccessPool()
+	require.NoError(s.t, err)
+	var lastText string
+	require.NoError(s.t, pool.QueryRow(s.t.Context(),
+		"SELECT last_text FROM metadata_last_sync_state WHERE job_name = $1",
+		flowConnConfig.FlowJobName,
+	).Scan(&lastText))
+
+	if rest, isFilePos := strings.CutPrefix(lastText, "!f:"); isFilePos {
+		comma := strings.LastIndexByte(rest, ',')
+		require.NotEqual(s.t, -1, comma, "malformed file/pos offset %q", lastText)
+		storedPos, parseErr := strconv.ParseUint(rest[comma+1:], 16, 32)
+		require.NoError(s.t, parseErr)
+		require.GreaterOrEqual(s.t, uint32(storedPos), payloadEndPos,
+			"checkpoint did not advance past the TRANSACTION_PAYLOAD_EVENT")
+	} else {
+		storedSet, parseErr := mysql.ParseGTIDSet(mySource.Flavor(), lastText)
+		require.NoError(s.t, parseErr)
+		require.True(s.t, storedSet.Contain(gtidAfter),
+			"checkpoint GTID set %s does not cover the committed payload transaction %s", storedSet, gtidAfter)
+	}
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 // Test_MySQL_Binary_Trailing_Zeros reproduces trailing-0x00 truncation of fixed-length
 // BINARY(N) columns over CDC.
 //

flow/go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/cockroachdb/pebble/v2 v2.1.5
 	github.com/elastic/go-elasticsearch/v8 v8.19.6
 	github.com/go-logr/logr v1.4.3
-	github.com/go-mysql-org/go-mysql v1.15.1-0.20260615132323-06b4fb6b476d
+	github.com/go-mysql-org/go-mysql v1.15.1-0.20260622021642-60bfbfea8dd0
 	github.com/gogo/googleapis v1.4.1
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3

flow/go.sum

@@ -315,8 +315,8 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
 github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
-github.com/go-mysql-org/go-mysql v1.15.1-0.20260615132323-06b4fb6b476d h1:Kl+pOPwscRCgRjhN76nQkJEiXk49q10BxwHCdSrzyOE=
-github.com/go-mysql-org/go-mysql v1.15.1-0.20260615132323-06b4fb6b476d/go.mod h1:VjBTZTTDKL8OMXUAhNbg3VHaVVq9HOXJEBLpAKBFIfE=
+github.com/go-mysql-org/go-mysql v1.15.1-0.20260622021642-60bfbfea8dd0 h1:WaAkvbSCNg814bfKOUPvWP70ZXzeDLolfoMjf9BB0lE=
+github.com/go-mysql-org/go-mysql v1.15.1-0.20260622021642-60bfbfea8dd0/go.mod h1:VjBTZTTDKL8OMXUAhNbg3VHaVVq9HOXJEBLpAKBFIfE=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
 github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=

flow/pkg/mysql/validation.go

@@ -13,8 +13,9 @@ import (
 )
 
 const (
-	MySQLMinVersionForBinlogRowMetadata   = "8.0.1"
-	MariaDBMinVersionForBinlogRowMetadata = "10.5.0"
+	MySQLMinVersionForBinlogRowMetadata            = "8.0.1"
+	MariaDBMinVersionForBinlogRowMetadata          = "10.5.0"
+	MySQLMinVersionForBinlogTransactionCompression = "8.0.20"
 )
 
 func CompareServerVersion(conn *client.Conn, version string) (int, error) {