[Feature][API] Support hocon style declare row type in generic type (apache#6187)

CheneyYin · web-flow · commit a9bc5ca58ddc · 2024-01-17T22:55:31.000+08:00
diff --git a/docs/en/concept/schema-feature.md b/docs/en/concept/schema-feature.md
diff --git a/release-note.md b/release-note.md
@@ -165,6 +165,7 @@
 - [Core] [Shade] Add guava shade module (#4358)
 - [Core] [Spark] Support SeaTunnel Time Type (#5188)
 - [Core] [Flink] Support Decimal Type with configurable precision and scale (#5419)
+- [Core] [API] Support hocon style declare row type in generic type (#6187)
 
 ### Connector-V2
 
diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/SeaTunnelDataTypeConvertorUtil.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/SeaTunnelDataTypeConvertorUtil.java
@@ -17,7 +17,10 @@
 
 package org.apache.seatunnel.api.table.catalog;
 
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigObject;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigValue;
 
 import org.apache.seatunnel.api.table.type.ArrayType;
 import org.apache.seatunnel.api.table.type.BasicType;
@@ -29,9 +32,6 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.api.table.type.SqlType;
 import org.apache.seatunnel.common.exception.CommonError;
-import org.apache.seatunnel.common.utils.JsonUtils;
-
-import java.util.Map;
 
 public class SeaTunnelDataTypeConvertorUtil {
 
@@ -87,13 +87,13 @@ public static SeaTunnelDataType<?> deserializeSeaTunnelDataType(
     private static SeaTunnelDataType<?> parseComplexDataType(String field, String columnStr) {
         String column = columnStr.toUpperCase().replace(" ", "");
         if (column.startsWith(SqlType.MAP.name())) {
-            return parseMapType(field, column);
+            return parseMapType(field, columnStr);
         }
         if (column.startsWith(SqlType.ARRAY.name())) {
-            return parseArrayType(field, column);
+            return parseArrayType(field, columnStr);
         }
         if (column.startsWith(SqlType.DECIMAL.name())) {
-            return parseDecimalType(column);
+            return parseDecimalType(columnStr);
         }
         if (column.trim().startsWith("{")) {
             return parseRowType(columnStr);
@@ -102,31 +102,64 @@ private static SeaTunnelDataType<?> parseComplexDataType(String field, String co
     }
 
     private static SeaTunnelDataType<?> parseRowType(String columnStr) {
-        ObjectNode jsonNodes = JsonUtils.parseObject(columnStr);
-        Map<String, String> fieldsMap = JsonUtils.toStringMap(jsonNodes);
-        String[] fieldsName = new String[fieldsMap.size()];
-        SeaTunnelDataType<?>[] seaTunnelDataTypes = new SeaTunnelDataType<?>[fieldsMap.size()];
-        int i = 0;
-        for (Map.Entry<String, String> entry : fieldsMap.entrySet()) {
-            fieldsName[i] = entry.getKey();
-            seaTunnelDataTypes[i] = deserializeSeaTunnelDataType(entry.getKey(), entry.getValue());
-            i++;
+        String confPayload = "{conf = " + columnStr + "}";
+        Config conf;
+        try {
+            conf = ConfigFactory.parseString(confPayload);
+        } catch (RuntimeException e) {
+            throw new IllegalArgumentException(
+                    String.format("HOCON Config parse from %s failed.", confPayload), e);
+        }
+        return parseRowType(conf.getObject("conf"));
+    }
+
+    private static SeaTunnelDataType<?> parseRowType(ConfigObject conf) {
+        String[] fieldNames = new String[conf.size()];
+        SeaTunnelDataType<?>[] fieldTypes = new SeaTunnelDataType[conf.size()];
+        conf.keySet().toArray(fieldNames);
+
+        for (int idx = 0; idx < fieldNames.length; idx++) {
+            String fieldName = fieldNames[idx];
+            ConfigValue typeVal = conf.get(fieldName);
+            switch (typeVal.valueType()) {
+                case STRING:
+                    {
+                        fieldTypes[idx] =
+                                deserializeSeaTunnelDataType(
+                                        fieldNames[idx], (String) typeVal.unwrapped());
+                    }
+                    break;
+                case OBJECT:
+                    {
+                        fieldTypes[idx] = parseRowType((ConfigObject) typeVal);
+                    }
+                    break;
+                case LIST:
+                case NUMBER:
+                case BOOLEAN:
+                case NULL:
+                default:
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "Unsupported parse SeaTunnel Type from '%s'.",
+                                    typeVal.unwrapped()));
+            }
         }
-        return new SeaTunnelRowType(fieldsName, seaTunnelDataTypes);
+        return new SeaTunnelRowType(fieldNames, fieldTypes);
     }
 
     private static SeaTunnelDataType<?> parseMapType(String field, String columnStr) {
-        String genericType = getGenericType(columnStr);
+        String genericType = getGenericType(columnStr).trim();
         int index =
-                genericType.startsWith(SqlType.DECIMAL.name())
+                genericType.toUpperCase().startsWith(SqlType.DECIMAL.name())
                         ?
                         // if map key is decimal, we should find the index of second ','
                         genericType.indexOf(",", genericType.indexOf(",") + 1)
                         :
                         // if map key is not decimal, we should find the index of first ','
                         genericType.indexOf(",");
-        String keyGenericType = genericType.substring(0, index);
-        String valueGenericType = genericType.substring(index + 1);
+        String keyGenericType = genericType.substring(0, index).trim();
+        String valueGenericType = genericType.substring(index + 1).trim();
         return new MapType<>(
                 deserializeSeaTunnelDataType(field, keyGenericType),
                 deserializeSeaTunnelDataType(field, valueGenericType));
@@ -138,7 +171,7 @@ private static String getGenericType(String columnStr) {
     }
 
     private static SeaTunnelDataType<?> parseArrayType(String field, String columnStr) {
-        String genericType = getGenericType(columnStr);
+        String genericType = getGenericType(columnStr).trim();
         SeaTunnelDataType<?> dataType = deserializeSeaTunnelDataType(field, genericType);
         switch (dataType.getSqlType()) {
             case STRING:
@@ -158,7 +191,7 @@ private static SeaTunnelDataType<?> parseArrayType(String field, String columnSt
             case DOUBLE:
                 return ArrayType.DOUBLE_ARRAY_TYPE;
             default:
-                throw CommonError.unsupportedDataType("SeaTunnel", columnStr, field);
+                throw CommonError.unsupportedDataType("SeaTunnel", genericType, field);
         }
     }
 
diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtilTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtilTest.java
@@ -27,6 +27,7 @@
 import org.apache.seatunnel.api.table.type.DecimalType;
 import org.apache.seatunnel.api.table.type.MapType;
 import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.api.table.type.SqlType;
 import org.apache.seatunnel.common.utils.SeaTunnelException;
@@ -137,6 +138,42 @@ public void testCatalogUtilGetCatalogTable() throws FileNotFoundException, URISy
                                 Thread.currentThread().getContextClassLoader()));
     }
 
+    @Test
+    public void testGenericRowSchemaTest() throws FileNotFoundException, URISyntaxException {
+        String path = getTestConfigFile("/conf/generic_row.schema.conf");
+        Config config = ConfigFactory.parseFile(new File(path));
+        SeaTunnelRowType seaTunnelRowType =
+                CatalogTableUtil.buildWithConfig(config).getSeaTunnelRowType();
+        Assertions.assertNotNull(seaTunnelRowType);
+        Assertions.assertArrayEquals(
+                new String[] {"map0", "map1"}, seaTunnelRowType.getFieldNames());
+
+        MapType<String, SeaTunnelRowType> mapType0 =
+                (MapType<String, SeaTunnelRowType>) seaTunnelRowType.getFieldType(0);
+        MapType<String, SeaTunnelRowType> mapType1 =
+                (MapType<String, SeaTunnelRowType>) seaTunnelRowType.getFieldType(1);
+        Assertions.assertNotNull(mapType0);
+        Assertions.assertNotNull(mapType1);
+        Assertions.assertEquals(BasicType.STRING_TYPE, mapType0.getKeyType());
+
+        SeaTunnelRowType expectedVal =
+                new SeaTunnelRowType(
+                        new String[] {"c_int", "c_string", "c_row"},
+                        new SeaTunnelDataType[] {
+                            BasicType.INT_TYPE,
+                            BasicType.STRING_TYPE,
+                            new SeaTunnelRowType(
+                                    new String[] {"c_int"},
+                                    new SeaTunnelDataType[] {BasicType.INT_TYPE})
+                        });
+        SeaTunnelRowType mapType0ValType =
+                (SeaTunnelRowType) ((SeaTunnelDataType<?>) mapType0.getValueType());
+        Assertions.assertEquals(expectedVal, mapType0ValType);
+        SeaTunnelRowType mapType1ValType =
+                (SeaTunnelRowType) ((SeaTunnelDataType<?>) mapType1.getValueType());
+        Assertions.assertEquals(expectedVal, mapType1ValType);
+    }
+
     public static String getTestConfigFile(String configFile)
             throws FileNotFoundException, URISyntaxException {
         URL resource = CatalogTableUtilTest.class.getResource(configFile);
diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/SeaTunnelDataTypeConvertorUtilTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/SeaTunnelDataTypeConvertorUtilTest.java
@@ -66,13 +66,27 @@ void testParseWithUnsupportedType() {
                 "ErrorCode:[COMMON-07], ErrorDescription:['SeaTunnel' unsupported data type 'uuid' of 'test']",
                 exception4.getMessage());
 
-        RuntimeException exception5 =
+        IllegalArgumentException exception5 =
                 Assertions.assertThrows(
-                        RuntimeException.class,
+                        IllegalArgumentException.class,
                         () ->
                                 SeaTunnelDataTypeConvertorUtil.deserializeSeaTunnelDataType(
                                         "test", "{uuid}"));
-        Assertions.assertEquals(
-                "String json deserialization exception.{uuid}", exception5.getMessage());
+        String expectedMsg5 =
+                String.format("HOCON Config parse from %s failed.", "{conf = {uuid}}");
+        Assertions.assertEquals(expectedMsg5, exception5.getMessage());
+
+        String invalidTypeDeclaration = "[e]";
+        IllegalArgumentException exception6 =
+                Assertions.assertThrows(
+                        IllegalArgumentException.class,
+                        () ->
+                                SeaTunnelDataTypeConvertorUtil.deserializeSeaTunnelDataType(
+                                        "test",
+                                        String.format("{c_0 = %s}", invalidTypeDeclaration)));
+        String expectedMsg6 =
+                String.format(
+                        "Unsupported parse SeaTunnel Type from '%s'.", invalidTypeDeclaration);
+        Assertions.assertEquals(expectedMsg6, exception6.getMessage());
     }
 }
diff --git a/seatunnel-api/src/test/resources/conf/generic_row.schema.conf b/seatunnel-api/src/test/resources/conf/generic_row.schema.conf
@@ -0,0 +1,25 @@
+#
+# 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.
+#
+
+schema {
+  fields {
+    # Hocon style declare row type in generic type
+    map0 = "map<string, {c_int = int, c_string = string, c_row = {c_int = int}}>"
+    # Json style declare row type in generic type
+    map1 = "map<string, {\"c_int\":\"int\", \"c_string\":\"string\", \"c_row\":{\"c_int\":\"int\"}}>"
+  }
+}
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/java/org/apache/seatunnel/e2e/connector/fake/FakeIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/java/org/apache/seatunnel/e2e/connector/fake/FakeIT.java
@@ -38,5 +38,8 @@ public void testFakeConnector(TestContainer container)
         Container.ExecResult fakeWithTemplate =
                 container.executeJob("/fake_to_assert_with_template.conf");
         Assertions.assertEquals(0, fakeWithTemplate.getExitCode());
+        Container.ExecResult fakeComplex =
+                container.executeJob("/fake_generic_row_type_to_assert.conf");
+        Assertions.assertEquals(0, fakeWithTemplate.getExitCode());
     }
 }
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_generic_row_type_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_generic_row_type_to_assert.conf
@@ -0,0 +1,66 @@
+#
+# 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.
+#
+
+env {
+  parallelism = 1
+  job.mode = BATCH
+  # checkpoint.interval = 10000
+}
+
+source {
+  FakeSource {
+    row.num = 1
+    schema = {
+      fields {
+        c_0 = "map<string, {c_int=int\nc_string=string}>"
+        c_1 = "map<string, {c_int=int,c_string=string}>"
+        c_2 = "map<string, {c_int=int,c_string=string,c_row={c_int=int}}>"
+        c_3 = "map<string, {\"c_int\":\"int\",\"c_string\":\"string\"}>"
+      }
+    }
+    result_table_name = "fake"
+  }
+}
+
+sink{
+  Assert {
+    source_table_name = "fake"
+    rules =
+      {
+        catalog_table_rule {
+          column_rule = [
+            {
+              name = "c_0"
+              type = "map<string, {c_int=int\nc_string=string}>"
+            }
+            {
+              name = "c_1"
+              type = "map<string, {c_int=int,c_string=string}>"
+            }
+            {
+              name = "c_2"
+              type = "map<string, {c_int=int,c_string=string,c_row={c_int=int}}>"
+            }
+            {
+              name = "c_3"
+              type = "map<string, {\"c_int\":\"int\",\"c_string\":\"string\"}>"
+            }
+          ]
+        }
+      }
+  }
+}

Original file line number	Diff line number	Diff line change
`@@ -38,5 +38,8 @@ public void testFakeConnector(TestContainer container)`
`38`	`38`	`Container.ExecResult fakeWithTemplate =`
`39`	`39`	`container.executeJob("/fake_to_assert_with_template.conf");`
`40`	`40`	`Assertions.assertEquals(0, fakeWithTemplate.getExitCode());`
	`41`	`+ Container.ExecResult fakeComplex =`
	`42`	`+ container.executeJob("/fake_generic_row_type_to_assert.conf");`
	`43`	`+ Assertions.assertEquals(0, fakeWithTemplate.getExitCode());`
`41`	`44`	`}`
`42`	`45`	`}`