return WKT(s) as a result of GeoValidate

flow/connectors/postgres/cdc.go

@@ -537,16 +537,21 @@ func (p *PostgresCDCSource) decodeColumnData(data []byte, dataType uint32, forma
 
 	typeName, ok := p.customTypeMapping[dataType]
 	if ok {
+		var wkt string
 		customQKind := customTypeToQKind(typeName)
 		if customQKind == qvalue.QValueKindGeography || customQKind == qvalue.QValueKindGeometry {
-			err := GeoValidate(string(data))
+			wkt, err = GeoValidate(string(data))
 			if err != nil {
-				return &qvalue.QValue{Kind: customQKind,
-					Value: nil}, nil
+				return &qvalue.QValue{
+					Kind:  customQKind,
+					Value: nil,
+				}, nil
 			}
 		}
-		return &qvalue.QValue{Kind: customQKind,
-			Value: string(data)}, nil
+		return &qvalue.QValue{
+			Kind:  customQKind,
+			Value: wkt,
+		}, nil
 	}
 
 	return &qvalue.QValue{Kind: qvalue.QValueKindString, Value: string(data)}, nil

flow/connectors/postgres/qrep_query_executor.go

@@ -436,9 +436,11 @@ func mapRowToQRecord(row pgx.Rows, fds []pgconn.FieldDescription,
 			customQKind := customTypeToQKind(typeName)
 			if customQKind == qvalue.QValueKindGeography || customQKind == qvalue.QValueKindGeometry {
 				wkbString, ok := values[i].(string)
-				err := GeoValidate(wkbString)
+				wkt, err := GeoValidate(wkbString)
 				if err != nil || !ok {
 					values[i] = nil
+				} else {
+					values[i] = wkt
 				}
 			}
 			customTypeVal := qvalue.QValue{

flow/connectors/postgres/qvalue_convert.go

@@ -407,24 +407,28 @@ func customTypeToQKind(typeName string) qvalue.QValueKind {
 	return qValueKind
 }
 
-func GeoValidate(hexWkb string) error {
-	log.Infof("Validating geometry shape %s", hexWkb)
+// returns the WKT representation of the geometry object if it is valid
+func GeoValidate(hexWkb string) (string, error) {
 	// Decode the WKB hex string into binary
 	wkb, hexErr := hex.DecodeString(hexWkb)
 	if hexErr != nil {
 		log.Warnf("Ignoring invalid WKB: %s", hexWkb)
-		return hexErr
+		return "", hexErr
 	}
+
 	// UnmarshalWKB performs geometry validation along with WKB parsing
 	geometryObject, geoErr := geom.NewGeomFromWKB(wkb)
 	if geoErr != nil {
 		log.Warnf("Ignoring invalid geometry WKB %s: %v", hexWkb, geoErr)
-		return geoErr
+		return "", geoErr
 	}
+
 	invalidReason := geometryObject.IsValidReason()
 	if invalidReason != "Valid Geometry" {
 		log.Warnf("Ignoring invalid geometry shape %s: %s", hexWkb, invalidReason)
-		return errors.New(invalidReason)
+		return "", errors.New(invalidReason)
 	}
-	return nil
+
+	wkt := geometryObject.ToWKT()
+	return wkt, nil
 }