[e2e](test) add partial column update and debezium ingestion e2e case (#38)
diff --git a/src/test/java/org/apache/doris/kafka/connector/e2e/sink/AbstractKafka2DorisSink.java b/src/test/java/org/apache/doris/kafka/connector/e2e/sink/AbstractKafka2DorisSink.java
index c4c7fe4..d50556f 100644
--- a/src/test/java/org/apache/doris/kafka/connector/e2e/sink/AbstractKafka2DorisSink.java
+++ b/src/test/java/org/apache/doris/kafka/connector/e2e/sink/AbstractKafka2DorisSink.java
@@ -108,6 +108,18 @@
LOG.info("Create doris table successfully. sql={}", sql);
}
+ protected void insertTable(String sql) {
+ LOG.info("Will insert data to Doris table. SQL: {}", sql);
+ try {
+ Statement statement = getJdbcConnection().createStatement();
+ int rowCount = statement.executeUpdate(sql);
+ LOG.info("Inserted {} item data into the Doris table.", rowCount);
+ } catch (SQLException e) {
+ throw new DorisException("Failed to insert data to Doris table.", e);
+ }
+ LOG.info("Data insertion to Doris table was successful. SQL: {}", sql);
+ }
+
private static void initDorisBase() {
if (Objects.nonNull(dorisContainerService)) {
return;
diff --git a/src/test/java/org/apache/doris/kafka/connector/e2e/sink/stringconverter/StringMsgE2ETest.java b/src/test/java/org/apache/doris/kafka/connector/e2e/sink/stringconverter/StringMsgE2ETest.java
index 227a203..3143461 100644
--- a/src/test/java/org/apache/doris/kafka/connector/e2e/sink/stringconverter/StringMsgE2ETest.java
+++ b/src/test/java/org/apache/doris/kafka/connector/e2e/sink/stringconverter/StringMsgE2ETest.java
@@ -125,6 +125,65 @@
checkResult(expected, query, 3);
}
+ @Test
+ public void testPartialUpdate() throws Exception {
+ initialize("src/test/resources/e2e/string_converter/partial_update.json");
+ String topic = "partial_update_test";
+ String msg1 =
+ "{\"id\":1,\"col1\":\"after_update_col1_1\",\"col2\":\"after_update_col2_1\"}";
+ String msg2 =
+ "{\"id\":2,\"col1\":\"after_update_col1_2\",\"col2\":\"after_update_col2_2\"}";
+
+ produceMsg2Kafka(topic, msg1);
+ produceMsg2Kafka(topic, msg2);
+
+ String tableSql =
+ loadContent("src/test/resources/e2e/string_converter/partial_update_tab.sql");
+ String insertSql =
+ loadContent(
+ "src/test/resources/e2e/string_converter/insert_partial_update_tab.sql");
+ createTable(tableSql);
+ Thread.sleep(2000);
+ insertTable(insertSql);
+ Thread.sleep(15000);
+ kafkaContainerService.registerKafkaConnector(connectorName, jsonMsgConnectorContent);
+
+ String table = dorisOptions.getTopicMapTable(topic);
+ List<String> expected =
+ Arrays.asList(
+ "1,after_update_col1_1,after_update_col2_1,before_update_col3_1",
+ "2,after_update_col1_2,after_update_col2_2,before_update_col3_2");
+ Thread.sleep(10000);
+ String query =
+ String.format("select id,col1,col2,col3 from %s.%s order by id", database, table);
+ checkResult(expected, query, 4);
+ }
+
+ @Test
+ public void testDebeziumIngestionFullTypes() throws Exception {
+ initialize("src/test/resources/e2e/string_converter/full_types.json");
+ String topic = "full_types";
+ String msg1 =
+ "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"struct\",\"fields\":[{\"type\":\"int64\",\"optional\":false,\"field\":\"id\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny_un_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny_un_z_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"small_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"small_un_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"small_un_z_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"medium_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"medium_un_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"medium_un_z_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"int_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"int_un_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"int_un_z_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"int11_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"big_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"big_un_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"big_un_z_c\"},{\"type\":\"string\",\"optional\":true,\"field\":\"varchar_c\"},{\"type\":\"string\",\"optional\":true,\"field\":\"char_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"real_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"float_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"float_un_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"float_un_z_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"double_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"double_un_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"double_un_z_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"4\",\"connect.decimal.precision\":\"8\"},\"field\":\"decimal_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"4\",\"connect.decimal.precision\":\"8\"},\"field\":\"decimal_un_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"4\",\"connect.decimal.precision\":\"8\"},\"field\":\"decimal_un_z_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"0\",\"connect.decimal.precision\":\"6\"},\"field\":\"numeric_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"1\",\"connect.decimal.precision\":\"65\"},\"field\":\"big_decimal_c\"},{\"type\":\"boolean\",\"optional\":true,\"field\":\"bit1_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny1_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"boolean_c\"},{\"type\":\"int32\",\"optional\":true,\"name\":\"io.debezium.time.Date\",\"version\":1,\"field\":\"date_c\"},{\"type\":\"int64\",\"optional\":true,\"name\":\"io.debezium.time.MicroTime\",\"version\":1,\"field\":\"time_c\"},{\"type\":\"int64\",\"optional\":true,\"name\":\"io.debezium.time.Timestamp\",\"version\":1,\"field\":\"datetime_c\"},{\"type\":\"string\",\"optional\":false,\"name\":\"io.debezium.time.ZonedTimestamp\",\"version\":1,\"default\":\"1970-01-01T00:00:00Z\",\"field\":\"timestamp_c\"},{\"type\":\"string\",\"optional\":true,\"field\":\"text_c\"},{\"type\":\"int32\",\"optional\":true,\"name\":\"io.debezium.time.Year\",\"version\":1,\"field\":\"year_c\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.Enum\",\"version\":1,\"parameters\":{\"allowed\":\"red,white\"},\"default\":\"red\",\"field\":\"enum_c\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.EnumSet\",\"version\":1,\"parameters\":{\"allowed\":\"a,b\"},\"field\":\"set_c\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.Json\",\"version\":1,\"field\":\"json_c\"},{\"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,\"doc\":\"Geometry (POINT)\",\"field\":\"point_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"geometry_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"linestring_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"polygon_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"multipoint_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"multiline_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"multipolygon_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"geometrycollection_c\"}],\"optional\":true,\"name\":\"mysql_test.doris_test.full_types.Value\",\"field\":\"before\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"int64\",\"optional\":false,\"field\":\"id\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny_un_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny_un_z_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"small_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"small_un_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"small_un_z_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"medium_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"medium_un_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"medium_un_z_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"int_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"int_un_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"int_un_z_c\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"int11_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"big_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"big_un_c\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"big_un_z_c\"},{\"type\":\"string\",\"optional\":true,\"field\":\"varchar_c\"},{\"type\":\"string\",\"optional\":true,\"field\":\"char_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"real_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"float_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"float_un_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"float_un_z_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"double_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"double_un_c\"},{\"type\":\"double\",\"optional\":true,\"field\":\"double_un_z_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"4\",\"connect.decimal.precision\":\"8\"},\"field\":\"decimal_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"4\",\"connect.decimal.precision\":\"8\"},\"field\":\"decimal_un_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"4\",\"connect.decimal.precision\":\"8\"},\"field\":\"decimal_un_z_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"0\",\"connect.decimal.precision\":\"6\"},\"field\":\"numeric_c\"},{\"type\":\"bytes\",\"optional\":true,\"name\":\"org.apache.kafka.connect.data.Decimal\",\"version\":1,\"parameters\":{\"scale\":\"1\",\"connect.decimal.precision\":\"65\"},\"field\":\"big_decimal_c\"},{\"type\":\"boolean\",\"optional\":true,\"field\":\"bit1_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"tiny1_c\"},{\"type\":\"int16\",\"optional\":true,\"field\":\"boolean_c\"},{\"type\":\"int32\",\"optional\":true,\"name\":\"io.debezium.time.Date\",\"version\":1,\"field\":\"date_c\"},{\"type\":\"int64\",\"optional\":true,\"name\":\"io.debezium.time.MicroTime\",\"version\":1,\"field\":\"time_c\"},{\"type\":\"int64\",\"optional\":true,\"name\":\"io.debezium.time.Timestamp\",\"version\":1,\"field\":\"datetime_c\"},{\"type\":\"string\",\"optional\":false,\"name\":\"io.debezium.time.ZonedTimestamp\",\"version\":1,\"default\":\"1970-01-01T00:00:00Z\",\"field\":\"timestamp_c\"},{\"type\":\"string\",\"optional\":true,\"field\":\"text_c\"},{\"type\":\"int32\",\"optional\":true,\"name\":\"io.debezium.time.Year\",\"version\":1,\"field\":\"year_c\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.Enum\",\"version\":1,\"parameters\":{\"allowed\":\"red,white\"},\"default\":\"red\",\"field\":\"enum_c\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.EnumSet\",\"version\":1,\"parameters\":{\"allowed\":\"a,b\"},\"field\":\"set_c\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.Json\",\"version\":1,\"field\":\"json_c\"},{\"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,\"doc\":\"Geometry (POINT)\",\"field\":\"point_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"geometry_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"linestring_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"polygon_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"multipoint_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"multiline_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"multipolygon_c\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"bytes\",\"optional\":false,\"field\":\"wkb\"},{\"type\":\"int32\",\"optional\":true,\"field\":\"srid\"}],\"optional\":true,\"name\":\"io.debezium.data.geometry.Geometry\",\"version\":1,\"doc\":\"Geometry\",\"field\":\"geometrycollection_c\"}],\"optional\":true,\"name\":\"mysql_test.doris_test.full_types.Value\",\"field\":\"after\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"string\",\"optional\":false,\"field\":\"version\"},{\"type\":\"string\",\"optional\":false,\"field\":\"connector\"},{\"type\":\"string\",\"optional\":false,\"field\":\"name\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"ts_ms\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.Enum\",\"version\":1,\"parameters\":{\"allowed\":\"true,last,false,incremental\"},\"default\":\"false\",\"field\":\"snapshot\"},{\"type\":\"string\",\"optional\":false,\"field\":\"db\"},{\"type\":\"string\",\"optional\":true,\"field\":\"sequence\"},{\"type\":\"string\",\"optional\":true,\"field\":\"table\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"server_id\"},{\"type\":\"string\",\"optional\":true,\"field\":\"gtid\"},{\"type\":\"string\",\"optional\":false,\"field\":\"file\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"pos\"},{\"type\":\"int32\",\"optional\":false,\"field\":\"row\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"thread\"},{\"type\":\"string\",\"optional\":true,\"field\":\"query\"}],\"optional\":false,\"name\":\"io.debezium.connector.mysql.Source\",\"field\":\"source\"},{\"type\":\"string\",\"optional\":false,\"field\":\"op\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"ts_ms\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"string\",\"optional\":false,\"field\":\"id\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"total_order\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"data_collection_order\"}],\"optional\":true,\"field\":\"transaction\"}],\"optional\":false,\"name\":\"mysql_test.doris_test.full_types.Envelope\"},\"payload\":{\"before\":null,\"after\":{\"id\":1,\"tiny_c\":127,\"tiny_un_c\":255,\"tiny_un_z_c\":255,\"small_c\":32767,\"small_un_c\":65535,\"small_un_z_c\":65535,\"medium_c\":8388607,\"medium_un_c\":16777215,\"medium_un_z_c\":16777215,\"int_c\":2147483647,\"int_un_c\":4294967295,\"int_un_z_c\":4294967295,\"int11_c\":2147483647,\"big_c\":9223372036854775807,\"big_un_c\":-1,\"big_un_z_c\":-1,\"varchar_c\":\"Hello World\",\"char_c\":\"abc\",\"real_c\":123.102,\"float_c\":123.10199737548828,\"float_un_c\":123.10299682617188,\"float_un_z_c\":123.10399627685547,\"double_c\":404.4443,\"double_un_c\":404.4444,\"double_un_z_c\":404.4445,\"decimal_c\":\"EtaH\",\"decimal_un_c\":\"EtaI\",\"decimal_un_z_c\":\"EtaJ\",\"numeric_c\":\"AVo=\",\"big_decimal_c\":\"FJqkSQ==\",\"bit1_c\":false,\"tiny1_c\":1,\"boolean_c\":1,\"date_c\":18460,\"time_c\":64822000000,\"datetime_c\":1595008822000,\"timestamp_c\":\"2020-07-17T10:00:22Z\",\"text_c\":\"text\",\"year_c\":2021,\"enum_c\":\"red\",\"set_c\":\"a,b\",\"json_c\":\"{\\\"key1\\\":\\\"value1\\\"}\",\"point_c\":{\"x\":3.0,\"y\":1.0,\"wkb\":\"AQEAAAAAAAAAAAAIQAAAAAAAAPA/\",\"srid\":null},\"geometry_c\":{\"wkb\":\"AQMAAAABAAAABQAAAAAAAAAAAPA/AAAAAAAA8D8AAAAAAAAAQAAAAAAAAPA/AAAAAAAAAEAAAAAAAAAAQAAAAAAAAPA/AAAAAAAAAEAAAAAAAADwPwAAAAAAAPA/\",\"srid\":null},\"linestring_c\":{\"wkb\":\"AQIAAAADAAAAAAAAAAAACEAAAAAAAAAAAAAAAAAAAAhAAAAAAAAACEAAAAAAAAAIQAAAAAAAABRA\",\"srid\":null},\"polygon_c\":{\"wkb\":\"AQMAAAABAAAABQAAAAAAAAAAAPA/AAAAAAAA8D8AAAAAAAAAQAAAAAAAAPA/AAAAAAAAAEAAAAAAAAAAQAAAAAAAAPA/AAAAAAAAAEAAAAAAAADwPwAAAAAAAPA/\",\"srid\":null},\"multipoint_c\":{\"wkb\":\"AQQAAAACAAAAAQEAAAAAAAAAAADwPwAAAAAAAPA/AQEAAAAAAAAAAAAAQAAAAAAAAABA\",\"srid\":null},\"multiline_c\":{\"wkb\":\"AQUAAAACAAAAAQIAAAADAAAAAAAAAAAA8D8AAAAAAADwPwAAAAAAAABAAAAAAAAAAEAAAAAAAAAIQAAAAAAAAAhAAQIAAAACAAAAAAAAAAAAEEAAAAAAAAAQQAAAAAAAABRAAAAAAAAAFEA=\",\"srid\":null},\"multipolygon_c\":{\"wkb\":\"AQYAAAACAAAAAQMAAAABAAAABQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAkQAAAAAAAAAAAAAAAAAAAJEAAAAAAAAAkQAAAAAAAAAAAAAAAAAAAJEAAAAAAAAAAAAAAAAAAAAAAAQMAAAABAAAABQAAAAAAAAAAABRAAAAAAAAAFEAAAAAAAAAcQAAAAAAAABRAAAAAAAAAHEAAAAAAAAAcQAAAAAAAABRAAAAAAAAAHEAAAAAAAAAUQAAAAAAAABRA\",\"srid\":null},\"geometrycollection_c\":{\"wkb\":\"AQcAAAADAAAAAQEAAAAAAAAAAAAkQAAAAAAAACRAAQEAAAAAAAAAAAA+QAAAAAAAAD5AAQIAAAACAAAAAAAAAAAALkAAAAAAAAAuQAAAAAAAADRAAAAAAAAANEA=\",\"srid\":null}},\"source\":{\"version\":\"1.9.8.Final\",\"connector\":\"mysql\",\"name\":\"mysql_test\",\"ts_ms\":1720517925000,\"snapshot\":\"false\",\"db\":\"doris_test\",\"sequence\":null,\"table\":\"full_types\",\"server_id\":1,\"gtid\":null,\"file\":\"mysql-bin.000401\",\"pos\":17237,\"row\":0,\"thread\":5,\"query\":null},\"op\":\"c\",\"ts_ms\":1720518881444,\"transaction\":null}}";
+
+ produceMsg2Kafka(topic, msg1);
+
+ String tableSql =
+ loadContent(
+ "src/test/resources/e2e/string_converter/full_types_debezium_ingestion.sql");
+ createTable(tableSql);
+ Thread.sleep(2000);
+ kafkaContainerService.registerKafkaConnector(connectorName, jsonMsgConnectorContent);
+
+ String table = dorisOptions.getTopicMapTable(topic);
+ List<String> expected =
+ Arrays.asList(
+ "1,127,255,255,32767,65535,65535,8388607,16777215,16777215,2147483647,4294967295,4294967295,2147483647,9223372036854775807,-1,-1,Hello World,abc,123.102,123.102,123.103,123.104,404.4443,404.4444,404.4445,123.4567,123.4568,123.4569,346,34567892.1,false,true,true,2020-07-17,18:00:22,2020-07-17T18:00:22,2020-07-17T18:00:22,text,2021,red,a,b,{\"key1\":\"value1\"},{coordinates=[3,1], type=Point, srid=0},{coordinates=[[[1,1],[2,1],[2,2],[1,2],[1,1]]], type=Polygon, srid=0},{coordinates=[[3,0],[3,3],[3,5]], type=LineString, srid=0},{coordinates=[[[1,1],[2,1],[2,2],[1,2],[1,1]]], type=Polygon, srid=0},{coordinates=[[1,1],[2,2]], type=MultiPoint, srid=0},{coordinates=[[[1,1],[2,2],[3,3]],[[4,4],[5,5]]], type=MultiLineString, srid=0},{coordinates=[[[[0,0],[10,0],[10,10],[0,10],[0,0]]],[[[5,5],[7,5],[7,7],[5,7],[5,5]]]], type=MultiPolygon, srid=0},{geometries=[{\"type\":\"Point\",\"coordinates\":[10,10]},{\"type\":\"Point\",\"coordinates\":[30,30]},{\"type\":\"LineString\",\"coordinates\":[[15,15],[20,20]]}], type=GeometryCollection, srid=0}");
+ Thread.sleep(10000);
+ String query = String.format("select * from %s.%s order by id", database, table);
+ checkResult(expected, query, 51);
+ }
+
public void checkResult(List<String> expected, String query, int columnSize) throws Exception {
List<String> actual = new ArrayList<>();
@@ -143,6 +202,8 @@
actual.add(StringUtils.join(row, ","));
}
}
+ LOG.info("expected result: {}", Arrays.toString(expected.toArray()));
+ LOG.info("actual result: {}", Arrays.toString(actual.toArray()));
Assert.assertArrayEquals(expected.toArray(), actual.toArray());
}
diff --git a/src/test/resources/e2e/string_converter/full_types.json b/src/test/resources/e2e/string_converter/full_types.json
new file mode 100644
index 0000000..ef95aef
--- /dev/null
+++ b/src/test/resources/e2e/string_converter/full_types.json
@@ -0,0 +1,22 @@
+{
+ "name":"mysql_all_types",
+ "config":{
+ "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
+ "topics":"full_types",
+ "tasks.max":"1",
+ "doris.topic2table.map": "full_types:full_types_tab",
+ "buffer.count.records":"1",
+ "buffer.flush.time":"10",
+ "buffer.size.bytes":"10000000",
+ "doris.urls":"127.0.0.1",
+ "doris.user":"root",
+ "doris.password":"",
+ "doris.http.port":"8030",
+ "doris.query.port":"9030",
+ "doris.database":"debezium_ingestion_msg",
+ "converter.mode": "debezium_ingestion",
+ "load.model":"stream_load",
+ "key.converter":"org.apache.kafka.connect.json.JsonConverter",
+ "value.converter":"org.apache.kafka.connect.json.JsonConverter"
+ }
+}
\ No newline at end of file
diff --git a/src/test/resources/e2e/string_converter/full_types_debezium_ingestion.sql b/src/test/resources/e2e/string_converter/full_types_debezium_ingestion.sql
new file mode 100644
index 0000000..7cd001c
--- /dev/null
+++ b/src/test/resources/e2e/string_converter/full_types_debezium_ingestion.sql
@@ -0,0 +1,59 @@
+CREATE TABLE debezium_ingestion_msg.full_types_tab
+(
+ `id` LARGEINT NULL,
+ `tiny_c` TINYINT NULL,
+ `tiny_un_c` SMALLINT NULL,
+ `tiny_un_z_c` SMALLINT NULL,
+ `small_c` SMALLINT NULL,
+ `small_un_c` INT NULL,
+ `small_un_z_c` INT NULL,
+ `medium_c` INT NULL,
+ `medium_un_c` BIGINT NULL,
+ `medium_un_z_c` BIGINT NULL,
+ `int_c` INT NULL,
+ `int_un_c` BIGINT NULL,
+ `int_un_z_c` BIGINT NULL,
+ `int11_c` INT NULL,
+ `big_c` BIGINT NULL,
+ `big_un_c` LARGEINT NULL,
+ `big_un_z_c` LARGEINT NULL,
+ `varchar_c` VARCHAR(765) NULL,
+ `char_c` VARCHAR(9) NULL,
+ `real_c` DOUBLE NULL,
+ `float_c` FLOAT NULL,
+ `float_un_c` FLOAT NULL,
+ `float_un_z_c` FLOAT NULL,
+ `double_c` DOUBLE NULL,
+ `double_un_c` DOUBLE NULL,
+ `double_un_z_c` DOUBLE NULL,
+ `decimal_c` DECIMAL(8, 4) NULL,
+ `decimal_un_c` DECIMAL(8, 4) NULL,
+ `decimal_un_z_c` DECIMAL(8, 4) NULL,
+ `numeric_c` DECIMAL(6, 0) NULL,
+ `big_decimal_c` TEXT NULL,
+ `bit1_c` BOOLEAN NULL,
+ `tiny1_c` BOOLEAN NULL,
+ `boolean_c` BOOLEAN NULL,
+ `date_c` DATE NULL,
+ `time_c` TEXT NULL,
+ `datetime_c` DATETIME NULL,
+ `timestamp_c` DATETIME NULL,
+ `text_c` TEXT NULL,
+ `year_c` INT NULL,
+ `enum_c` TEXT NULL,
+ `set_c` TEXT NULL,
+ `json_c` JSON NULL,
+ `point_c` TEXT NULL,
+ `geometry_c` TEXT NULL,
+ `linestring_c` TEXT NULL,
+ `polygon_c` TEXT NULL,
+ `multipoint_c` TEXT NULL,
+ `multiline_c` TEXT NULL,
+ `multipolygon_c` TEXT NULL,
+ `geometrycollection_c` TEXT NULL
+)UNIQUE KEY(`id`)
+DISTRIBUTED BY HASH(`id`) BUCKETS 1
+PROPERTIES (
+"replication_allocation" = "tag.location.default: 1",
+"light_schema_change" = "true"
+);
\ No newline at end of file
diff --git a/src/test/resources/e2e/string_converter/insert_partial_update_tab.sql b/src/test/resources/e2e/string_converter/insert_partial_update_tab.sql
new file mode 100644
index 0000000..3dff4dd
--- /dev/null
+++ b/src/test/resources/e2e/string_converter/insert_partial_update_tab.sql
@@ -0,0 +1,3 @@
+insert into string_msg.partial_update_tab (id, col1, col2, col3)
+values (1, "before_update_col1_1", "before_update_col2_1", "before_update_col3_1"),
+ (2, "before_update_col1_2", "before_update_col2_2", "before_update_col3_2");
\ No newline at end of file
diff --git a/src/test/resources/e2e/string_converter/partial_update.json b/src/test/resources/e2e/string_converter/partial_update.json
new file mode 100644
index 0000000..0caaa4b
--- /dev/null
+++ b/src/test/resources/e2e/string_converter/partial_update.json
@@ -0,0 +1,24 @@
+{
+ "name":"partial_update_connector",
+ "config":{
+ "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
+ "topics":"partial_update_test",
+ "tasks.max":"1",
+ "doris.topic2table.map": "partial_update_test:partial_update_tab",
+ "buffer.count.records":"2",
+ "buffer.flush.time":"10",
+ "buffer.size.bytes":"10000000",
+ "doris.urls":"127.0.0.1",
+ "doris.user":"root",
+ "doris.password":"",
+ "doris.http.port":"8030",
+ "doris.query.port":"9030",
+ "doris.database":"string_msg",
+ "sink.properties.partial_columns":"true",
+ "sink.properties.columns": "id,col1,col2",
+ "enable.2pc": "false",
+ "load.model":"stream_load",
+ "key.converter":"org.apache.kafka.connect.storage.StringConverter",
+ "value.converter":"org.apache.kafka.connect.storage.StringConverter"
+ }
+}
\ No newline at end of file
diff --git a/src/test/resources/e2e/string_converter/partial_update_tab.sql b/src/test/resources/e2e/string_converter/partial_update_tab.sql
new file mode 100644
index 0000000..64d3d5b
--- /dev/null
+++ b/src/test/resources/e2e/string_converter/partial_update_tab.sql
@@ -0,0 +1,15 @@
+-- Please note that the database here should be consistent with doris.database in the file where the connector is registered.
+CREATE TABLE string_msg.partial_update_tab (
+ id INT NULL,
+ col1 VARCHAR(20) NULL,
+ col2 varchar(20) NULL,
+ col3 varchar(20) NUll
+) ENGINE=OLAP
+UNIQUE KEY(`id`)
+COMMENT 'OLAP'
+DISTRIBUTED BY HASH(`id`) BUCKETS AUTO
+PROPERTIES (
+"replication_allocation" = "tag.location.default: 1",
+"light_schema_change"="true",
+"enable_unique_key_merge_on_write" = "true"
+);
\ No newline at end of file
