fixed it gis deserializer

Author: Mrart

docs/content.zh/docs/connectors/pipeline-connectors/postgres.md

@@ -621,11 +621,11 @@ PostgreSQL 通过 PostGIS 扩展支持空间数据类型：
     <tbody>
       <tr>
         <td>GEOMETRY(POINT, xx)</td>
-        <td>{"coordinates":[{"x":174.9479,"y":-36.7208,"z":"NaN","m":"NaN","valid":true}],"type":"Point","srid":3187}"</td>
+        <td>{"coordinates":"[174.9479, -36.7208]","type":"Point","srid":3187}"</td>
       </tr>
       <tr>
         <td>GEOGRAPHY(MULTILINESTRING)</td>
-        <td>{"coordinates":[{"x":169.1321,"y":-44.7032,"z":"NaN","m":"NaN","valid":true},{"x":167.8974,"y":-44.6414,"z":"NaN","m":"NaN","valid":true}],"type":"MultiLineString","srid":4326}</td>
+        <td>{"coordinates":"[169.1321, -44.7032],[167.8974, -44.6414]","type":"MultiLineString","srid":4326}</td>
       </tr>
     </tbody>
 </table>

docs/content/docs/connectors/pipeline-connectors/postgres.md

@@ -616,11 +616,11 @@ The former is used for small-area planar data, while the latter is used for larg
     <tbody>
       <tr>
         <td>GEOMETRY(POINT, xx)</td>
-        <td>{"coordinates":[{"x":174.9479,"y":-36.7208,"z":"NaN","m":"NaN","valid":true}],"type":"Point","srid":3187}"</td>
+        <td>{"coordinates":"[174.9479, -36.7208]","type":"Point","srid":3187}"</td>
       </tr>
       <tr>
         <td>GEOGRAPHY(MULTILINESTRING)</td>
-        <td>{"coordinates":[{"x":169.1321,"y":-44.7032,"z":"NaN","m":"NaN","valid":true},{"x":167.8974,"y":-44.6414,"z":"NaN","m":"NaN","valid":true}],"type":"MultiLineString","srid":4326}</td>
+        <td>{"coordinates":"[169.1321, -44.7032],[167.8974, -44.6414]","type":"MultiLineString","srid":4326}</td>
       </tr>
     </tbody>
 </table>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializer.java

@@ -34,13 +34,17 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.source.SourceRecord;
+import org.locationtech.jts.geom.Coordinate;
 import org.locationtech.jts.io.WKBReader;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 /** Event deserializer for {@link PostgresDataSource}. */
 @Internal
@@ -128,7 +132,20 @@ protected Object convertToString(Object dbzObj, Schema schema) {
                 if (geometryType.equals("GeometryCollection")) {
                     geometryInfo.put("geometries", jtsGeom.toText());
                 } else {
-                    geometryInfo.put("coordinates", jtsGeom.getCoordinates());
+                    Coordinate[] coordinates = jtsGeom.getCoordinates();
+                    List<double[]> coordinateList = new ArrayList<>();
+                    if (coordinates != null) {
+                        for (Coordinate coordinate : coordinates) {
+                            coordinateList.add(new double[] {coordinate.x, coordinate.y});
+                            geometryInfo.put(
+                                    "coordinates", new double[] {coordinate.x, coordinate.y});
+                        }
+                    }
+                    geometryInfo.put(
+                            "coordinates",
+                            coordinateList.stream()
+                                    .map(Arrays::toString)
+                                    .collect(Collectors.joining(",")));
                 }
                 geometryInfo.put("srid", srid.orElse(0));
                 return BinaryStringData.fromString(OBJECT_MAPPER.writeValueAsString(geometryInfo));

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java

@@ -201,9 +201,9 @@ public void testFullTypes() throws Exception {
                     64822000,
                     DecimalData.fromBigDecimal(new BigDecimal("500"), 10, 0),
                     BinaryStringData.fromString(
-                            "{\"coordinates\":[{\"x\":174.9479,\"y\":-36.7208,\"z\":\"NaN\",\"valid\":true,\"m\":\"NaN\"}],\"type\":\"Point\",\"srid\":3187}"),
+                            "{\"coordinates\":\"[174.9479, -36.7208]\",\"type\":\"Point\",\"srid\":3187}"),
                     BinaryStringData.fromString(
-                            "{\"coordinates\":[{\"x\":169.1321,\"y\":-44.7032,\"z\":\"NaN\",\"m\":\"NaN\",\"valid\":true},{\"x\":167.8974,\"y\":-44.6414,\"z\":\"NaN\",\"m\":\"NaN\",\"valid\":true}],\"type\":\"MultiLineString\",\"srid\":4326}"),
+                            "{\"coordinates\":\"[169.1321, -44.7032],[167.8974, -44.6414]\",\"type\":\"MultiLineString\",\"srid\":4326}"),
                     true,
                     new byte[] {10},
                     new byte[] {42},
@@ -224,7 +224,7 @@ public void testFullTypes() throws Exception {
                                     + "</preferences>\n"
                                     + "</user>"),
                     BinaryStringData.fromString(
-                            "{\"coordinates\":[{\"x\":3.456,\"y\":7.89,\"z\":\"NaN\",\"m\":\"NaN\",\"valid\":true}],\"type\":\"Point\",\"srid\":0}"),
+                            "{\"coordinates\":\"[3.456, 7.89]\",\"type\":\"Point\",\"srid\":0}"),
                     BinaryStringData.fromString("foo.bar.baz"),
                     BinaryStringData.fromString("JohnDoe"),
                     BinaryStringData.fromString("{\"size\":\"L\",\"color\":\"blue\"}"),