This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-kafka-connector.git
The following commit(s) were added to refs/heads/master by this push:
new f5040b7 [feature] Support importing MySQL geographic data types and
PostgreSQL array data types (#25)
f5040b7 is described below
commit f5040b77f0013333cfc12ad9aa82235454df8852
Author: Petrichor <[email protected]>
AuthorDate: Thu Jun 6 16:06:03 2024 +0800
[feature] Support importing MySQL geographic data types and PostgreSQL
array data types (#25)
---
pom.xml | 12 +++
.../kafka/connector/converter/RecordService.java | 2 +-
.../connector/converter/RecordTypeRegister.java | 8 ++
.../converter/type/AbstractGeometryType.java | 30 ++++++
.../doris/kafka/connector/converter/type/Type.java | 4 +
.../converter/type/debezium/ArrayType.java | 107 +++++++++++++++++++++
.../converter/type/debezium/GeographyType.java | 32 ++++++
.../converter/type/debezium/GeometryType.java | 47 +++++++++
.../converter/type/debezium/PointType.java | 46 +++++++++
.../connector/converter/type/util/GeoUtils.java | 67 +++++++++++++
.../connector/converter/TestRecordService.java | 62 +++++++++++-
11 files changed, 414 insertions(+), 3 deletions(-)
diff --git a/pom.xml b/pom.xml
index ddbeb2f..29005cb 100644
--- a/pom.xml
+++ b/pom.xml
@@ -90,6 +90,7 @@
<jackson.version>2.13.2.1</jackson.version>
<httpcomponents.version>4.5.13</httpcomponents.version>
<commons-io.version>2.3</commons-io.version>
+ <geometry.version>2.2.0</geometry.version>
</properties>
<repositories>
@@ -282,6 +283,17 @@
<artifactId>debezium-core</artifactId>
<version>${debezium.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.esri.geometry</groupId>
+ <artifactId>esri-geometry-api</artifactId>
+ <version>${geometry.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
<build>
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/RecordService.java
b/src/main/java/org/apache/doris/kafka/connector/converter/RecordService.java
index cd76c42..7c75139 100644
---
a/src/main/java/org/apache/doris/kafka/connector/converter/RecordService.java
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/RecordService.java
@@ -278,7 +278,7 @@ public class RecordService {
field.getSchema().isOptional()
? source.getWithoutDefault(fieldName)
: source.get(fieldName);
- Object convertValue = type.getValue(value);
+ Object convertValue = type.getValue(value, field.getSchema());
if (Objects.nonNull(convertValue) && !type.isNumber()) {
filedMapping.put(fieldName, convertValue.toString());
} else {
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/RecordTypeRegister.java
b/src/main/java/org/apache/doris/kafka/connector/converter/RecordTypeRegister.java
index e78909d..5cf1001 100644
---
a/src/main/java/org/apache/doris/kafka/connector/converter/RecordTypeRegister.java
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/RecordTypeRegister.java
@@ -36,11 +36,15 @@ import
org.apache.doris.kafka.connector.converter.type.connect.ConnectMapToConne
import
org.apache.doris.kafka.connector.converter.type.connect.ConnectStringType;
import org.apache.doris.kafka.connector.converter.type.connect.ConnectTimeType;
import
org.apache.doris.kafka.connector.converter.type.connect.ConnectTimestampType;
+import org.apache.doris.kafka.connector.converter.type.debezium.ArrayType;
import org.apache.doris.kafka.connector.converter.type.debezium.DateType;
+import org.apache.doris.kafka.connector.converter.type.debezium.GeographyType;
+import org.apache.doris.kafka.connector.converter.type.debezium.GeometryType;
import org.apache.doris.kafka.connector.converter.type.debezium.MicroTimeType;
import
org.apache.doris.kafka.connector.converter.type.debezium.MicroTimestampType;
import org.apache.doris.kafka.connector.converter.type.debezium.NanoTimeType;
import
org.apache.doris.kafka.connector.converter.type.debezium.NanoTimestampType;
+import org.apache.doris.kafka.connector.converter.type.debezium.PointType;
import org.apache.doris.kafka.connector.converter.type.debezium.TimeType;
import org.apache.doris.kafka.connector.converter.type.debezium.TimestampType;
import
org.apache.doris.kafka.connector.converter.type.debezium.VariableScaleDecimalType;
@@ -73,6 +77,10 @@ public class RecordTypeRegister {
registerType(ZonedTimeType.INSTANCE);
registerType(ZonedTimestampType.INSTANCE);
registerType(VariableScaleDecimalType.INSTANCE);
+ registerType(PointType.INSTANCE);
+ registerType(GeographyType.INSTANCE);
+ registerType(GeometryType.INSTANCE);
+ registerType(ArrayType.INSTANCE);
// Supported connect data types
registerType(ConnectBooleanType.INSTANCE);
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/AbstractGeometryType.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/AbstractGeometryType.java
new file mode 100644
index 0000000..e88fbaf
--- /dev/null
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/type/AbstractGeometryType.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.doris.kafka.connector.converter.type;
+
+import org.apache.doris.kafka.connector.converter.type.doris.DorisType;
+import org.apache.kafka.connect.data.Schema;
+
+public abstract class AbstractGeometryType extends AbstractType {
+ @Override
+ public String getTypeName(Schema schema) {
+ return DorisType.STRING;
+ }
+}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/Type.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/Type.java
index 698e838..6be040a 100644
--- a/src/main/java/org/apache/doris/kafka/connector/converter/type/Type.java
+++ b/src/main/java/org/apache/doris/kafka/connector/converter/type/Type.java
@@ -43,6 +43,10 @@ public interface Type {
/** Get the actual converted value based on the column type. */
Object getValue(Object sourceValue);
+ default Object getValue(Object sourcevalue, Schema schema) {
+ return getValue(sourcevalue);
+ }
+
String getTypeName(Schema schema);
boolean isNumber();
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/ArrayType.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/ArrayType.java
new file mode 100644
index 0000000..870a346
--- /dev/null
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/ArrayType.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.doris.kafka.connector.converter.type.debezium;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import org.apache.doris.kafka.connector.cfg.DorisOptions;
+import org.apache.doris.kafka.connector.converter.RecordTypeRegister;
+import org.apache.doris.kafka.connector.converter.type.AbstractType;
+import org.apache.doris.kafka.connector.converter.type.Type;
+import org.apache.doris.kafka.connector.converter.type.doris.DorisType;
+import org.apache.kafka.connect.data.Schema;
+
+public class ArrayType extends AbstractType {
+ private static final String ARRAY_TYPE_TEMPLATE = "%s<%s>";
+ public static final ArrayType INSTANCE = new ArrayType();
+ private DorisOptions dorisOptions;
+ private RecordTypeRegister recordTypeRegister;
+
+ @Override
+ public void configure(DorisOptions dorisOptions) {
+ if (this.dorisOptions == null && this.recordTypeRegister == null) {
+ this.dorisOptions = dorisOptions;
+ registerNestedArrayType();
+ }
+ }
+
+ @Override
+ public String[] getRegistrationKeys() {
+ return new String[] {"ARRAY"};
+ }
+
+ @Override
+ public String getTypeName(Schema schema) {
+ if (schema.valueSchema().isOptional()) {
+ Schema valueSchema = schema.valueSchema();
+ String type =
+ Objects.nonNull(valueSchema.name())
+ ? valueSchema.name()
+ : valueSchema.type().name();
+ if (recordTypeRegister == null) {
+ registerNestedArrayType();
+ }
+ Type valueType = recordTypeRegister.getTypeRegistry().get(type);
+ if (valueType == null) {
+ return DorisType.STRING;
+ }
+ String typeName = valueType.getTypeName(schema);
+ return String.format(ARRAY_TYPE_TEMPLATE, DorisType.ARRAY,
typeName);
+ }
+ return DorisType.STRING;
+ }
+
+ @Override
+ public Object getValue(Object sourceValue, Schema schema) {
+
+ if (sourceValue == null) {
+ return null;
+ }
+ Schema valueSchema = schema.valueSchema();
+ String type =
+ Objects.nonNull(valueSchema.name())
+ ? valueSchema.name()
+ : valueSchema.type().name();
+
+ if (sourceValue instanceof List) {
+ List<Object> resultList = new ArrayList<>();
+ ArrayList<?> convertedValue = (ArrayList<?>) sourceValue;
+ if (recordTypeRegister == null) {
+ registerNestedArrayType();
+ }
+ Type valueType = recordTypeRegister.getTypeRegistry().get(type);
+ if (valueType == null) {
+ return sourceValue;
+ }
+
+ for (Object value : convertedValue) {
+ resultList.add(valueType.getValue(value, valueSchema));
+ }
+ return resultList;
+ }
+
+ return sourceValue;
+ }
+
+ private void registerNestedArrayType() {
+ this.recordTypeRegister = new RecordTypeRegister(dorisOptions);
+ }
+}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/GeographyType.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/GeographyType.java
new file mode 100644
index 0000000..85b8881
--- /dev/null
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/GeographyType.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.doris.kafka.connector.converter.type.debezium;
+
+import io.debezium.data.geometry.Geography;
+import org.apache.doris.kafka.connector.converter.type.AbstractGeometryType;
+
+public class GeographyType extends AbstractGeometryType {
+ public static final GeographyType INSTANCE = new GeographyType();
+
+ @Override
+ public String[] getRegistrationKeys() {
+ return new String[] {Geography.LOGICAL_NAME};
+ }
+}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/GeometryType.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/GeometryType.java
new file mode 100644
index 0000000..c6e153b
--- /dev/null
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/GeometryType.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.doris.kafka.connector.converter.type.debezium;
+
+import io.debezium.data.geometry.Geometry;
+import org.apache.doris.kafka.connector.converter.type.AbstractGeometryType;
+import org.apache.doris.kafka.connector.converter.type.util.GeoUtils;
+import org.apache.kafka.connect.data.Struct;
+
+public class GeometryType extends AbstractGeometryType {
+ public static final GeometryType INSTANCE = new GeometryType();
+
+ @Override
+ public String[] getRegistrationKeys() {
+ return new String[] {Geometry.LOGICAL_NAME};
+ }
+
+ @Override
+ public Object getValue(Object sourceValue) {
+ if (sourceValue == null) {
+ return null;
+ }
+
+ if (sourceValue instanceof Struct) {
+ return GeoUtils.handleGeoStructData(sourceValue);
+ }
+
+ return sourceValue;
+ }
+}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/PointType.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/PointType.java
new file mode 100644
index 0000000..da01507
--- /dev/null
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/type/debezium/PointType.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.doris.kafka.connector.converter.type.debezium;
+
+import io.debezium.data.geometry.Point;
+import org.apache.doris.kafka.connector.converter.type.AbstractGeometryType;
+import org.apache.doris.kafka.connector.converter.type.util.GeoUtils;
+import org.apache.kafka.connect.data.Struct;
+
+public class PointType extends AbstractGeometryType {
+ public static final PointType INSTANCE = new PointType();
+
+ @Override
+ public String[] getRegistrationKeys() {
+ return new String[] {Point.LOGICAL_NAME};
+ }
+
+ @Override
+ public Object getValue(Object sourceValue) {
+ if (sourceValue == null) {
+ return null;
+ }
+ if (sourceValue instanceof Struct) {
+ return GeoUtils.handleGeoStructData(sourceValue);
+ }
+
+ return sourceValue;
+ }
+}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/converter/type/util/GeoUtils.java
b/src/main/java/org/apache/doris/kafka/connector/converter/type/util/GeoUtils.java
new file mode 100644
index 0000000..9c0fc65
--- /dev/null
+++
b/src/main/java/org/apache/doris/kafka/connector/converter/type/util/GeoUtils.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.doris.kafka.connector.converter.type.util;
+
+import com.esri.core.geometry.ogc.OGCGeometry;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.kafka.connect.data.Struct;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GeoUtils {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(GeoUtils.class);
+
+ private GeoUtils() {}
+
+ public static Object handleGeoStructData(Object sourceValue) {
+ // the Geometry datatype in MySQL will be converted to
+ // a String with Json format
+ final ObjectMapper objectMapper = new ObjectMapper();
+ Struct geometryStruct = (Struct) sourceValue;
+
+ try {
+ byte[] wkb = geometryStruct.getBytes("wkb");
+ String geoJson =
OGCGeometry.fromBinary(ByteBuffer.wrap(wkb)).asGeoJson();
+ JsonNode originGeoNode = objectMapper.readTree(geoJson);
+
+ Optional<Integer> srid =
Optional.ofNullable(geometryStruct.getInt32("srid"));
+ Map<String, Object> geometryInfo = new HashMap<>();
+ String geometryType = originGeoNode.get("type").asText();
+
+ geometryInfo.put("type", geometryType);
+ if ("GeometryCollection".equals(geometryType)) {
+ geometryInfo.put("geometries",
originGeoNode.get("geometries"));
+ } else {
+ geometryInfo.put("coordinates",
originGeoNode.get("coordinates"));
+ }
+
+ geometryInfo.put("srid", srid.orElse(0));
+ return geometryInfo;
+ } catch (Exception e) {
+ LOGGER.warn("Failed to parse Geometry datatype, converting the
value to null", e);
+ return null;
+ }
+ }
+}
diff --git
a/src/test/java/org/apache/doris/kafka/connector/converter/TestRecordService.java
b/src/test/java/org/apache/doris/kafka/connector/converter/TestRecordService.java
index 0142259..74af7c9 100644
---
a/src/test/java/org/apache/doris/kafka/connector/converter/TestRecordService.java
+++
b/src/test/java/org/apache/doris/kafka/connector/converter/TestRecordService.java
@@ -75,7 +75,7 @@ public class TestRecordService {
props.put("debezium.schema.evolution", "basic");
props.put(
"doris.topic2table.map",
-
"avro_schema.wdl_test.example_table:example_table,normal.wdl_test.test_sink_normal:test_sink_normal");
+
"avro_schema.wdl_test.example_table:example_table,normal.wdl_test.test_sink_normal:test_sink_normal,mysql_test.doris_test.geo_table:geo_table,pg_test.doris_cdc.all_array_types:all_array_types");
recordService = new RecordService(new DorisOptions((Map) props));
HashMap<String, String> config = new HashMap<>();
jsonConverter.configure(config, false);
@@ -237,6 +237,65 @@ public class TestRecordService {
"{\"name\":\"doris\",\"key\":\"1\"}\n{\"name\":\"doris\",\"key\":\"2\"}", s);
}
+ @Test
+ public void processMySQlGeoStructRecord() throws IOException {
+ String schemaStr =
+
"{\"keysType\":\"UNIQUE_KEYS\",\"properties\":[{\"name\":\"id\",\"aggregation_type\":\"\",\"comment\":\"\",\"type\":\"INT\"},{\"name\":\"geo_point\",\"aggregation_type\":\"REPLACE\",\"comment\":\"\",\"type\":\"STRING\"},{\"name\":\"geo_linestring\",\"aggregation_type\":\"REPLACE\",\"comment\":\"\",\"type\":\"STRING\"},{\"name\":\"geo_polygon\",\"aggregation_type\":\"REPLACE\",\"comment\":\"\",\"type\":\"STRING\"},{\"name\":\"geo_multipoint\",\"aggregation_type\":\"REPLACE
[...]
+ Schema schema = null;
+ try {
+ schema = objectMapper.readValue(schemaStr, Schema.class);
+ } catch (JsonProcessingException e) {
+ throw new DorisException(e);
+ }
+ mockRestService
+ .when(() -> RestService.getSchema(any(), any(), any(), any()))
+ .thenReturn(schema);
+
+ String topic = "mysql_test.doris_test.geo_table";
+
+ // no delete value
+ String noDeleteValue =
+
"{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"double\",\"optional\":false,\"field\":\"x\"},{\"type\":\"double\",\"optional\":false,\"field\":\"y\"},{\"type\":\"bytes\",\"optional\":true,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Point\",\"version\":1,
[...]
+ String expectedNoDeleteValue =
+ "{\"id\":1,\"geo_point\":\"{coordinates=[1,1], type=Point,
srid=0}\",\"geo_linestring\":\"{coordinates=[[0,0],[1,1],[2,2]],
type=LineString,
srid=0}\",\"geo_polygon\":\"{coordinates=[[[0,0],[1,0],[1,1],[0,0]]],
type=Polygon, srid=0}\",\"geo_multipoint\":\"{coordinates=[[0,0],[1,1]],
type=MultiPoint,
srid=0}\",\"geo_multilinestring\":\"{coordinates=[[[0,0],[1,1]],[[2,2],[3,3]]],
type=MultiLineString,
srid=0}\",\"geo_multipolygon\":\"{coordinates=[[[[0,0],[1,0],[1,1],[0,0]] [...]
+ buildProcessStructRecord(topic, noDeleteValue, expectedNoDeleteValue);
+
+ // delete value
+ String deleteValue =
+
"{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"double\",\"optional\":false,\"field\":\"x\"},{\"type\":\"double\",\"optional\":false,\"field\":\"y\"},{\"type\":\"bytes\",\"optional\":true,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Point\",\"version\":1,
[...]
+ String expectedDeleteValue =
+ "{\"id\":1,\"geo_point\":\"{coordinates=[1,1], type=Point,
srid=0}\",\"geo_linestring\":\"{coordinates=[[0,0],[1,1],[2,2]],
type=LineString,
srid=0}\",\"geo_polygon\":\"{coordinates=[[[0,0],[1,0],[1,1],[0,0]]],
type=Polygon, srid=0}\",\"geo_multipoint\":\"{coordinates=[[0,0],[1,1]],
type=MultiPoint,
srid=0}\",\"geo_multilinestring\":\"{coordinates=[[[0,0],[1,1]],[[2,2],[3,3]]],
type=MultiLineString,
srid=0}\",\"geo_multipolygon\":\"{coordinates=[[[[0,0],[1,0],[1,1],[0,0]] [...]
+ buildProcessStructRecord(topic, deleteValue, expectedDeleteValue);
+ }
+
+ @Test
+ public void processPostgreArrayStructRecord() throws IOException {
+ String schemaStr =
+
"{\"keysType\":\"DUP_KEYS\",\"properties\":[{\"name\":\"int_array\",\"aggregation_type\":\"\",\"comment\":\"\",\"type\":\"INT\"},{\"name\":\"smallint_array\",\"aggregation_type\":\"NONE\",\"comment\":\"\",\"type\":\"ARRAY\"},{\"name\":\"bigint_array\",\"aggregation_type\":\"NONE\",\"comment\":\"\",\"type\":\"ARRAY\"},{\"name\":\"float_array\",\"aggregation_type\":\"NONE\",\"comment\":\"\",\"type\":\"ARRAY\"},{\"name\":\"double_array\",\"aggregation_type\":\"NONE\",\"comme
[...]
+ Schema schema = null;
+ try {
+ schema = objectMapper.readValue(schemaStr, Schema.class);
+ } catch (JsonProcessingException e) {
+ throw new DorisException(e);
+ }
+ mockRestService
+ .when(() -> RestService.getSchema(any(), any(), any(), any()))
+ .thenReturn(schema);
+
+ String topic = "pg_test.doris_cdc.all_array_types";
+ String noDeleteArrayValue =
+
"{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":true,\"field\":\"id\"},{\"type\":\"array\",\"items\":{\"type\":\"int16\",\"optional\":true},\"optional\":true,\"field\":\"smallint_array\"},{\"type\":\"array\",\"items\":{\"type\":\"int64\",\"optional\":true},\"optional\":true,\"field\":\"bigint_array\"},{\"type\":\"array\",\"items\":{\"type\":\"float\",\"optional\":true},\"optional\":true,\"field\":\"float_arra
[...]
+ String expectedNoDeleteArrayValue =
+ "{\"id\":1,\"smallint_array\":\"[100, 200,
300]\",\"bigint_array\":\"[10000000000, 20000000000,
30000000000]\",\"float_array\":\"[1.1, 2.2, 3.3]\",\"double_array\":\"[10.1,
20.2, 30.3]\",\"numeric_array\":\"[123.45, 678.90]\",\"text_array\":\"[text1,
text2, text3]\",\"varchar_array\":\"[varchar1, varchar2,
varchar3]\",\"char_array\":\"[c, h, a]\",\"bool_array\":\"[true, false,
true]\",\"date_array\":\"[2024-01-01,
2024-01-02]\",\"timestamp_array\":\"[2022-01-01T10:00, 202 [...]
+ buildProcessStructRecord(topic, noDeleteArrayValue,
expectedNoDeleteArrayValue);
+
+ String deletedArrayValue =
+
"{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":true,\"field\":\"id\"},{\"type\":\"array\",\"items\":{\"type\":\"int16\",\"optional\":true},\"optional\":true,\"field\":\"smallint_array\"},{\"type\":\"array\",\"items\":{\"type\":\"int64\",\"optional\":true},\"optional\":true,\"field\":\"bigint_array\"},{\"type\":\"array\",\"items\":{\"type\":\"float\",\"optional\":true},\"optional\":true,\"field\":\"float_arra
[...]
+ String expectedDeleteArrayValue =
+ "{\"id\":1,\"smallint_array\":\"[100, 200,
300]\",\"bigint_array\":\"[10000000000, 20000000000,
30000000000]\",\"float_array\":\"[1.1, 2.2, 3.3]\",\"double_array\":\"[10.1,
20.2, 30.3]\",\"numeric_array\":\"[123.45, 678.90]\",\"text_array\":\"[text1,
text2, text3]\",\"varchar_array\":\"[varchar1, varchar2,
varchar3]\",\"char_array\":\"[c, h, a]\",\"bool_array\":\"[true, false,
true]\",\"date_array\":\"[2024-01-01,
2024-01-02]\",\"timestamp_array\":\"[2022-01-01T10:00, 202 [...]
+ buildProcessStructRecord(topic, deletedArrayValue,
expectedDeleteArrayValue);
+ }
+
@Test
public void processMapRecord() throws JsonProcessingException {
Map<String, String> mapValue = new HashMap<>();
@@ -264,6 +323,5 @@ public class TestRecordService {
@After
public void close() {
mockRestService.close();
- ;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
