This is an automated email from the ASF dual-hosted git repository.
zongwen pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 9f59fc187 [Improve][SeaTunnel-Schema] Support parse row type from
config file (#2771)
9f59fc187 is described below
commit 9f59fc1874fec7909c37c207c9be7e88499e11e8
Author: TyrantLucifer <[email protected]>
AuthorDate: Fri Sep 23 10:36:20 2022 +0800
[Improve][SeaTunnel-Schema] Support parse row type from config file (#2771)
* [Hotfix][SeaTunnel-Schema] Fix the bug that can not parse decimal
* [Improve][SeaTunnel-Schema] Support parse row type from config file
---
.../seatunnel/common/schema/SeaTunnelSchema.java | 54 ++++++++++++++++------
.../connector/common/schema/SchemaParseTest.java | 2 +
.../src/test/resources/complex.schema.conf | 38 +++++++++++++++
3 files changed, 81 insertions(+), 13 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/schema/SeaTunnelSchema.java
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/schema/SeaTunnelSchema.java
index 9a9e035fc..144a02c24 100644
---
a/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/schema/SeaTunnelSchema.java
+++
b/seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/schema/SeaTunnelSchema.java
@@ -33,7 +33,12 @@ import org.apache.seatunnel.common.utils.JsonUtils;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
import org.apache.seatunnel.shade.com.typesafe.config.ConfigRenderOptions;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.JsonNodeType;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
import java.io.Serializable;
+import java.util.LinkedHashMap;
import java.util.Map;
public class SeaTunnelSchema implements Serializable {
@@ -108,6 +113,11 @@ public class SeaTunnelSchema implements Serializable {
String valueGenericType = "";
// convert type to uppercase
type = type.toUpperCase();
+ String originContent = type;
+ if (type.contains("{") || type.contains("}")) {
+ // Row type
+ type = SqlType.ROW.name();
+ }
if (type.contains("<") || type.contains(">")) {
// Map type or Array type
if (type.startsWith(SqlType.MAP.name())) {
@@ -188,26 +198,34 @@ public class SeaTunnelSchema implements Serializable {
case TIMESTAMP:
return LocalTimeType.LOCAL_DATE_TIME_TYPE;
default:
- throw new RuntimeException("Not support [row] type now");
+ return mapToSeaTunnelRowType(convertJsonToMap(originContent));
}
}
- private static Map<String, String> convertConfig2Map(Config config) {
+ private static Map<String, String> convertConfigToMap(Config config) {
// Because the entrySet in typesafe config couldn't keep key-value
order
// So use jackson parsing schema information into a map to keep
key-value order
ConfigRenderOptions options = ConfigRenderOptions.concise();
String schema = config.root().render(options);
- return JsonUtils.toMap(schema);
+ return convertJsonToMap(schema);
}
- public static SeaTunnelSchema buildWithConfig(Config schemaConfig) {
- CheckResult checkResult = CheckConfigUtil.checkAllExists(schemaConfig,
FIELD_KEY);
- if (!checkResult.isSuccess()) {
- String errorMsg = String.format("Schema config need option [%s],
please correct your config first", FIELD_KEY);
- throw new RuntimeException(errorMsg);
- }
- Config fields = schemaConfig.getConfig(FIELD_KEY);
- Map<String, String> fieldsMap = convertConfig2Map(fields);
+ private static Map<String, String> convertJsonToMap(String json) {
+ ObjectNode jsonNodes = JsonUtils.parseObject(json);
+ LinkedHashMap<String, String> fieldsMap = new LinkedHashMap<>();
+ jsonNodes.fields().forEachRemaining(field -> {
+ String key = field.getKey();
+ JsonNode value = field.getValue();
+ if (value.getNodeType() == JsonNodeType.OBJECT) {
+ fieldsMap.put(key, value.toString());
+ } else {
+ fieldsMap.put(key, value.textValue());
+ }
+ });
+ return fieldsMap;
+ }
+
+ private static SeaTunnelRowType mapToSeaTunnelRowType(Map<String, String>
fieldsMap) {
int fieldsNum = fieldsMap.size();
int i = 0;
String[] fieldsName = new String[fieldsNum];
@@ -220,8 +238,18 @@ public class SeaTunnelSchema implements Serializable {
seaTunnelDataTypes[i] = dataType;
i++;
}
- SeaTunnelRowType seaTunnelRowType = new SeaTunnelRowType(fieldsName,
seaTunnelDataTypes);
- return new SeaTunnelSchema(seaTunnelRowType);
+ return new SeaTunnelRowType(fieldsName, seaTunnelDataTypes);
+ }
+
+ public static SeaTunnelSchema buildWithConfig(Config schemaConfig) {
+ CheckResult checkResult = CheckConfigUtil.checkAllExists(schemaConfig,
FIELD_KEY);
+ if (!checkResult.isSuccess()) {
+ String errorMsg = String.format("Schema config need option [%s],
please correct your config first", FIELD_KEY);
+ throw new RuntimeException(errorMsg);
+ }
+ Config fields = schemaConfig.getConfig(FIELD_KEY);
+ Map<String, String> fieldsMap = convertConfigToMap(fields);
+ return new SeaTunnelSchema(mapToSeaTunnelRowType(fieldsMap));
}
public static SeaTunnelRowType buildSimpleTextSchema() {
diff --git
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connector/common/schema/SchemaParseTest.java
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connector/common/schema/SchemaParseTest.java
index 1baad8f05..c41339245 100644
---
a/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connector/common/schema/SchemaParseTest.java
+++
b/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connector/common/schema/SchemaParseTest.java
@@ -23,6 +23,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.SeaTunnelRowType;
+import org.apache.seatunnel.api.table.type.SqlType;
import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
@@ -73,6 +74,7 @@ public class SchemaParseTest {
new MapType<>(BasicType.STRING_TYPE, new
MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE)));
Assertions.assertEquals(seaTunnelRowType.getFieldType(1),
new MapType<>(BasicType.STRING_TYPE, new
MapType<>(BasicType.STRING_TYPE, ArrayType.INT_ARRAY_TYPE)));
+
Assertions.assertEquals(seaTunnelRowType.getFieldType(17).getSqlType(),
SqlType.ROW);
}
public static String getTestConfigFile(String configFile) throws
FileNotFoundException, URISyntaxException {
diff --git
a/seatunnel-connectors-v2/connector-common/src/test/resources/complex.schema.conf
b/seatunnel-connectors-v2/connector-common/src/test/resources/complex.schema.conf
index 6a06dbf06..d71d87c5c 100644
---
a/seatunnel-connectors-v2/connector-common/src/test/resources/complex.schema.conf
+++
b/seatunnel-connectors-v2/connector-common/src/test/resources/complex.schema.conf
@@ -34,5 +34,43 @@ schema {
date = date
time = time
timestamp = timestamp
+ row = {
+ map = "map<string, map<string, string>>"
+ map_array = "map<string, map<string, array<int>>>"
+ array = "array<tinyint>"
+ string = string
+ boolean = boolean
+ tinyint = tinyint
+ smallint = smallint
+ int = int
+ bigint = bigint
+ float = float
+ double = double
+ decimal = "decimal(30, 8)"
+ null = "null"
+ bytes = bytes
+ date = date
+ time = time
+ timestamp = timestamp
+ row = {
+ map = "map<string, map<string, string>>"
+ map_array = "map<string, map<string, array<int>>>"
+ array = "array<tinyint>"
+ string = string
+ boolean = boolean
+ tinyint = tinyint
+ smallint = smallint
+ int = int
+ bigint = bigint
+ float = float
+ double = double
+ decimal = "decimal(30, 8)"
+ null = "null"
+ bytes = bytes
+ date = date
+ time = time
+ timestamp = timestamp
+ }
+ }
}
}
\ No newline at end of file
