detect when binlog_row_metadata becomes minimal (#3698)

And add a unit test to make sure exceptions are always using pointer receiver for consistency

Author: jgao54

flow/alerting/classifier.go

@@ -115,6 +115,9 @@ var (
 	ErrorNotifyBinlogInvalid = ErrorClass{
 		Class: "NOTIFY_BINLOG_INVALID", action: NotifyUser,
 	}
+	ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{
+		Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser,
+	}
 	ErrorNotifyBadGTIDSetup = ErrorClass{
 		Class: "NOTIFY_BAD_MULTISOURCE_GTID_SETUP", action: NotifyUser,
 	}
@@ -831,6 +834,14 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	var unsupportedBinlogRowMetadataError *exceptions.MySQLUnsupportedBinlogRowMetadataError
+	if errors.As(err, &unsupportedBinlogRowMetadataError) {
+		return ErrorNotifyBinlogRowMetadataInvalid, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "UNSUPPORTED_BINLOG_ROW_METADATA",
+		}
+	}
+
 	var postgresPrimaryKeyModifiedError *exceptions.PrimaryKeyModifiedError
 	if errors.As(err, &postgresPrimaryKeyModifiedError) {
 		return ErrorUnsupportedSchemaChange, ErrorInfo{

flow/connectors/mysql/cdc.go

@@ -28,6 +28,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
+	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
@@ -488,6 +489,11 @@ func (c *MySqlConnector) PullRecords(
 				}
 			}
 		case *replication.RowsEvent:
+			if len(ev.Table.ColumnName) == 0 && len(ev.Table.ColumnType) > 0 {
+				e := exceptions.NewMySQLUnsupportedBinlogRowMetadataError(string(ev.Table.Schema), string(ev.Table.Table))
+				c.logger.Error(e.Error())
+				return e
+			}
 			sourceTableName := string(ev.Table.Schema) + "." + string(ev.Table.Table) // TODO this is fragile
 			destinationTableName := req.TableNameMapping[sourceTableName].Name
 			exclusion := req.TableNameMapping[sourceTableName].Exclude

flow/shared/exceptions/constructor_test.go

@@ -49,3 +49,53 @@ func TestErrorConstructorsShouldReturnPointers(t *testing.T) {
 		})
 	}
 }
+
+// For consistency/convention, tests that all error types implement Error() method with pointer receiver
+func TestErrorMethodsShouldHavePointerReceivers(t *testing.T) {
+	cfg := &packages.Config{
+		Mode: packages.NeedFiles | packages.NeedSyntax | packages.NeedTypes | packages.NeedTypesInfo,
+		Dir:  ".",
+	}
+
+	pkgs, err := packages.Load(cfg, ".")
+	require.NoError(t, err)
+	require.Len(t, pkgs, 1, "Expected exactly one package")
+
+	pkg := pkgs[0]
+
+	// Track error type names (structs that end with "Error")
+	errorTypes := make(map[string]bool)
+	for _, file := range pkg.Syntax {
+		ast.Inspect(file, func(n ast.Node) bool {
+			// Find all struct types that end with "Error"
+			if typeSpec, ok := n.(*ast.TypeSpec); ok {
+				if _, isStruct := typeSpec.Type.(*ast.StructType); isStruct {
+					if strings.HasSuffix(typeSpec.Name.Name, "Error") {
+						errorTypes[typeSpec.Name.Name] = true
+					}
+				}
+			}
+			return true
+		})
+	}
+
+	for _, file := range pkg.Syntax {
+		ast.Inspect(file, func(n ast.Node) bool {
+			if fn, ok := n.(*ast.FuncDecl); ok {
+				// Look for receiver methods
+				if fn.Recv != nil && len(fn.Recv.List) > 0 {
+					recvType := fn.Recv.List[0].Type
+					// Fail if found value receivers in error types
+					if recv, ok := recvType.(*ast.Ident); ok && errorTypes[recv.Name] {
+						pos := pkg.Fset.Position(fn.Pos())
+						assert.Fail(
+							t, "Error() method should have pointer receiver",
+							"%s: func (%s) Error() should be func (*%s) Error()",
+							pos.Filename, recv.Name, recv.Name)
+					}
+				}
+			}
+			return true
+		})
+	}
+}

flow/shared/exceptions/mysql.go

@@ -24,7 +24,21 @@ func NewMySQLIncompatibleColumnTypeError(
 	}
 }
 
-func (e MySQLIncompatibleColumnTypeError) Error() string {
+func (e *MySQLIncompatibleColumnTypeError) Error() string {
 	return fmt.Sprintf("Incompatible type for column %s in table %s, expect qkind %s but data is %s (mysql type %d)",
 		e.ColumnName, e.TableName, e.qkind, e.dataType, e.columnType)
 }
+
+type MySQLUnsupportedBinlogRowMetadataError struct {
+	SchemaName string
+	TableName  string
+}
+
+func NewMySQLUnsupportedBinlogRowMetadataError(schema string, table string) *MySQLUnsupportedBinlogRowMetadataError {
+	return &MySQLUnsupportedBinlogRowMetadataError{SchemaName: schema, TableName: table}
+}
+
+func (e *MySQLUnsupportedBinlogRowMetadataError) Error() string {
+	return fmt.Sprintf("Detected binlog_row_metadata change from FULL to MINIMAL while processing %s.%s",
+		e.SchemaName, e.TableName)
+}