This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 3da64218a [INLONG-3993][Manager] Support HiveLoadNode in lightweight
mode (#4009)
3da64218a is described below
commit 3da64218aa11cac4fc0195d31b227db1a2f5ff7a
Author: kipshi <[email protected]>
AuthorDate: Thu Apr 28 21:16:32 2022 +0800
[INLONG-3993][Manager] Support HiveLoadNode in lightweight mode (#4009)
---
.../inlong/manager/client/api/sink/HiveSink.java | 12 ++-
.../client/api/util/InlongStreamSinkTransfer.java | 12 ++-
.../api/util/InlongStreamSourceTransfer.java | 98 +---------------------
.../api/util/InlongStreamTransformTransfer.java | 7 +-
.../inlong/manager/common/enums/MetaFieldType.java | 37 +++++++-
.../manager/common/pojo/sink/hive/HiveSinkDTO.java | 8 ++
.../pojo/sink/hive/HiveSinkListResponse.java | 6 ++
.../common/pojo/sink/hive/HiveSinkRequest.java | 6 ++
.../common/pojo/sink/hive/HiveSinkResponse.java | 6 ++
.../manager/common/pojo/stream/StreamField.java | 10 +++
.../common/pojo/transform/TransformRequest.java | 4 +-
.../common/pojo/transform/TransformResponse.java | 4 +-
.../manager/common/util/StreamParseUtils.java | 88 ++++++++++++++++++-
.../dao/entity/StreamTransformFieldEntity.java | 2 +
.../mappers/StreamTransformFieldEntityMapper.xml | 24 ++++--
.../service/sort/util/ExtractNodeUtils.java | 15 ++--
.../manager/service/sort/util/FieldInfoUtils.java | 25 ++++++
.../service/sort/util/FieldRelationShipUtils.java | 75 +++++++++--------
.../manager/service/sort/util/LoadNodeUtils.java | 44 ++++++++++
.../service/sort/util/TransformNodeUtils.java | 19 +----
.../transform/StreamTransformServiceImpl.java | 17 ++--
.../main/resources/sql/apache_inlong_manager.sql | 5 +-
.../manager-web/sql/apache_inlong_manager.sql | 1 +
23 files changed, 343 insertions(+), 182 deletions(-)
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/HiveSink.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/HiveSink.java
index 80e7497e5..ba166ffd9 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/HiveSink.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/sink/HiveSink.java
@@ -23,14 +23,14 @@ import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
-import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.client.api.DataSeparator;
-import org.apache.inlong.manager.common.pojo.stream.SinkField;
-import org.apache.inlong.manager.common.pojo.stream.StreamSink;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
+import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.common.enums.FileFormat;
import org.apache.inlong.manager.common.enums.SinkType;
import org.apache.inlong.manager.common.pojo.sink.hive.HivePartitionField;
+import org.apache.inlong.manager.common.pojo.stream.SinkField;
+import org.apache.inlong.manager.common.pojo.stream.StreamSink;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
@@ -81,5 +81,11 @@ public class HiveSink extends StreamSink {
@ApiModelProperty("Data format type for stream sink")
private DataFormat dataFormat;
+
+ @ApiModelProperty("Version for hive")
+ private String hiveVersion;
+
+ @ApiModelProperty("Config directory of hive, needed by sort in light mode")
+ private String hiveConfDir;
}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
index fe052febc..7cf90d27b 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSinkTransfer.java
@@ -20,14 +20,12 @@ package org.apache.inlong.manager.client.api.util;
import com.google.common.collect.Lists;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.client.api.DataSeparator;
-import org.apache.inlong.manager.common.pojo.stream.SinkField;
-import org.apache.inlong.manager.common.pojo.stream.StreamSink;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
import org.apache.inlong.manager.client.api.sink.ClickHouseSink;
import org.apache.inlong.manager.client.api.sink.HiveSink;
import org.apache.inlong.manager.client.api.sink.KafkaSink;
+import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.common.enums.FieldType;
import org.apache.inlong.manager.common.enums.FileFormat;
import org.apache.inlong.manager.common.enums.GlobalConstants;
@@ -43,6 +41,8 @@ import
org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkResponse;
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkRequest;
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
+import org.apache.inlong.manager.common.pojo.stream.SinkField;
+import org.apache.inlong.manager.common.pojo.stream.StreamSink;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import java.nio.charset.Charset;
@@ -223,6 +223,8 @@ public class InlongStreamSinkTransfer {
hiveSinkRequest.setFileFormat(hiveSink.getFileFormat().name());
hiveSinkRequest.setSinkType(hiveSink.getSinkType().name());
hiveSinkRequest.setPartitionFieldList(hiveSink.getPartitionFieldList());
+ hiveSinkRequest.setHiveConfDir(hiveSink.getHiveConfDir());
+ hiveSinkRequest.setHiveVersion(hiveSink.getHiveVersion());
DefaultAuthentication defaultAuthentication =
hiveSink.getAuthentication();
AssertUtil.notNull(defaultAuthentication,
String.format("Hive storage:%s must be authenticated",
hiveSink.getDbName()));
@@ -268,6 +270,8 @@ public class InlongStreamSinkTransfer {
hiveSink.setTableName(snapshot.getTableName());
hiveSink.setDbName(snapshot.getDbName());
hiveSink.setDataPath(snapshot.getDataPath());
+ hiveSink.setHiveVersion(snapshot.getHiveVersion());
+ hiveSink.setHiveConfDir(snapshot.getHiveConfDir());
hiveSink.setPartitionFieldList(snapshot.getPartitionFieldList());
} else {
hiveSink.setSinkName(sinkResponse.getSinkName());
@@ -281,6 +285,8 @@ public class InlongStreamSinkTransfer {
hiveSink.setTableName(sinkResponse.getTableName());
hiveSink.setDbName(sinkResponse.getDbName());
hiveSink.setDataPath(sinkResponse.getDataPath());
+ hiveSink.setHiveConfDir(sinkResponse.getHiveConfDir());
+ hiveSink.setHiveVersion(sinkResponse.getHiveVersion());
hiveSink.setPartitionFieldList(sinkResponse.getPartitionFieldList());
}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
index dc7847fe8..af9ce5b32 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamSourceTransfer.java
@@ -20,33 +20,27 @@ package org.apache.inlong.manager.client.api.util;
import com.google.common.base.Joiner;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.inlong.manager.common.enums.DataFormat;
-import org.apache.inlong.manager.common.pojo.source.kafka.KafkaOffset;
-import org.apache.inlong.manager.common.pojo.stream.StreamSource;
-import org.apache.inlong.manager.common.pojo.stream.StreamSource.State;
-import org.apache.inlong.manager.common.pojo.stream.StreamSource.SyncType;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
import org.apache.inlong.manager.client.api.source.AgentFileSource;
import org.apache.inlong.manager.client.api.source.AutoPushSource;
import org.apache.inlong.manager.client.api.source.KafkaSource;
import org.apache.inlong.manager.client.api.source.MySQLBinlogSource;
+import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.common.enums.SourceType;
-import org.apache.inlong.manager.common.pojo.source.SourceListResponse;
import org.apache.inlong.manager.common.pojo.source.SourceRequest;
import org.apache.inlong.manager.common.pojo.source.SourceResponse;
-import
org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceListResponse;
import
org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceRequest;
import
org.apache.inlong.manager.common.pojo.source.autopush.AutoPushSourceResponse;
-import
org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceRequest;
import
org.apache.inlong.manager.common.pojo.source.binlog.BinlogSourceResponse;
-import
org.apache.inlong.manager.common.pojo.source.file.FileSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.file.FileSourceRequest;
import org.apache.inlong.manager.common.pojo.source.file.FileSourceResponse;
-import
org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceListResponse;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceRequest;
import org.apache.inlong.manager.common.pojo.source.kafka.KafkaSourceResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
+import org.apache.inlong.manager.common.pojo.stream.StreamSource;
+import org.apache.inlong.manager.common.pojo.stream.StreamSource.State;
+import org.apache.inlong.manager.common.pojo.stream.StreamSource.SyncType;
import java.util.Arrays;
@@ -89,24 +83,6 @@ public class InlongStreamSourceTransfer {
throw new IllegalArgumentException(String.format("Unsupported source
type : %s for Inlong", sourceType));
}
- public static StreamSource parseStreamSource(SourceListResponse
sourceListResponse) {
- String type = sourceListResponse.getSourceType();
- SourceType sourceType = SourceType.forType(type);
- if (sourceType == SourceType.KAFKA && sourceListResponse instanceof
KafkaSourceListResponse) {
- return parseKafkaSource((KafkaSourceListResponse)
sourceListResponse);
- }
- if (sourceType == SourceType.BINLOG && sourceListResponse instanceof
BinlogSourceListResponse) {
- return parseMySQLBinlogSource((BinlogSourceListResponse)
sourceListResponse);
- }
- if (sourceType == SourceType.FILE && sourceListResponse instanceof
FileSourceListResponse) {
- return parseAgentFileSource((FileSourceListResponse)
sourceListResponse);
- }
- if (sourceType == SourceType.AUTO_PUSH && sourceListResponse
instanceof AutoPushSourceListResponse) {
- return parseAutoPushSource((AutoPushSourceListResponse)
sourceListResponse);
- }
- throw new IllegalArgumentException(String.format("Unsupported source
type : %s for Inlong", sourceType));
- }
-
private static KafkaSource parseKafkaSource(KafkaSourceResponse response) {
KafkaSource kafkaSource = new KafkaSource();
kafkaSource.setSourceName(response.getSourceName());
@@ -129,25 +105,6 @@ public class InlongStreamSourceTransfer {
return kafkaSource;
}
- private static KafkaSource parseKafkaSource(KafkaSourceListResponse
response) {
- KafkaSource kafkaSource = new KafkaSource();
- kafkaSource.setSourceName(response.getSourceName());
- kafkaSource.setConsumerGroup(response.getGroupId());
- kafkaSource.setState(State.parseByStatus(response.getStatus()));
- DataFormat dataFormat =
DataFormat.forName(response.getSerializationType());
- kafkaSource.setDataFormat(dataFormat);
- kafkaSource.setTopic(response.getTopic());
- kafkaSource.setBootstrapServers(response.getBootstrapServers());
- kafkaSource.setByteSpeedLimit(response.getByteSpeedLimit());
-
kafkaSource.setTopicPartitionOffset(response.getTopicPartitionOffset());
-
- KafkaOffset offset =
KafkaOffset.forName(response.getAutoOffsetReset());
- kafkaSource.setAutoOffsetReset(offset);
- kafkaSource.setRecordSpeedLimit(response.getRecordSpeedLimit());
- kafkaSource.setSyncType(SyncType.FULL);
- return kafkaSource;
- }
-
private static MySQLBinlogSource
parseMySQLBinlogSource(BinlogSourceResponse response) {
MySQLBinlogSource binlogSource = new MySQLBinlogSource();
binlogSource.setSourceName(response.getSourceName());
@@ -177,33 +134,6 @@ public class InlongStreamSourceTransfer {
return binlogSource;
}
- private static MySQLBinlogSource
parseMySQLBinlogSource(BinlogSourceListResponse response) {
- MySQLBinlogSource binlogSource = new MySQLBinlogSource();
- binlogSource.setSourceName(response.getSourceName());
- binlogSource.setHostname(response.getHostname());
- binlogSource.setDataFormat(DataFormat.NONE);
- binlogSource.setPort(response.getPort());
- binlogSource.setState(State.parseByStatus(response.getStatus()));
- binlogSource.setServerId(response.getServerId());
- DefaultAuthentication defaultAuthentication = new
DefaultAuthentication(
- response.getUser(),
- response.getPassword());
- binlogSource.setAuthentication(defaultAuthentication);
- binlogSource.setIncludeSchema(response.getIncludeSchema());
- binlogSource.setServerTimezone(response.getServerTimezone());
- binlogSource.setMonitoredDdl(response.getMonitoredDdl());
-
binlogSource.setTimestampFormatStandard(response.getTimestampFormatStandard());
- binlogSource.setAllMigration(response.isAllMigration());
- binlogSource.setPrimaryKey(response.getPrimaryKey());
- if (StringUtils.isNotBlank(response.getDatabaseWhiteList())) {
-
binlogSource.setDbNames(Arrays.asList(response.getDatabaseWhiteList().split(",")));
- }
- if (StringUtils.isNotBlank(response.getTableWhiteList())) {
-
binlogSource.setTableNames(Arrays.asList(response.getTableWhiteList().split(",")));
- }
- return binlogSource;
- }
-
private static AgentFileSource parseAgentFileSource(FileSourceResponse
response) {
AgentFileSource fileSource = new AgentFileSource();
fileSource.setSourceName(response.getSourceName());
@@ -216,17 +146,6 @@ public class InlongStreamSourceTransfer {
return fileSource;
}
- private static AgentFileSource parseAgentFileSource(FileSourceListResponse
response) {
- AgentFileSource fileSource = new AgentFileSource();
- fileSource.setSourceName(response.getSourceName());
- fileSource.setState(State.parseByStatus(response.getStatus()));
- fileSource.setDataFormat(DataFormat.NONE);
- fileSource.setPattern(response.getPattern());
- fileSource.setIp(response.getIp());
- fileSource.setTimeOffset(response.getTimeOffset());
- return fileSource;
- }
-
private static AutoPushSource parseAutoPushSource(AutoPushSourceResponse
response) {
AutoPushSource autoPushSource = new AutoPushSource();
autoPushSource.setSourceName(response.getSourceName());
@@ -237,15 +156,6 @@ public class InlongStreamSourceTransfer {
return autoPushSource;
}
- private static AutoPushSource
parseAutoPushSource(AutoPushSourceListResponse response) {
- AutoPushSource autoPushSource = new AutoPushSource();
- autoPushSource.setSourceName(response.getSourceName());
- autoPushSource.setState(State.parseByStatus(response.getStatus()));
- autoPushSource.setDataFormat(DataFormat.NONE);
- autoPushSource.setDataProxyGroup(response.getDataProxyGroup());
- return autoPushSource;
- }
-
private static KafkaSourceRequest createKafkaSourceRequest(KafkaSource
kafkaSource, InlongStreamInfo streamInfo) {
KafkaSourceRequest sourceRequest = new KafkaSourceRequest();
sourceRequest.setSourceName(kafkaSource.getSourceName());
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransformTransfer.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransformTransfer.java
index aa9c20776..2e606da24 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransformTransfer.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransformTransfer.java
@@ -22,11 +22,11 @@ import com.google.common.base.Splitter;
import com.google.common.collect.Sets;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.inlong.manager.common.pojo.stream.StreamTransform;
import org.apache.inlong.manager.client.api.transform.MultiDependencyTransform;
import
org.apache.inlong.manager.client.api.transform.SingleDependencyTransform;
import org.apache.inlong.manager.common.enums.TransformType;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
+import org.apache.inlong.manager.common.pojo.stream.StreamTransform;
import org.apache.inlong.manager.common.pojo.transform.TransformDefinition;
import org.apache.inlong.manager.common.pojo.transform.TransformRequest;
import org.apache.inlong.manager.common.pojo.transform.TransformResponse;
@@ -56,8 +56,7 @@ public class InlongStreamTransformTransfer {
transformRequest.setPostNodeNames(Joiner.on(",").join(streamTransform.getPostNodes()));
}
if (CollectionUtils.isNotEmpty(streamTransform.getFields())) {
- transformRequest.setFieldList(
-
InlongStreamTransfer.createStreamFields(streamTransform.getFields(),
streamInfo));
+ transformRequest.setFieldList(streamTransform.getFields());
}
return transformRequest;
}
@@ -84,7 +83,7 @@ public class InlongStreamTransformTransfer {
streamTransform.setPostNodes(Sets.newHashSet(postNodes));
}
if (CollectionUtils.isNotEmpty(transformResponse.getFieldList())) {
-
streamTransform.setFields(InlongStreamTransfer.parseStreamFields(transformResponse.getFieldList()));
+ streamTransform.setFields(transformResponse.getFieldList());
}
return streamTransform;
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/MetaFieldType.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/MetaFieldType.java
index c440d1f26..6e9d8f49f 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/MetaFieldType.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/MetaFieldType.java
@@ -55,7 +55,42 @@ public enum MetaFieldType {
/**
* event_type
*/
- EVENT_TYPE("event_type", "meta field event_type used in canal json or
mysql binlong and so on");
+ EVENT_TYPE("event_type", "meta field event_type used in canal json or
mysql binlong and so on"),
+
+ /**
+ * data
+ */
+ MYSQL_DATA("data", "MySQL binlog data Row"),
+
+ /**
+ * update_before
+ */
+ UPDATE_BEFORE("update_before", "The value of the field before update"),
+
+ /**
+ * Batch id of binlog
+ */
+ BATCH_ID("batch_id", "Batch id of binlog"),
+
+ /**
+ * sql_type
+ */
+ SQL_TYPE("sql_type", "Mapping of sql_type table fields to java data type
IDs"),
+
+ /**
+ * ts
+ */
+ TS("ts", "The current time when the ROW was received and processed"),
+
+ /**
+ * mysql_type
+ */
+ MYSQL_TYPE("mysql_type", "The table structure"),
+
+ /**
+ * pk_names
+ */
+ PK_NAMES("pk_names", "Primary key field name");
private final String name;
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkDTO.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkDTO.java
index 44eea624f..7c3ea4df3 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkDTO.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkDTO.java
@@ -81,6 +81,12 @@ public class HiveSinkDTO {
@ApiModelProperty("Properties for hive")
private Map<String, Object> properties;
+ @ApiModelProperty("Version for hive")
+ private String hiveVersion;
+
+ @ApiModelProperty("Config directory of hive, needed by sort in light mode")
+ private String hiveConfDir;
+
/**
* Get the dto instance from the request
*/
@@ -98,6 +104,8 @@ public class HiveSinkDTO {
.fileFormat(request.getFileFormat())
.dataEncoding(request.getDataEncoding())
.dataSeparator(request.getDataSeparator())
+ .hiveVersion(request.getHiveVersion())
+ .hiveConfDir(request.getHiveConfDir())
.properties(request.getProperties())
.build();
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkListResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkListResponse.java
index a6c98c193..ff26a353d 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkListResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkListResponse.java
@@ -49,4 +49,10 @@ public class HiveSinkListResponse extends SinkListResponse {
@ApiModelProperty("partition type, like: H-hour, D-day, W-week, M-month,
O-once, R-regulation")
private String partitionType;
+ @ApiModelProperty("Version for hive")
+ private String hiveVersion;
+
+ @ApiModelProperty("Config directory of hive, needed by sort in light mode")
+ private String hiveConfDir;
+
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkRequest.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkRequest.java
index 80d8dc52b..c098bee96 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkRequest.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkRequest.java
@@ -77,4 +77,10 @@ public class HiveSinkRequest extends SinkRequest {
@ApiModelProperty("Data field separator")
private String dataSeparator;
+ @ApiModelProperty("Version for hive")
+ private String hiveVersion;
+
+ @ApiModelProperty("Config directory of hive, needed by sort in light mode")
+ private String hiveConfDir;
+
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkResponse.java
index d055c3bf4..659ebf167 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/hive/HiveSinkResponse.java
@@ -72,6 +72,12 @@ public class HiveSinkResponse extends SinkResponse {
@ApiModelProperty("Data field separator")
private String dataSeparator;
+ @ApiModelProperty("Version for hive")
+ private String hiveVersion;
+
+ @ApiModelProperty("Config directory of hive, needed by sort in light mode")
+ private String hiveConfDir;
+
public HiveSinkResponse() {
this.sinkType = SinkType.SINK_HIVE;
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
index 1b1fc525c..e2e4b1efc 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
@@ -49,6 +49,13 @@ public class StreamField {
this.isMetaField = isMetaField;
}
+ public StreamField(int index, FieldType fieldType, String fieldName,
String fieldComment, String fieldValue,
+ Integer isMetaField, String originNodeName) {
+ this(index, fieldType, fieldName, fieldComment, fieldValue);
+ this.isMetaField = isMetaField;
+ this.originNodeName = originNodeName;
+ }
+
@ApiModelProperty("Field index")
private Integer id;
@@ -71,4 +78,7 @@ public class StreamField {
+ " and custom such as 'yyyy-MM-dd HH:mm:ss'. This is mainly used
for time format")
private String fieldFormat;
+ @ApiModelProperty("Origin Node name which stream field belongs")
+ private String originNodeName;
+
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformRequest.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformRequest.java
index 861979434..caf3267f5 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformRequest.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformRequest.java
@@ -20,7 +20,7 @@ package org.apache.inlong.manager.common.pojo.transform;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import javax.validation.constraints.NotNull;
import java.util.List;
@@ -66,6 +66,6 @@ public class TransformRequest {
private Integer version;
@ApiModelProperty(value = "Field list")
- private List<InlongStreamFieldInfo> fieldList;
+ private List<StreamField> fieldList;
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformResponse.java
index 63c314a83..3467675af 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/transform/TransformResponse.java
@@ -21,7 +21,7 @@ import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import org.apache.commons.compress.utils.Lists;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import java.util.List;
@@ -59,5 +59,5 @@ public class TransformResponse {
private Integer version;
@ApiModelProperty(value = "Field list")
- private List<InlongStreamFieldInfo> fieldList = Lists.newArrayList();
+ private List<StreamField> fieldList = Lists.newArrayList();
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/StreamParseUtils.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/StreamParseUtils.java
index ad81da958..c3fc2a0d5 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/StreamParseUtils.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/util/StreamParseUtils.java
@@ -18,8 +18,17 @@
package org.apache.inlong.manager.common.util;
import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.apache.inlong.manager.common.enums.DataFormat;
+import org.apache.inlong.manager.common.enums.SinkType;
+import org.apache.inlong.manager.common.enums.SourceType;
import org.apache.inlong.manager.common.enums.TransformType;
+import org.apache.inlong.manager.common.pojo.stream.SinkField;
+import org.apache.inlong.manager.common.pojo.stream.StreamNode;
import org.apache.inlong.manager.common.pojo.stream.StreamPipeline;
+import org.apache.inlong.manager.common.pojo.stream.StreamSink;
+import org.apache.inlong.manager.common.pojo.stream.StreamSource;
+import org.apache.inlong.manager.common.pojo.stream.StreamTransform;
import org.apache.inlong.manager.common.pojo.transform.TransformDefinition;
import
org.apache.inlong.manager.common.pojo.transform.deduplication.DeDuplicationDefinition;
import org.apache.inlong.manager.common.pojo.transform.filter.FilterDefinition;
@@ -27,8 +36,19 @@ import
org.apache.inlong.manager.common.pojo.transform.joiner.JoinerDefinition;
import
org.apache.inlong.manager.common.pojo.transform.replacer.StringReplacerDefinition;
import
org.apache.inlong.manager.common.pojo.transform.splitter.SplitterDefinition;
+import java.util.List;
+
public class StreamParseUtils {
+ public static final String LEFT_NODE = "leftNode";
+ public static final String RIGHT_NODE = "rightNode";
+ public static final String SOURCE_TYPE = "sourceType";
+ public static final String SOURCE_NAME = "sourceName";
+ public static final String SINK_TYPE = "sinkType";
+ public static final String SINK_NAME = "sinkName";
+ public static final String TRANSFORM_TYPE = "transformType";
+ public static final String TRANSFORM_NAME = "transformName";
+
private static Gson gson = new Gson();
public static TransformDefinition parseTransformDefinition(String
transformDefinition,
@@ -37,7 +57,7 @@ public class StreamParseUtils {
case FILTER:
return gson.fromJson(transformDefinition,
FilterDefinition.class);
case JOINER:
- return gson.fromJson(transformDefinition,
JoinerDefinition.class);
+ return parseJoinerDefinition(transformDefinition);
case SPLITTER:
return gson.fromJson(transformDefinition,
SplitterDefinition.class);
case DE_DUPLICATION:
@@ -49,6 +69,72 @@ public class StreamParseUtils {
}
}
+ public static JoinerDefinition parseJoinerDefinition(String
transformDefinition) {
+ JoinerDefinition joinerDefinition = gson.fromJson(transformDefinition,
JoinerDefinition.class);
+ JsonObject joinerJson = gson.fromJson(transformDefinition,
JsonObject.class);
+ JsonObject leftNode = joinerJson.getAsJsonObject(LEFT_NODE);
+ StreamNode leftStreamNode = parseNode(leftNode);
+ joinerDefinition.setLeftNode(leftStreamNode);
+ JsonObject rightNode = joinerJson.getAsJsonObject("rightNode");
+ StreamNode rightStreamNode = parseNode(rightNode);
+ joinerDefinition.setRightNode(rightStreamNode);
+ return joinerDefinition;
+ }
+
+ private static StreamNode parseNode(JsonObject jsonObject) {
+ if (jsonObject.has(SOURCE_TYPE)) {
+ String sourceName = jsonObject.get(SOURCE_NAME).getAsString();
+ StreamSource source = new StreamSource() {
+ @Override
+ public SourceType getSourceType() {
+ return null;
+ }
+
+ @Override
+ public SyncType getSyncType() {
+ return null;
+ }
+
+ @Override
+ public DataFormat getDataFormat() {
+ return null;
+ }
+ };
+ source.setSourceName(sourceName);
+ return source;
+ } else if (jsonObject.has(SINK_TYPE)) {
+ String sinkName = jsonObject.get(SINK_NAME).getAsString();
+ StreamSink sink = new StreamSink() {
+ @Override
+ public SinkType getSinkType() {
+ return null;
+ }
+
+ @Override
+ public List<SinkField> getSinkFields() {
+ return null;
+ }
+
+ @Override
+ public DataFormat getDataFormat() {
+ return null;
+ }
+ };
+ sink.setSinkName(sinkName);
+ return sink;
+ } else {
+ String transformName =
jsonObject.get(TRANSFORM_NAME).getAsString();
+ StreamTransform transform = new StreamTransform() {
+ @Override
+ public String getTransformName() {
+ return super.getTransformName();
+ }
+ };
+ transform.setTransformName(transformName);
+ return transform;
+ }
+ }
+
public static StreamPipeline parseStreamPipeline(String tempView, String
inlongStreamId) {
Preconditions.checkNotEmpty(tempView,
String.format(" should not be null for streamId=%s",
inlongStreamId));
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamTransformFieldEntity.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamTransformFieldEntity.java
index 8be669cc0..9fe374b15 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamTransformFieldEntity.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamTransformFieldEntity.java
@@ -51,4 +51,6 @@ public class StreamTransformFieldEntity implements
Serializable {
private Short rankNum;
private Integer isDeleted;
+
+ private String originNodeName;
}
\ No newline at end of file
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformFieldEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformFieldEntityMapper.xml
index 7e058bc68..c0e8fff14 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformFieldEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamTransformFieldEntityMapper.xml
@@ -35,12 +35,13 @@
<result column="field_format" jdbcType="VARCHAR"
property="fieldFormat"/>
<result column="rank_num" jdbcType="SMALLINT" property="rankNum"/>
<result column="is_deleted" jdbcType="INTEGER" property="isDeleted"/>
+ <result column="origin_node_name" jdbcType="VARCHAR"
property="originNodeName"/>
</resultMap>
<sql id="Base_Column_List">
id
, inlong_group_id, inlong_stream_id, transform_id, transform_type,
field_name,
field_value, pre_expression, field_type, field_comment, is_meta_field,
field_format,
- rank_num, is_deleted
+ rank_num, is_deleted, origin_node_name
</sql>
<select id="selectByTransformId"
resultType="org.apache.inlong.manager.dao.entity.StreamTransformFieldEntity">
select
@@ -76,12 +77,12 @@
transform_id, transform_type,
field_name,
field_value, pre_expression,
field_type,
field_comment, is_meta_field,
field_format,
- rank_num, is_deleted)
+ rank_num, is_deleted,
origin_node_name)
values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR},
#{inlongStreamId,jdbcType=VARCHAR},
#{transformId,jdbcType=INTEGER},
#{transformType,jdbcType=VARCHAR}, #{fieldName,jdbcType=VARCHAR},
#{fieldValue,jdbcType=VARCHAR},
#{preExpression,jdbcType=VARCHAR}, #{fieldType,jdbcType=VARCHAR},
#{fieldComment,jdbcType=VARCHAR},
#{isMetaField,jdbcType=SMALLINT}, #{fieldFormat,jdbcType=VARCHAR},
- #{rankNum,jdbcType=SMALLINT}, #{isDeleted,jdbcType=INTEGER})
+ #{rankNum,jdbcType=SMALLINT}, #{isDeleted,jdbcType=INTEGER},
#{originNodeName,jdbcType=VARCHAR})
</insert>
<insert id="insertSelective" useGeneratedKeys="true" keyProperty="id"
parameterType="org.apache.inlong.manager.dao.entity.StreamTransformFieldEntity">
@@ -129,6 +130,9 @@
<if test="isDeleted != null">
is_deleted,
</if>
+ <if test="originNodeName != null">
+ origin_node_name,
+ </if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="id != null">
@@ -173,6 +177,9 @@
<if test="isDeleted != null">
#{isDeleted,jdbcType=INTEGER},
</if>
+ <if test="originNodeName != null">
+ #{originNodeName,jdbcType=VARCHAR},
+ </if>
</trim>
</insert>
<update id="updateByPrimaryKeySelective"
@@ -218,6 +225,9 @@
<if test="isDeleted != null">
is_deleted = #{isDeleted,jdbcType=INTEGER},
</if>
+ <if test="originNodeName != null">
+ origin_node_name = #{originNodeName,jdbcType=VARCHAR},
+ </if>
</set>
where id = #{id,jdbcType=INTEGER}
</update>
@@ -235,7 +245,8 @@
is_meta_field = #{isMetaField,jdbcType=SMALLINT},
field_format = #{fieldFormat,jdbcType=VARCHAR},
rank_num = #{rankNum,jdbcType=SMALLINT},
- is_deleted = #{isDeleted,jdbcType=INTEGER}
+ is_deleted = #{isDeleted,jdbcType=INTEGER},
+ origin_node_name = #{originNodeName,jdbcType=VARCHAR}
where id = #{id,jdbcType=INTEGER}
</update>
@@ -244,7 +255,7 @@
transform_id, transform_type, field_name,
field_value, pre_expression, field_type,
field_comment, is_meta_field, field_format,
- rank_num, is_deleted)
+ rank_num, is_deleted, origin_node_name)
values
<foreach collection="list" index="index" item="item" separator=",">
(#{item.id,jdbcType=INTEGER},
#{item.inlongGroupId,jdbcType=VARCHAR},
@@ -255,7 +266,8 @@
#{item.fieldType,jdbcType=VARCHAR},
#{item.fieldComment,jdbcType=VARCHAR},
#{item.isMetaField,jdbcType=SMALLINT},
#{item.fieldFormat,jdbcType=VARCHAR},
- #{item.rankNum,jdbcType=SMALLINT},
#{item.isDeleted,jdbcType=INTEGER})
+ #{item.rankNum,jdbcType=SMALLINT},
#{item.isDeleted,jdbcType=INTEGER},
+ #{item.originNodeName,jdbcType=VARCHAR})
</foreach>
</insert>
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
index 817381f34..5f4681c3d 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java
@@ -20,8 +20,6 @@ package org.apache.inlong.manager.service.sort.util;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import java.util.List;
-import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.common.enums.DataTypeEnum;
@@ -43,6 +41,9 @@ import
org.apache.inlong.sort.protocol.node.format.DebeziumJsonFormat;
import org.apache.inlong.sort.protocol.node.format.Format;
import org.apache.inlong.sort.protocol.node.format.JsonFormat;
+import java.util.List;
+import java.util.stream.Collectors;
+
/**
* Parse SourceResponse to ExtractNode which sort needed
*/
@@ -93,9 +94,8 @@ public class ExtractNodeUtils {
List<String> tableNames = Splitter.on(",").splitToList(tables);
List<InlongStreamFieldInfo> streamFieldInfos =
binlogSourceResponse.getFieldList();
List<FieldInfo> fieldInfos = streamFieldInfos.stream()
- .map(streamFieldInfo -> new
FieldInfo(streamFieldInfo.getFieldName(), name,
-
FieldInfoUtils.convertFieldFormat(streamFieldInfo.getFieldType(),
-
streamFieldInfo.getFieldFormat()))).collect(Collectors.toList());
+ .map(streamFieldInfo ->
FieldInfoUtils.parseStreamFieldInfo(streamFieldInfo, name))
+ .collect(Collectors.toList());
String serverTimeZone = binlogSourceResponse.getServerTimezone();
return new MySqlExtractNode(id,
name,
@@ -125,9 +125,8 @@ public class ExtractNodeUtils {
String name = kafkaSourceResponse.getSourceName();
List<InlongStreamFieldInfo> streamFieldInfos =
kafkaSourceResponse.getFieldList();
List<FieldInfo> fieldInfos = streamFieldInfos.stream()
- .map(streamFieldInfo -> new
FieldInfo(streamFieldInfo.getFieldName(), name,
-
FieldInfoUtils.convertFieldFormat(streamFieldInfo.getFieldType(),
-
streamFieldInfo.getFieldFormat()))).collect(Collectors.toList());
+ .map(streamFieldInfo ->
FieldInfoUtils.parseStreamFieldInfo(streamFieldInfo, name))
+ .collect(Collectors.toList());
String topic = kafkaSourceResponse.getTopic();
String bootstrapServers = kafkaSourceResponse.getBootstrapServers();
Format format = null;
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldInfoUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldInfoUtils.java
index 37068228b..eb5ead03a 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldInfoUtils.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldInfoUtils.java
@@ -22,6 +22,7 @@ import org.apache.inlong.manager.common.enums.FieldType;
import org.apache.inlong.manager.common.enums.MetaFieldType;
import org.apache.inlong.manager.common.pojo.sink.SinkFieldResponse;
import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.sort.formats.common.ArrayFormatInfo;
import org.apache.inlong.sort.formats.common.BooleanFormatInfo;
import org.apache.inlong.sort.formats.common.ByteFormatInfo;
@@ -64,6 +65,30 @@ public class FieldInfoUtils {
BUILT_IN_FIELD_MAP.put(MetaFieldType.EVENT_TIME.getName(),
BuiltInField.MYSQL_METADATA_EVENT_TIME);
BUILT_IN_FIELD_MAP.put(MetaFieldType.IS_DDL.getName(),
BuiltInField.MYSQL_METADATA_IS_DDL);
BUILT_IN_FIELD_MAP.put(MetaFieldType.EVENT_TYPE.getName(),
BuiltInField.MYSQL_METADATA_EVENT_TYPE);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.PROCESSING_TIME.getName(),
BuiltInField.PROCESS_TIME);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.MYSQL_DATA.getName(),
BuiltInField.MYSQL_METADATA_DATA);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.UPDATE_BEFORE.getName(),
BuiltInField.METADATA_UPDATE_BEFORE);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.BATCH_ID.getName(),
BuiltInField.METADATA_BATCH_ID);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.SQL_TYPE.getName(),
BuiltInField.METADATA_SQL_TYPE);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.TS.getName(),
BuiltInField.METADATA_TS);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.MYSQL_TYPE.getName(),
BuiltInField.METADATA_MYSQL_TYPE);
+ BUILT_IN_FIELD_MAP.put(MetaFieldType.PK_NAMES.getName(),
BuiltInField.METADATA_PK_NAMES);
+ }
+
+ public static FieldInfo parseStreamFieldInfo(InlongStreamFieldInfo
streamField, String name) {
+ boolean isBuiltIn = streamField.getIsMetaField() == 1;
+ FieldInfo fieldInfo = getFieldInfo(streamField.getFieldName(),
streamField.getFieldType(), isBuiltIn,
+ streamField.getFieldFormat());
+ fieldInfo.setNodeId(name);
+ return fieldInfo;
+ }
+
+ public static FieldInfo parseStreamField(StreamField streamField) {
+ boolean isBuiltIn = streamField.getIsMetaField() == 1;
+ FieldInfo fieldInfo = getFieldInfo(streamField.getFieldName(),
streamField.getFieldType().toString(), isBuiltIn,
+ streamField.getFieldFormat());
+ fieldInfo.setNodeId(streamField.getOriginNodeName());
+ return fieldInfo;
}
/**
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldRelationShipUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldRelationShipUtils.java
index bef4df867..10a9bd361 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldRelationShipUtils.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/FieldRelationShipUtils.java
@@ -19,8 +19,8 @@ package org.apache.inlong.manager.service.sort.util;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import org.apache.inlong.manager.common.enums.FieldType;
import org.apache.inlong.manager.common.enums.TransformType;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.pojo.transform.TransformDefinition;
import org.apache.inlong.manager.common.pojo.transform.TransformResponse;
@@ -29,6 +29,7 @@ import
org.apache.inlong.manager.common.pojo.transform.replacer.StringReplacerDe
import
org.apache.inlong.manager.common.pojo.transform.replacer.StringReplacerDefinition.ReplaceRule;
import
org.apache.inlong.manager.common.pojo.transform.splitter.SplitterDefinition;
import
org.apache.inlong.manager.common.pojo.transform.splitter.SplitterDefinition.SplitRule;
+import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.common.util.StreamParseUtils;
import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.transformation.ConstantParam;
@@ -48,60 +49,67 @@ public class FieldRelationShipUtils {
TransformType transformType =
TransformType.forType(transformResponse.getTransformType());
TransformDefinition transformDefinition =
StreamParseUtils.parseTransformDefinition(
transformResponse.getTransformDefinition(), transformType);
- List<InlongStreamFieldInfo> fieldList =
transformResponse.getFieldList();
+ List<StreamField> fieldList = transformResponse.getFieldList();
String transformName = transformResponse.getTransformName();
+ String preNodes = transformResponse.getPreNodeNames();
switch (transformType) {
case SPLITTER:
SplitterDefinition splitterDefinition = (SplitterDefinition)
transformDefinition;
- return createFieldRelationShips(fieldList, transformName,
splitterDefinition);
+ return createSplitterFieldRelationShips(fieldList,
transformName, splitterDefinition, preNodes);
+ case STRING_REPLACER:
+ StringReplacerDefinition replacerDefinition =
(StringReplacerDefinition) transformDefinition;
+ return createReplacerFieldRelationShips(fieldList,
transformName, replacerDefinition, preNodes);
case DE_DUPLICATION:
- case JOINER:
case FILTER:
+ case JOINER:
return createFieldRelationShips(fieldList, transformName);
- case STRING_REPLACER:
- StringReplacerDefinition replacerDefinition =
(StringReplacerDefinition) transformDefinition;
- return createFieldRelationShips(fieldList, transformName,
replacerDefinition);
default:
throw new UnsupportedOperationException(
String.format("Unsupported transformType=%s for
Inlong", transformType));
}
}
- private static List<FieldRelationShip>
createFieldRelationShips(List<InlongStreamFieldInfo> fieldList,
- String transformName) {
- return fieldList.stream().map(streamFieldInfo ->
- new FieldInfo(streamFieldInfo.getFieldName(),
transformName,
-
FieldInfoUtils.convertFieldFormat(streamFieldInfo.getFieldType(),
- streamFieldInfo.getFieldFormat())))
- .map(fieldInfo -> new FieldRelationShip(fieldInfo, fieldInfo))
- .collect(Collectors.toList());
+ private static List<FieldRelationShip>
createFieldRelationShips(List<StreamField> fieldList, String transformName) {
+ return fieldList.stream()
+ .map(streamFieldInfo ->
FieldInfoUtils.parseStreamField(streamFieldInfo))
+ .map(fieldInfo -> {
+ FieldInfo inputField = new FieldInfo(fieldInfo.getName(),
fieldInfo.getNodeId(),
+ fieldInfo.getFormatInfo());
+ FieldInfo outputField = new FieldInfo(fieldInfo.getName(),
transformName,
+ fieldInfo.getFormatInfo());
+ return new FieldRelationShip(inputField, outputField);
+ }).collect(Collectors.toList());
}
- private static List<FieldRelationShip>
createFieldRelationShips(List<InlongStreamFieldInfo> fieldList,
- String transformName, SplitterDefinition splitterDefinition) {
+ private static List<FieldRelationShip>
createSplitterFieldRelationShips(List<StreamField> fieldList,
+ String transformName, SplitterDefinition splitterDefinition,
String preNodes) {
+ Preconditions.checkNotEmpty(preNodes, "PreNodes of splitter should not
be null");
+ String preNode = preNodes.split(",")[0];
List<SplitRule> splitRules = splitterDefinition.getSplitRules();
Set<String> splitFields = Sets.newHashSet();
List<FieldRelationShip> fieldRelationShips = splitRules.stream()
- .map(splitRule -> parseSplitRule(splitRule, splitFields,
transformName))
+ .map(splitRule -> parseSplitRule(splitRule, splitFields,
transformName, preNode))
.reduce(Lists.newArrayList(), (list1, list2) -> {
list1.addAll(list2);
return list1;
});
- List<InlongStreamFieldInfo> filteredFieldList = fieldList.stream()
+ List<StreamField> filteredFieldList = fieldList.stream()
.filter(streamFieldInfo ->
splitFields.contains(streamFieldInfo.getFieldName()))
.collect(Collectors.toList());
fieldRelationShips.addAll(createFieldRelationShips(filteredFieldList,
transformName));
return fieldRelationShips;
}
- private static List<FieldRelationShip>
createFieldRelationShips(List<InlongStreamFieldInfo> fieldList,
- String transformName, StringReplacerDefinition replacerDefinition)
{
+ private static List<FieldRelationShip>
createReplacerFieldRelationShips(List<StreamField> fieldList,
+ String transformName, StringReplacerDefinition replacerDefinition,
String preNodes) {
+ Preconditions.checkNotEmpty(preNodes, "PreNodes of splitter should not
be null");
+ String preNode = preNodes.split(",")[0];
List<ReplaceRule> replaceRules = replacerDefinition.getReplaceRules();
Set<String> replaceFields = Sets.newHashSet();
List<FieldRelationShip> fieldRelationShips = replaceRules.stream()
- .map(replaceRule -> parseReplaceRule(replaceRule,
replaceFields, transformName))
+ .map(replaceRule -> parseReplaceRule(replaceRule,
replaceFields, transformName, preNode))
.collect(Collectors.toList());
- List<InlongStreamFieldInfo> filteredFieldList = fieldList.stream()
+ List<StreamField> filteredFieldList = fieldList.stream()
.filter(streamFieldInfo ->
replaceFields.contains(streamFieldInfo.getFieldName()))
.collect(Collectors.toList());
fieldRelationShips.addAll(createFieldRelationShips(filteredFieldList,
transformName));
@@ -109,15 +117,17 @@ public class FieldRelationShipUtils {
}
private static FieldRelationShip parseReplaceRule(ReplaceRule replaceRule,
Set<String> replaceFields,
- String transformName) {
+ String transformName, String preNode) {
StreamField sourceField = replaceRule.getSourceField();
+ final String fieldName = sourceField.getFieldName();
String regex = replaceRule.getRegex();
String targetValue = replaceRule.getTargetValue();
ReplaceMode replaceMode = replaceRule.getMode();
- FieldInfo fieldInfo = new FieldInfo(sourceField.getFieldName(),
transformName,
-
FieldInfoUtils.convertFieldFormat(sourceField.getFieldType().name(),
sourceField.getFieldFormat()));
- FieldInfo targetFieldInfo = new FieldInfo(sourceField.getFieldName(),
transformName,
-
FieldInfoUtils.convertFieldFormat(sourceField.getFieldType().name(),
sourceField.getFieldFormat()));
+ FieldInfo fieldInfo = FieldInfoUtils.parseStreamField(sourceField);
+ fieldInfo.setNodeId(preNode);
+ FieldInfo targetFieldInfo = new FieldInfo(fieldName, transformName,
+ FieldInfoUtils.convertFieldFormat(FieldType.STRING.name()));
+ replaceFields.add(fieldName);
if (replaceMode == ReplaceMode.RELACE_ALL) {
RegexpReplaceFunction regexpReplaceFunction = new
RegexpReplaceFunction(fieldInfo,
new StringConstantParam(regex), new
StringConstantParam(targetValue));
@@ -130,11 +140,11 @@ public class FieldRelationShipUtils {
}
private static List<FieldRelationShip> parseSplitRule(SplitRule splitRule,
Set<String> splitFields,
- String transformName) {
+ String transformName, String preNode) {
StreamField sourceField = splitRule.getSourceField();
splitFields.add(sourceField.getFieldName());
- FieldInfo fieldInfo = new FieldInfo(sourceField.getFieldName(),
transformName,
-
FieldInfoUtils.convertFieldFormat(sourceField.getFieldType().name(),
sourceField.getFieldFormat()));
+ FieldInfo fieldInfo = FieldInfoUtils.parseStreamField(sourceField);
+ fieldInfo.setNodeId(preNode);
String seperator = splitRule.getSeperator();
List<String> targetSources = splitRule.getTargetFields();
List<FieldRelationShip> splitRelationShips = Lists.newArrayList();
@@ -142,11 +152,10 @@ public class FieldRelationShipUtils {
SplitIndexFunction splitIndexFunction = new SplitIndexFunction(
fieldInfo, new StringConstantParam(seperator), new
ConstantParam(index));
FieldInfo targetFieldInfo = new FieldInfo(
- targetSources.get(index), transformName,
FieldInfoUtils.convertFieldFormat("STRING")
+ targetSources.get(index), transformName,
FieldInfoUtils.convertFieldFormat(FieldType.STRING.name())
);
splitRelationShips.add(new FieldRelationShip(splitIndexFunction,
targetFieldInfo));
}
return splitRelationShips;
}
-
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/LoadNodeUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/LoadNodeUtils.java
index 928e701ce..4e1080b8b 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/LoadNodeUtils.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/LoadNodeUtils.java
@@ -24,6 +24,7 @@ import org.apache.inlong.common.enums.DataTypeEnum;
import org.apache.inlong.manager.common.enums.SinkType;
import org.apache.inlong.manager.common.pojo.sink.SinkFieldResponse;
import org.apache.inlong.manager.common.pojo.sink.SinkResponse;
+import org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkResponse;
import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkResponse;
import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.node.LoadNode;
@@ -33,6 +34,7 @@ import org.apache.inlong.sort.protocol.node.format.CsvFormat;
import org.apache.inlong.sort.protocol.node.format.DebeziumJsonFormat;
import org.apache.inlong.sort.protocol.node.format.Format;
import org.apache.inlong.sort.protocol.node.format.JsonFormat;
+import org.apache.inlong.sort.protocol.node.load.HiveLoadNode;
import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
@@ -55,6 +57,8 @@ public class LoadNodeUtils {
switch (sinkType) {
case KAFKA:
return createLoadNode((KafkaSinkResponse) sinkResponse);
+ case HIVE:
+ return createLoadNode((HiveSinkResponse) sinkResponse);
default:
throw new IllegalArgumentException(
String.format("Unsupported sinkType=%s to create
loadNode", sinkType));
@@ -114,6 +118,46 @@ public class LoadNodeUtils {
primaryKey);
}
+ public static HiveLoadNode createLoadNode(HiveSinkResponse
hiveSinkResponse) {
+ String id = hiveSinkResponse.getSinkName();
+ String name = hiveSinkResponse.getSinkName();
+ String database = hiveSinkResponse.getDbName();
+ String tableName = hiveSinkResponse.getTableName();
+ String hiveConfDir = hiveSinkResponse.getHiveConfDir();
+ String hiveVersion = hiveSinkResponse.getHiveVersion();
+ List<SinkFieldResponse> sinkFieldResponses =
hiveSinkResponse.getFieldList();
+ List<FieldInfo> fields = sinkFieldResponses.stream()
+ .map(sinkFieldResponse -> new
FieldInfo(sinkFieldResponse.getFieldName(), name,
+
FieldInfoUtils.convertFieldFormat(sinkFieldResponse.getFieldType(),
+
sinkFieldResponse.getFieldFormat()))).collect(Collectors.toList());
+ List<FieldRelationShip> fieldRelationShips =
parseSinkFields(sinkFieldResponses, name);
+ Map<String, String> properties =
hiveSinkResponse.getProperties().entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().toString()));
+ List<FieldInfo> partitionFields = Lists.newArrayList();
+ if
(CollectionUtils.isNotEmpty(hiveSinkResponse.getPartitionFieldList())) {
+ partitionFields = hiveSinkResponse.getPartitionFieldList().stream()
+ .map(hivePartitionField -> new
FieldInfo(hivePartitionField.getFieldName(), name,
+
FieldInfoUtils.convertFieldFormat(hivePartitionField.getFieldType(),
+
hivePartitionField.getFieldFormat()))).collect(Collectors.toList());
+ }
+ return new HiveLoadNode(
+ id,
+ name,
+ fields,
+ fieldRelationShips,
+ Lists.newArrayList(),
+ null,
+ properties,
+ null,
+ database,
+ tableName,
+ hiveConfDir,
+ hiveVersion,
+ null,
+ partitionFields
+ );
+ }
+
public static List<FieldRelationShip>
parseSinkFields(List<SinkFieldResponse> sinkFieldResponses, String sinkName) {
if (CollectionUtils.isEmpty(sinkFieldResponses)) {
return Lists.newArrayList();
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/TransformNodeUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/TransformNodeUtils.java
index 8952414b7..5221a300c 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/TransformNodeUtils.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/TransformNodeUtils.java
@@ -70,16 +70,12 @@ public class TransformNodeUtils {
*/
public static DistinctNode createDistinctNode(DeDuplicationDefinition
deDuplicationDefinition,
TransformResponse transformResponse) {
- String transformName = transformResponse.getTransformName();
List<StreamField> streamFields =
deDuplicationDefinition.getDupFields();
List<FieldInfo> distinctFields = streamFields.stream()
- .map(streamField -> new FieldInfo(streamField.getFieldName(),
transformName,
-
FieldInfoUtils.convertFieldFormat(streamField.getFieldType().name(),
- streamField.getFieldFormat())))
+ .map(streamField ->
FieldInfoUtils.parseStreamField(streamField))
.collect(Collectors.toList());
StreamField timingField = deDuplicationDefinition.getTimingField();
- FieldInfo orderField = new FieldInfo(timingField.getFieldName(),
transformName,
-
FieldInfoUtils.convertFieldFormat(timingField.getFieldType().name(),
timingField.getFieldFormat()));
+ FieldInfo orderField = FieldInfoUtils.parseStreamField(timingField);
DeDuplicationStrategy deDuplicationStrategy =
deDuplicationDefinition.getDeDuplicationStrategy();
OrderDirection orderDirection = null;
switch (deDuplicationStrategy) {
@@ -115,15 +111,8 @@ public class TransformNodeUtils {
TransformNode transformNode = new TransformNode();
transformNode.setId(transformResponse.getTransformName());
transformNode.setName(transformResponse.getTransformName());
- List<FieldInfo> fieldInfos =
transformResponse.getFieldList().stream().map(streamFieldInfo -> {
- String transformName = transformResponse.getTransformName();
- String fieldType = streamFieldInfo.getFieldType();
- String fieldFormat = streamFieldInfo.getFieldFormat();
- String fieldName = streamFieldInfo.getFieldName();
- FieldInfo fieldInfo = new FieldInfo(fieldName, transformName,
- FieldInfoUtils.convertFieldFormat(fieldType, fieldFormat));
- return fieldInfo;
- }).collect(Collectors.toList());
+ List<FieldInfo> fieldInfos = transformResponse.getFieldList().stream()
+ .map(streamFieldInfo ->
FieldInfoUtils.parseStreamField(streamFieldInfo)).collect(Collectors.toList());
transformNode.setFields(fieldInfos);
transformNode.setFieldRelationShips(FieldRelationShipUtils.createFieldRelationShips(transformResponse));
transformNode.setFilters(
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
index 42785b8c4..9d3a31b2a 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/transform/StreamTransformServiceImpl.java
@@ -24,7 +24,7 @@ import org.apache.commons.lang3.tuple.Pair;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.GlobalConstants;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.common.pojo.stream.InlongStreamFieldInfo;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.common.pojo.transform.TransformRequest;
import org.apache.inlong.manager.common.pojo.transform.TransformResponse;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
@@ -105,10 +105,10 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
.collect(Collectors.toList());
List<StreamTransformFieldEntity> transformFieldEntities =
transformFieldEntityMapper.selectByTransformIds(
transformIds);
- Map<Integer, List<InlongStreamFieldInfo>> fieldInfoMap =
transformFieldEntities.stream()
+ Map<Integer, List<StreamField>> fieldInfoMap =
transformFieldEntities.stream()
.map(transformFieldEntity -> {
- InlongStreamFieldInfo fieldInfo =
CommonBeanUtils.copyProperties(transformFieldEntity,
- InlongStreamFieldInfo::new);
+ StreamField fieldInfo =
CommonBeanUtils.copyProperties(transformFieldEntity,
+ StreamField::new);
return Pair.of(transformFieldEntity.getTransformId(),
fieldInfo);
}).collect(Collectors.groupingBy(Pair::getLeft,
Collectors.mapping(Pair::getRight,
Collectors.toList())));
@@ -117,7 +117,7 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
.collect(Collectors.toList());
transformResponses.stream().forEach(transformResponse -> {
int transformId = transformResponse.getId();
- List<InlongStreamFieldInfo> fieldInfos =
fieldInfoMap.get(transformId);
+ List<StreamField> fieldInfos = fieldInfoMap.get(transformId);
if (CollectionUtils.isNotEmpty(fieldInfos)) {
transformResponse.setFieldList(fieldInfos);
}
@@ -184,7 +184,7 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
Preconditions.checkNotNull(transformName,
ErrorCodeEnum.TRANSFORM_NAME_IS_NULL.getMessage());
}
- private void updateFieldOpt(StreamTransformEntity entity,
List<InlongStreamFieldInfo> fieldInfos) {
+ private void updateFieldOpt(StreamTransformEntity entity,
List<StreamField> fieldInfos) {
Integer transformId = entity.getId();
if (CollectionUtils.isEmpty(fieldInfos)) {
return;
@@ -198,7 +198,7 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
log.info("success to update transform field");
}
- private void saveFieldOpt(StreamTransformEntity entity,
List<InlongStreamFieldInfo> fieldInfos) {
+ private void saveFieldOpt(StreamTransformEntity entity, List<StreamField>
fieldInfos) {
log.info("begin to save transform field={}", fieldInfos);
if (CollectionUtils.isEmpty(fieldInfos)) {
return;
@@ -210,7 +210,7 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
String streamId = entity.getInlongStreamId();
String transformType = entity.getTransformType();
Integer transformId = entity.getId();
- for (InlongStreamFieldInfo fieldInfo : fieldInfos) {
+ for (StreamField fieldInfo : fieldInfos) {
StreamTransformFieldEntity fieldEntity =
CommonBeanUtils.copyProperties(fieldInfo,
StreamTransformFieldEntity::new);
if (StringUtils.isEmpty(fieldEntity.getFieldComment())) {
@@ -221,6 +221,7 @@ public class StreamTransformServiceImpl implements
StreamTransformService {
fieldEntity.setTransformId(transformId);
fieldEntity.setTransformType(transformType);
fieldEntity.setIsDeleted(GlobalConstants.UN_DELETED);
+ fieldEntity.setOriginNodeName(fieldInfo.getOriginNodeName());
entityList.add(fieldEntity);
}
diff --git
a/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
b/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
index 7d8146bb3..c86a00abc 100644
---
a/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
+++
b/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
@@ -633,8 +633,9 @@ CREATE TABLE `stream_transform_field`
`field_comment` varchar(2000) DEFAULT NULL COMMENT 'Field description',
`is_meta_field` smallint(3) DEFAULT '0' COMMENT 'Is this field a meta
field? 0: no, 1: yes',
`field_format` varchar(50) DEFAULT NULL COMMENT 'Field format,
including: MICROSECONDS, MILLISECONDS, SECONDS, SQL, ISO_8601 and custom such
as yyyy-MM-dd HH:mm:ss',
- `rank_num` smallint(6) DEFAULT '0' COMMENT 'Field order
(front-end display field order)',
- `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete,
0: not deleted, > 0: deleted',
+ `rank_num` smallint(6) DEFAULT '0' COMMENT 'Field order (front-end
display field order)',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not
deleted, > 0: deleted',
+ `origin_node_name` varchar(256) DEFAULT '' COMMENT 'Origin Node name
which stream field belongs',
PRIMARY KEY (`id`),
KEY `index_transform_id` (`transform_id`)
) ENGINE = InnoDB
diff --git a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
index 5d501a001..5d8a4ee63 100644
--- a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
@@ -665,6 +665,7 @@ CREATE TABLE `stream_transform_field`
`field_format` varchar(50) DEFAULT NULL COMMENT 'Field format,
including: MICROSECONDS, MILLISECONDS, SECONDS, SQL, ISO_8601 and custom such
as yyyy-MM-dd HH:mm:ss',
`rank_num` smallint(6) DEFAULT '0' COMMENT 'Field order
(front-end display field order)',
`is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete,
0: not deleted, > 0: deleted',
+ `origin_node_name` varchar(256) DEFAULT '' COMMENT 'Origin Node name
which stream field belongs',
PRIMARY KEY (`id`),
KEY `index_transform_id` (`transform_id`)
) ENGINE = InnoDB
