fix(mysql): fail fast on partial JSON binlog events (PARTIAL_JSON)

flow/alerting/classifier.go

@@ -145,6 +145,9 @@ var (
 	ErrorNotifyBinlogEventExceededMaxAllowedPacket = ErrorClass{
 		Class: "NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET", action: NotifyUser,
 	}
+	ErrorNotifyBinlogPartialJsonUnsupported = ErrorClass{
+		Class: "NOTIFY_BINLOG_PARTIAL_JSON_UNSUPPORTED", action: NotifyUser,
+	}
 	ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{
 		Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser,
 	}
@@ -1119,6 +1122,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if _, ok := errors.AsType[*exceptions.MySQLUnsupportedBinlogRowValueOptionsError](err); ok {
+		return ErrorNotifyBinlogPartialJsonUnsupported, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "UNSUPPORTED_BINLOG_ROW_VALUE_OPTIONS",
+		}
+	}
+
 	if unsupportedDDLError, ok := errors.AsType[*exceptions.MySQLUnsupportedDDLError](err); ok {
 		return ErrorNotifyBinlogRowMetadataInvalid, ErrorInfo{
 			Source: ErrorSourceMySQL,

flow/alerting/classifier_test.go

@@ -1243,3 +1243,15 @@ func TestMySQLBinlogIncidentErrorShouldBeNotifyBinlogInvalid(t *testing.T) {
 		Code:   "BINLOG_INCIDENT",
 	}, errInfo)
 }
+
+func TestMySQLUnsupportedBinlogRowValueOptionsErrorShouldBeNotifyBinlogPartialJsonUnsupported(t *testing.T) {
+	t.Parallel()
+
+	err := exceptions.NewMySQLUnsupportedBinlogRowValueOptionsError("mydb", "mytable")
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err))
+	assert.Equal(t, ErrorNotifyBinlogPartialJsonUnsupported, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "UNSUPPORTED_BINLOG_ROW_VALUE_OPTIONS",
+	}, errInfo)
+}

flow/connectors/mysql/cdc.go

@@ -813,6 +813,11 @@ 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")
+				case replication.PARTIAL_UPDATE_ROWS_EVENT:
+					// Emitted only when binlog_row_value_options=PARTIAL_JSON is enabled at runtime.
+					e := exceptions.NewMySQLUnsupportedBinlogRowValueOptionsError(string(ev.Table.Schema), string(ev.Table.Table))
+					c.logger.Error(e.Error())
+					return e
 				}
 			}
 			if event.Header.Timestamp > 0 {

flow/connectors/mysql/qvalue_convert.go

@@ -2,6 +2,7 @@ package connmysql
 
 import (
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"log/slog"
 	"math"
@@ -547,8 +548,9 @@ func QValueFromMysqlRowEvent(
 	case time.Time:
 		return types.QValueTimestamp{Val: val}, nil
 	case *replication.JsonDiff:
-		// TODO support somehow??
-		return types.QValueNull(types.QValueKindJSON), nil
+		// Partial JSON updates (binlog_row_value_options=PARTIAL_JSON) cannot be applied; the caller
+		// fails fast on PARTIAL_UPDATE_ROWS_EVENT, so this is a defensive backstop against silent data loss.
+		return nil, errors.New("partial JSON update value is not supported; binlog_row_value_options must be disabled")
 	case []byte:
 		switch qkind {
 		case types.QValueKindBytes:

flow/shared/exceptions/mysql.go

@@ -55,6 +55,21 @@ func (e *MySQLUnsupportedBinlogRowMetadataError) Error() string {
 		e.SchemaName, e.TableName)
 }
 
+type MySQLUnsupportedBinlogRowValueOptionsError struct {
+	SchemaName string
+	TableName  string
+}
+
+func NewMySQLUnsupportedBinlogRowValueOptionsError(schema string, table string) *MySQLUnsupportedBinlogRowValueOptionsError {
+	return &MySQLUnsupportedBinlogRowValueOptionsError{SchemaName: schema, TableName: table}
+}
+
+func (e *MySQLUnsupportedBinlogRowValueOptionsError) Error() string {
+	return fmt.Sprintf(
+		"Received a partial JSON update event while processing %s.%s; binlog_row_value_options must be disabled (set to '')",
+		e.SchemaName, e.TableName)
+}
+
 type MySQLUnsupportedDDLError struct {
 	TableName string
 }