BugFix: Fix string array type write failure. (#117)

oceanbase · Jan 7, 2025 · 9caf977 · 9caf977
1 parent 0cdb668
commit 9caf977
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 13 deletions.
diff --git a/...rc/main/java/com/oceanbase/connector/flink/table/OceanBaseRowDataSerializationSchema.java b/...rc/main/java/com/oceanbase/connector/flink/table/OceanBaseRowDataSerializationSchema.java
@@ -20,6 +20,7 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.data.binary.BinaryArrayData;
+import org.apache.flink.table.data.binary.BinaryStringData;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.LogicalTypeRoot;
@@ -146,7 +147,9 @@ private String parseArrayData(BinaryArrayData arrayData, LogicalType type) {
                     data.stream()
                             .map(
                                     element -> {
-                                        if (element instanceof Integer) {
+                                        if (element instanceof BinaryStringData) {
+                                            return String.format("\"%s\"", element);
+                                        } else if (element instanceof Integer) {
                                             return element.toString();
                                         } else if (element instanceof Boolean) {
                                             return element.toString();

diff --git a/...-oceanbase/src/test/java/com/oceanbase/connector/flink/OceanBaseMySQLConnectorITCase.java b/...-oceanbase/src/test/java/com/oceanbase/connector/flink/OceanBaseMySQLConnectorITCase.java
@@ -403,7 +403,8 @@ public void testSinkArrayType() throws Exception {
                         + " arr2 ARRAY<ARRAY<boolean>>,"
                         + " arr3 ARRAY<ARRAY<ARRAY<float>>>,"
                         + " arr4 ARRAY<ARRAY<ARRAY<ARRAY<double>>>>,"
-                        + " arr5 ARRAY<ARRAY<ARRAY<ARRAY<ARRAY<bigint>>>>>"
+                        + " arr5 ARRAY<ARRAY<ARRAY<ARRAY<ARRAY<bigint>>>>>,"
+                        + " arr6 ARRAY<ARRAY<ARRAY<varchar(225)>>>"
                         + ") with ("
                         + "  'connector'='oceanbase',"
                         + "  'table-name'='test_array',"
@@ -412,20 +413,20 @@ public void testSinkArrayType() throws Exception {
 
         tEnv.executeSql(
                         "INSERT INTO target "
-                                + "VALUES (101, ARRAY[1], ARRAY[ARRAY[true],ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]]),"
-                                + "       (101, ARRAY[1], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45],ARRAY[3.42]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]]),"
-                                + "       (101, ARRAY[1], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]],ARRAY[ARRAY[343343]]]]]),"
-                                + "       (101, ARRAY[1,2,3,4,5,6,7,8], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]]),"
-                                + "       (101, ARRAY[1], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.2,3.1],ARRAY[3.0]],ARRAY[ARRAY[2.9, 2.8, 2.67],ARRAY[2.54]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]]);")
+                                + "VALUES (101, ARRAY[1], ARRAY[ARRAY[true],ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]], ARRAY[ARRAY[ARRAY['xxx']]]),"
+                                + "       (102, ARRAY[1], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45],ARRAY[3.42]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]],ARRAY[ARRAY[ARRAY['生态']]]),"
+                                + "       (103, ARRAY[1], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]],ARRAY[ARRAY[343343]]]]],ARRAY[ARRAY[ARRAY['zz']]]),"
+                                + "       (104, ARRAY[1,2,3,4,5,6,7,8], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.4]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]],ARRAY[ARRAY[ARRAY['yyy']]]),"
+                                + "       (105, ARRAY[1], ARRAY[ARRAY[true, false]], ARRAY[ARRAY[ARRAY[3.2,3.1],ARRAY[3.0]],ARRAY[ARRAY[2.9, 2.8, 2.67],ARRAY[2.54]]], ARRAY[ARRAY[ARRAY[ARRAY[3.45]]]], ARRAY[ARRAY[ARRAY[ARRAY[ARRAY[12121]]]]],ARRAY[ARRAY[ARRAY['xxx', 'yyy']]]);")
                 .await();
 
         List<String> expected =
                 Arrays.asList(
-                        "101,[1],[[1],[1,0]],[[[3.4]]],[[[[3.45]]]],[[[[[12121]]]]]",
-                        "101,[1],[[1,0]],[[[3.4]]],[[[[3.45],[3.42]]]],[[[[[12121]]]]]",
-                        "101,[1],[[1,0]],[[[3.4]]],[[[[3.45]]]],[[[[[12121]],[[343343]]]]]",
-                        "101,[1,2,3,4,5,6,7,8],[[1,0]],[[[3.4]]],[[[[3.45]]]],[[[[[12121]]]]]",
-                        "101,[1],[[1,0]],[[[3.2,3.1],[3]],[[2.9,2.8,2.67],[2.54]]],[[[[3.45]]]],[[[[[12121]]]]]");
+                        "101,[1],[[1],[1,0]],[[[3.4]]],[[[[3.45]]]],[[[[[12121]]]]],[[[\"xxx\"]]]",
+                        "102,[1],[[1,0]],[[[3.4]]],[[[[3.45],[3.42]]]],[[[[[12121]]]]],[[[\"生态\"]]]",
+                        "103,[1],[[1,0]],[[[3.4]]],[[[[3.45]]]],[[[[[12121]],[[343343]]]]],[[[\"zz\"]]]",
+                        "104,[1,2,3,4,5,6,7,8],[[1,0]],[[[3.4]]],[[[[3.45]]]],[[[[[12121]]]]],[[[\"yyy\"]]]",
+                        "105,[1],[[1,0]],[[[3.2,3.1],[3]],[[2.9,2.8,2.67],[2.54]]],[[[[3.45]]]],[[[[[12121]]]]],[[[\"xxx\",\"yyy\"]]]");
 
         waitingAndAssertTableCount(testTable, expected.size());
 
@@ -437,6 +438,7 @@ public void testSinkArrayType() throws Exception {
     }
 
     // Only for Array data-type of OB
+    @Override
     protected String getRowString(ResultSet rs, int columnCount) throws SQLException {
         StringBuilder sb = new StringBuilder();
         for (int i = 0; i < columnCount; i++) {

diff --git a/flink-connector-oceanbase/src/test/resources/sql/mysql/array_type.sql b/flink-connector-oceanbase/src/test/resources/sql/mysql/array_type.sql
@@ -18,5 +18,6 @@ CREATE TABLE test_array
   arr2 bool[][],
   arr3 float[][][],
   arr4 double[][][][],
-  arr5 bigint[][][][][]
+  arr5 bigint[][][][][],
+  arr6 varchar(225)[][][]
 );