Classify MySQL LinearRing WKB parse error as unsupported datatype

flow/alerting/classifier.go

@@ -42,6 +42,11 @@ const (
 	MongoShutdownInProgress              = "(ShutdownInProgress) The server is in quiesce mode and will shut down"
 	MongoInterruptedDueToReplStateChange = "(InterruptedDueToReplStateChange) operation was interrupted"
 	MongoIncompleteReadOfMessageHeader   = "incomplete read of message header"
+
+	// mysqlGeometryLinearRingNotClosedError is the specific WKB parse failure raised by the
+	// go-geos library when a LinearRing's points do not close. Used to give a more specific code
+	// once we already know the error came from MySQL geometry parsing.
+	mysqlGeometryLinearRingNotClosedError = "Points of LinearRing do not form a closed linestring"
 )
 
 var (
@@ -1101,6 +1106,15 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	var mysqlGeometryParseError *exceptions.MySQLGeometryParseError
+	if errors.As(err, &mysqlGeometryParseError) &&
+		strings.Contains(mysqlGeometryParseError.Error(), mysqlGeometryLinearRingNotClosedError) {
+		return ErrorUnsupportedDatatype, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "UNSUPPORTED_GEOMETRY_LINEAR_RING_NOT_CLOSED",
+		}
+	}
+
 	var postgresPrimaryKeyModifiedError *exceptions.PrimaryKeyModifiedError
 	if errors.As(err, &postgresPrimaryKeyModifiedError) {
 		return ErrorUnsupportedSchemaChange, ErrorInfo{

flow/alerting/classifier_test.go

@@ -993,3 +993,25 @@ func TestClickHouseStdExceptionObjectStorageIOErrorShouldBeRecoverable(t *testin
 		Code:   strconv.Itoa(int(chproto.ErrStdException)),
 	}, errInfo)
 }
+
+func TestMySQLGeometryLinearRingNotClosedShouldBeUnsupportedDatatype(t *testing.T) {
+	geosErr := fmt.Errorf("IllegalArgumentException: Points of LinearRing do not form a closed linestring")
+	wrapped := fmt.Errorf("mysql error: %w", exceptions.NewMySQLGeometryParseError(geosErr))
+	errorClass, errInfo := GetErrorClass(t.Context(), wrapped)
+	assert.Equal(t, ErrorUnsupportedDatatype, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "UNSUPPORTED_GEOMETRY_LINEAR_RING_NOT_CLOSED",
+	}, errInfo)
+}
+
+func TestMySQLGeometryParseErrorUnknownShouldFallThrough(t *testing.T) {
+	geosErr := fmt.Errorf("ParseException: Unknown WKB type 99")
+	wrapped := fmt.Errorf("mysql error: %w", exceptions.NewMySQLGeometryParseError(geosErr))
+	errorClass, errInfo := GetErrorClass(t.Context(), wrapped)
+	assert.Equal(t, ErrorOther, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceOther,
+		Code:   "UNKNOWN",
+	}, errInfo)
+}

flow/connectors/mysql/qvalue_convert.go

@@ -137,7 +137,7 @@ func processGeometryData(data []byte) (types.QValueGeometry, error) {
 	}
 	g, err := geom.NewGeomFromWKB(data[4:])
 	if err != nil {
-		return types.QValueGeometry{}, fmt.Errorf("failed to parse geometry WKB: %w", err)
+		return types.QValueGeometry{}, exceptions.NewMySQLGeometryParseError(err)
 	}
 	return types.QValueGeometry{Val: g.ToWKT()}, nil
 }

flow/shared/exceptions/mysql.go

@@ -67,6 +67,25 @@ func (e *MySQLUnsupportedDDLError) Error() string {
 		"Detected position-shifting DDL on table %s but binlog_row_metadata is not supported by this MySQL version.", e.TableName)
 }
 
+// MySQLGeometryParseError wraps go-geos WKB parse failures so they can be
+// classified as MySQL-source errors without string-matching at the alerting layer.
+// The underlying message comes from go-geos C code and is not unique to MySQL on its own.
+type MySQLGeometryParseError struct {
+	error
+}
+
+func NewMySQLGeometryParseError(err error) *MySQLGeometryParseError {
+	return &MySQLGeometryParseError{err}
+}
+
+func (e *MySQLGeometryParseError) Error() string {
+	return "failed to parse MySQL geometry WKB: " + e.error.Error()
+}
+
+func (e *MySQLGeometryParseError) Unwrap() error {
+	return e.error
+}
+
 type MySQLStreamingError struct {
 	error
 	Retryable bool