Draft: Do not review, add support for specific geometry types

flow/connectors/mysql/qvalue_convert.go

@@ -92,7 +92,26 @@ func qkindFromMysql(field *mysql.Field) (qvalue.QValueKind, error) {
 	case mysql.MYSQL_TYPE_VAR_STRING, mysql.MYSQL_TYPE_STRING:
 		return qvalue.QValueKindString, nil
 	case mysql.MYSQL_TYPE_GEOMETRY:
-		return qvalue.QValueKindGeometry, nil
+		// Check the column type name to determine specific geometry type
+		colType := strings.ToLower(string(field.Name))
+		switch {
+		case strings.Contains(colType, "point"):
+			return qvalue.QValueKindPoint, nil
+		case strings.Contains(colType, "linestring"):
+			return qvalue.QValueKindLineString, nil
+		case strings.Contains(colType, "polygon"):
+			return qvalue.QValueKindPolygon, nil
+		case strings.Contains(colType, "multipoint"):
+			return qvalue.QValueKindMultiPoint, nil
+		case strings.Contains(colType, "multilinestring"):
+			return qvalue.QValueKindMultiLineString, nil
+		case strings.Contains(colType, "multipolygon"):
+			return qvalue.QValueKindMultiPolygon, nil
+		case strings.Contains(colType, "geometrycollection"):
+			return qvalue.QValueKindGeometryCollection, nil
+		default:
+			return qvalue.QValueKindGeometry, nil
+		}
 	case mysql.MYSQL_TYPE_VECTOR:
 		return qvalue.QValueKindArrayFloat32, nil
 	default:
@@ -154,8 +173,25 @@ func qkindFromMysqlColumnType(ct string) (qvalue.QValueKind, error) {
 		}
 	case "vector":
 		return qvalue.QValueKindArrayFloat32, nil
-	case "geometry", "point", "polygon", "linestring", "multipoint", "multipolygon", "geomcollection":
-		return qvalue.QValueKindGeometry, nil
+	case "geometry", "point", "linestring", "polygon", "multipoint", "multilinestring", "multipolygon", "geometrycollection":
+		switch ct {
+		case "point":
+			return qvalue.QValueKindPoint, nil
+		case "linestring":
+			return qvalue.QValueKindLineString, nil
+		case "polygon":
+			return qvalue.QValueKindPolygon, nil
+		case "multipoint":
+			return qvalue.QValueKindMultiPoint, nil
+		case "multilinestring":
+			return qvalue.QValueKindMultiLineString, nil
+		case "multipolygon":
+			return qvalue.QValueKindMultiPolygon, nil
+		case "geometrycollection":
+			return qvalue.QValueKindGeometryCollection, nil
+		default:
+			return qvalue.QValueKindGeometry, nil
+		}
 	default:
 		return qvalue.QValueKind(""), fmt.Errorf("unknown mysql type %s", ct)
 	}
@@ -213,17 +249,39 @@ func geometryValueFromBytes(wkbData []byte) (string, error) {
 	return wkt, nil
 }
 
-// Helper function to process geometry data and return a QValueGeometry
-func processGeometryData(data []byte) qvalue.QValueGeometry {
+// Helper function to process geometry data and return a QValue
+func processGeometryData(data []byte, qkind qvalue.QValueKind) qvalue.QValue {
+	var strVal string
 	// For geometry data, we need to convert from MySQL's binary format to WKT
 	if len(data) > 4 {
 		wkt, err := geometryValueFromBytes(data)
 		if err == nil {
-			return qvalue.QValueGeometry{Val: wkt}
+			strVal = wkt
+		} else {
+			strVal = string(data)
 		}
+	} else {
+		strVal = string(data)
+	}
+
+	switch qkind {
+	case qvalue.QValueKindPoint:
+		return qvalue.QValuePoint{Val: strVal}
+	case qvalue.QValueKindLineString:
+		return qvalue.QValueLineString{Val: strVal}
+	case qvalue.QValueKindPolygon:
+		return qvalue.QValuePolygon{Val: strVal}
+	case qvalue.QValueKindMultiPoint:
+		return qvalue.QValueMultiPoint{Val: strVal}
+	case qvalue.QValueKindMultiLineString:
+		return qvalue.QValueMultiLineString{Val: strVal}
+	case qvalue.QValueKindMultiPolygon:
+		return qvalue.QValueMultiPolygon{Val: strVal}
+	case qvalue.QValueKindGeometryCollection:
+		return qvalue.QValueGeometryCollection{Val: strVal}
+	default:
+		return qvalue.QValueGeometry{Val: strVal}
 	}
-	strVal := string(data)
-	return qvalue.QValueGeometry{Val: strVal}
 }
 
 func QValueFromMysqlFieldValue(qkind qvalue.QValueKind, fv mysql.FieldValue) (qvalue.QValue, error) {
@@ -306,8 +364,10 @@ func QValueFromMysqlFieldValue(qkind qvalue.QValueKind, fv mysql.FieldValue) (qv
 			return qvalue.QValueBytes{Val: slices.Clone(v)}, nil
 		case qvalue.QValueKindJSON:
 			return qvalue.QValueJSON{Val: string(v)}, nil
-		case qvalue.QValueKindGeometry:
-			return processGeometryData(v), nil
+		case qvalue.QValueKindGeometry, qvalue.QValueKindPoint, qvalue.QValueKindLineString,
+			qvalue.QValueKindPolygon, qvalue.QValueKindMultiPoint, qvalue.QValueKindMultiLineString,
+			qvalue.QValueKindMultiPolygon, qvalue.QValueKindGeometryCollection:
+			return processGeometryData(v, qkind), nil
 		case qvalue.QValueKindNumeric:
 			val, err := decimal.NewFromString(unsafeString)
 			if err != nil {
@@ -438,7 +498,7 @@ func QValueFromMysqlRowEvent(
 			return qvalue.QValueJSON{Val: string(val)}, nil
 		case qvalue.QValueKindGeometry:
 			// Handle geometry data as binary (WKB format)
-			return processGeometryData(val), nil
+			return processGeometryData(val, qkind), nil
 		case qvalue.QValueKindArrayFloat32:
 			floats := make([]float32, 0, len(val)/4)
 			for i := 0; i < len(val); i += 4 {