fuweng11 commented on code in PR #7689:
URL: https://github.com/apache/inlong/pull/7689#discussion_r1150038806
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java:
##########
@@ -677,24 +692,64 @@ public Boolean updateAfterApprove(List<SinkApproveDTO>
approveList, String opera
}
@Override
- public List<SinkField> parseFields(String fieldsJson) {
+ public List<SinkField> parseFields(ParseFieldRequest parseFieldRequest) {
try {
- Map<String, String> fieldsMap = objectMapper.readValue(fieldsJson,
- new TypeReference<Map<String, String>>() {
- });
- return fieldsMap.keySet().stream().map(fieldName -> {
+ String method = parseFieldRequest.getMethod();
+ String statement = parseFieldRequest.getStatement();
+
+ Map<String, String> fieldsMap;
+ if (STATEMENT_TYPE_JSON.equals(method)) {
+ fieldsMap = parseFieldsByJson(statement);
+ } else {
+ fieldsMap = parseFieldsBySql(statement);
+ }
+ return fieldsMap.entrySet().stream().map(entry -> {
SinkField field = new SinkField();
- field.setFieldName(fieldName);
- field.setFieldType(fieldsMap.get(fieldName));
+ field.setFieldName(entry.getKey());
+ field.setFieldType(entry.getValue());
return field;
}).collect(Collectors.toList());
+
} catch (Exception e) {
LOGGER.error("parse sink fields error", e);
throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
String.format("parse sink fields error : %s",
e.getMessage()));
}
}
+ private Map<String, String> parseFieldsBySql(String sql) throws
JSQLParserException {
+ CCJSqlParserManager pm = new CCJSqlParserManager();
+ Statement statement = pm.parse(new StringReader(sql));
+ LinkedHashMap<String, String> fields = new LinkedHashMap<>();
+ if (statement instanceof CreateTable) {
+ CreateTable createTable = (CreateTable) statement;
+ List<ColumnDefinition> columnDefinitions =
createTable.getColumnDefinitions();
+ // get column definition
+ for (ColumnDefinition definition : columnDefinitions) {
+ // get field name
+ String columnName = definition.getColumnName();
+ ColDataType colDataType = definition.getColDataType();
+ String sqlDataType = colDataType.getDataType();
+ // Convert SQL type to TypeInfo
+ FormatInfo formatInfo =
FieldInfoUtils.convertFieldFormat(sqlDataType, "");
Review Comment:
Is this field conversion valid？As far as I know
`FieldInfoUtils.convertFieldFormat` method is turning sourceFieldType into
sinkFIeldType. However, the field type in the table creation statement you
passed in belongs to the `sinkFieldType`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
