This is an automated email from the ASF dual-hosted git repository.
gosonzhang 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 31a33832a [INLONG-3542][Manager] Add Iceberg params and change SQL
files (#3555)
31a33832a is described below
commit 31a33832ae4b6961302c7bb3ab7593058b67a889
Author: healchow <[email protected]>
AuthorDate: Thu Apr 7 09:36:57 2022 +0800
[INLONG-3542][Manager] Add Iceberg params and change SQL files (#3555)
---
.../manager/common/pojo/sink/SinkFieldBase.java | 12 ++++++
.../manager/common/pojo/sink/SinkForSortDTO.java | 3 +-
.../manager/common/pojo/sink/SinkListResponse.java | 4 +-
.../manager/common/pojo/sink/SinkRequest.java | 6 +--
.../manager/common/pojo/sink/SinkResponse.java | 15 ++++----
.../common/pojo/sink/iceberg/IcebergSinkDTO.java | 44 +++++++++++++++++++--
.../pojo/sink/iceberg/IcebergSinkListResponse.java | 20 +++++++++-
.../pojo/sink/iceberg/IcebergSinkRequest.java | 31 ++++++++++++++-
.../pojo/sink/iceberg/IcebergSinkResponse.java | 33 ++++++++++++++--
.../manager/dao/entity/StreamSinkEntity.java | 2 +-
.../manager/dao/entity/StreamSinkFieldEntity.java | 6 +++
.../resources/mappers/StreamSinkEntityMapper.xml | 26 ++++++-------
.../mappers/StreamSinkFieldEntityMapper.xml | 45 +++++++++++++++-------
.../service/sink/hive/HiveStreamSinkOperation.java | 2 +-
.../sink/iceberg/IcebergStreamSinkOperation.java | 4 +-
.../core/sink/IcebergStreamSinkServiceTest.java | 3 +-
.../main/resources/sql/apache_inlong_manager.sql | 34 ++++++++--------
.../manager-web/sql/apache_inlong_manager.sql | 34 ++++++++--------
18 files changed, 235 insertions(+), 89 deletions(-)
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkFieldBase.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkFieldBase.java
index a737e402c..ff355e32c 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkFieldBase.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkFieldBase.java
@@ -43,6 +43,18 @@ public class SinkFieldBase {
@ApiModelProperty("Source field type")
private String sourceFieldType;
+ @ApiModelProperty("Field length")
+ private Integer fieldLength;
+
+ @ApiModelProperty("Field precision")
+ private Integer fieldPrecision;
+
+ @ApiModelProperty("Field scale")
+ private Integer fieldScale;
+
+ @ApiModelProperty("Partition strategy, including: Identity, Year, Month,
Day, Hour, Bucket")
+ private String partitionStrategy;
+
@ApiModelProperty("Field format, including: MICROSECONDS, MILLISECONDS,
SECONDS, SQL, ISO_8601"
+ " and custom such as 'yyyy-MM-dd HH:mm:ss'. This is mainly used
for time format")
private String fieldFormat;
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkForSortDTO.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkForSortDTO.java
index 6b6d356eb..581f6e8de 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkForSortDTO.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkForSortDTO.java
@@ -32,7 +32,7 @@ public class SinkForSortDTO {
private String inlongStreamId;
private String sinkType;
private String sinkName;
- private Integer storagePeriod;
+ private String description;
private Integer enableCreateResource;
private String extParams;
private Integer status;
@@ -42,7 +42,6 @@ public class SinkForSortDTO {
private String mqResourceObj;
private String dataSourceType;
private String dataType;
- private String description;
private String sourceSeparator; // Source separator configured in the
stream info
private String dataEscapeChar;
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
index 4c46022d2..b8212a99b 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkListResponse.java
@@ -48,8 +48,8 @@ public class SinkListResponse {
@ApiModelProperty("Sink name, unique in one stream.")
private String sinkName;
- @ApiModelProperty("Sink period, unit: day")
- private Integer storagePeriod;
+ @ApiModelProperty("Sink description")
+ private String description;
@ApiModelProperty(value = "Whether to enable create sink resource? 0:
disable, 1: enable. default is 1")
private Integer enableCreateResource;
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
index 68bbc24c4..52793f588 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkRequest.java
@@ -22,11 +22,11 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import com.google.common.collect.Maps;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
-import java.util.Map;
import lombok.Data;
import javax.validation.constraints.NotNull;
import java.util.List;
+import java.util.Map;
/**
* Request of sink
@@ -54,8 +54,8 @@ public class SinkRequest {
@NotNull
private String sinkName;
- @ApiModelProperty("Data storage period, unit: day")
- private Integer storagePeriod;
+ @ApiModelProperty("Sink description")
+ private String description;
@ApiModelProperty(value = "Whether to enable create sink resource? 0:
disable, 1: enable. default is 1")
private Integer enableCreateResource = 1;
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkResponse.java
index e0683faee..6b25928fc 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkResponse.java
@@ -20,12 +20,12 @@ package org.apache.inlong.manager.common.pojo.sink;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
-import java.util.Map;
import lombok.Data;
+import lombok.NoArgsConstructor;
import java.util.Date;
import java.util.List;
-import lombok.NoArgsConstructor;
+import java.util.Map;
/**
* Response of the stream sink
@@ -35,6 +35,10 @@ import lombok.NoArgsConstructor;
@NoArgsConstructor
public class SinkResponse {
+ @ApiModelProperty("Sink type, including: HIVE, ICEBERG, etc.")
+ protected String sinkType;
+
+ @ApiModelProperty("Sink id")
private Integer id;
@ApiModelProperty("Inlong group id")
@@ -46,11 +50,8 @@ public class SinkResponse {
@ApiModelProperty("Sink name, unique in one stream.")
private String sinkName;
- @ApiModelProperty("Sink type, including: HIVE, ES, etc.")
- protected String sinkType;
-
- @ApiModelProperty("Data storage period, unit: day")
- private Integer storagePeriod;
+ @ApiModelProperty("Sink description")
+ private String description;
@ApiModelProperty(value = "Whether to enable create sink resource? 0:
disable, 1: enable. default is 1",
notes = "Such as create Hive table")
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkDTO.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkDTO.java
index 7d4737c84..27f716a95 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkDTO.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkDTO.java
@@ -20,7 +20,6 @@ package org.apache.inlong.manager.common.pojo.sink.iceberg;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.swagger.annotations.ApiModelProperty;
-import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@@ -29,6 +28,7 @@ import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.exceptions.BusinessException;
import javax.validation.constraints.NotNull;
+import java.util.Map;
/**
* Iceberg sink info
@@ -41,8 +41,35 @@ public class IcebergSinkDTO {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
- @ApiModelProperty("table Location like hdfs://")
- private String tableLocation;
+ @ApiModelProperty("Hive JDBC URL")
+ private String jdbcUrl;
+
+ @ApiModelProperty("Username for JDBC URL")
+ private String username;
+
+ @ApiModelProperty("User password")
+ private String password;
+
+ @ApiModelProperty("Target database name")
+ private String dbName;
+
+ @ApiModelProperty("Target table name")
+ private String tableName;
+
+ @ApiModelProperty("Data path, such as:
hdfs://ip:port/user/hive/warehouse/test.db")
+ private String dataPath;
+
+ @ApiModelProperty("File format, support: Parquet, Orc, Avro")
+ private String fileFormat;
+
+ @ApiModelProperty("Data encoding type")
+ private String dataEncoding;
+
+ @ApiModelProperty("Data field separator")
+ private String dataSeparator;
+
+ @ApiModelProperty("Data consistency strategy, support:
EXACTLY_ONCE(default), AT_LEAST_ONCE")
+ private String dataConsistency;
@ApiModelProperty("Properties for iceberg")
private Map<String, Object> properties;
@@ -52,7 +79,16 @@ public class IcebergSinkDTO {
*/
public static IcebergSinkDTO getFromRequest(IcebergSinkRequest request) {
return IcebergSinkDTO.builder()
- .tableLocation(request.getTableLocation())
+ .jdbcUrl(request.getJdbcUrl())
+ .username(request.getUsername())
+ .password(request.getPassword())
+ .dbName(request.getDbName())
+ .tableName(request.getTableName())
+ .dataPath(request.getDataPath())
+ .fileFormat(request.getFileFormat())
+ .dataEncoding(request.getDataEncoding())
+ .dataSeparator(request.getDataSeparator())
+ .dataConsistency(request.getDataConsistency())
.properties(request.getProperties())
.build();
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkListResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkListResponse.java
index 7aa495c5b..7458e65a5 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkListResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkListResponse.java
@@ -31,6 +31,22 @@ import
org.apache.inlong.manager.common.pojo.sink.SinkListResponse;
@ApiModel("Response of Iceberg sink paging list")
public class IcebergSinkListResponse extends SinkListResponse {
- @ApiModelProperty("table Location like hdfs://")
- private String tableLocation;
+ @ApiModelProperty("target database name")
+ private String dbName;
+
+ @ApiModelProperty("target table name")
+ private String tableName;
+
+ @ApiModelProperty("username")
+ private String username;
+
+ @ApiModelProperty("JDBC URL")
+ private String jdbcUrl;
+
+ @ApiModelProperty("Data path, such as:
hdfs://ip:port/user/hive/warehouse/test.db")
+ private String dataPath;
+
+ @ApiModelProperty("partition type, like: H-hour, D-day, W-week, M-month,
O-once, R-regulation")
+ private String partitionType;
+
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkRequest.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkRequest.java
index fa0ba9f6b..1fb0a24f4 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkRequest.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkRequest.java
@@ -36,7 +36,34 @@ import org.apache.inlong.manager.common.util.JsonTypeDefine;
@JsonTypeDefine(value = Constant.SINK_ICEBERG)
public class IcebergSinkRequest extends SinkRequest {
- @ApiModelProperty("table Location like hdfs://")
- private String tableLocation;
+ @ApiModelProperty("Hive JDBC URL")
+ private String jdbcUrl;
+
+ @ApiModelProperty("Username for JDBC URL")
+ private String username;
+
+ @ApiModelProperty("User password")
+ private String password;
+
+ @ApiModelProperty("Target database name")
+ private String dbName;
+
+ @ApiModelProperty("Target table name")
+ private String tableName;
+
+ @ApiModelProperty("Data path, such as:
hdfs://ip:port/user/hive/warehouse/test.db")
+ private String dataPath;
+
+ @ApiModelProperty("File format, support: Parquet, Orc, Avro")
+ private String fileFormat;
+
+ @ApiModelProperty("Data encoding type")
+ private String dataEncoding;
+
+ @ApiModelProperty("Data field separator")
+ private String dataSeparator;
+
+ @ApiModelProperty("Data consistency strategy, support:
EXACTLY_ONCE(default), AT_LEAST_ONCE")
+ private String dataConsistency;
}
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkResponse.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkResponse.java
index 1abe819b0..f40ceb8bc 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkResponse.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/iceberg/IcebergSinkResponse.java
@@ -34,10 +34,37 @@ import
org.apache.inlong.manager.common.pojo.sink.SinkResponse;
@ApiModel(value = "Response of the Iceberg sink")
public class IcebergSinkResponse extends SinkResponse {
+ @ApiModelProperty("Hive JDBC URL")
+ private String jdbcUrl;
+
+ @ApiModelProperty("Username for JDBC URL")
+ private String username;
+
+ @ApiModelProperty("User password")
+ private String password;
+
+ @ApiModelProperty("Target database name")
+ private String dbName;
+
+ @ApiModelProperty("Target table name")
+ private String tableName;
+
+ @ApiModelProperty("Data path, such as:
hdfs://ip:port/user/hive/warehouse/test.db")
+ private String dataPath;
+
+ @ApiModelProperty("File format, support: Parquet, Orc, Avro")
+ private String fileFormat;
+
+ @ApiModelProperty("Data encoding type")
+ private String dataEncoding;
+
+ @ApiModelProperty("Data field separator")
+ private String dataSeparator;
+
+ @ApiModelProperty("Data consistency strategy, support:
EXACTLY_ONCE(default), AT_LEAST_ONCE")
+ private String dataConsistency;
+
public IcebergSinkResponse() {
this.sinkType = Constant.SINK_ICEBERG;
}
-
- @ApiModelProperty("table Location like hdfs://")
- private String tableLocation;
}
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkEntity.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkEntity.java
index dff355d70..cd49ad404 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkEntity.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkEntity.java
@@ -31,7 +31,7 @@ public class StreamSinkEntity implements Serializable {
private String inlongStreamId;
private String sinkType;
private String sinkName;
- private Integer storagePeriod;
+ private String description;
private Integer enableCreateResource;
private String operateLog;
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkFieldEntity.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkFieldEntity.java
index b2ac03437..00dee4daf 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkFieldEntity.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/StreamSinkFieldEntity.java
@@ -37,6 +37,12 @@ public class StreamSinkFieldEntity implements Serializable {
private Integer isRequired;
private String sourceFieldName;
private String sourceFieldType;
+
+ private Integer fieldLength;
+ private Integer fieldPrecision;
+ private Integer fieldScale;
+ private String partitionStrategy; // For Iceberg, including: Identity,
Year, Month, Day, Hour, Bucket
+
private Integer isMetaField;
private String fieldFormat;
private Short rankNum;
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
index a1ecfc5c9..4e07890ab 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
@@ -27,7 +27,7 @@
<result column="inlong_stream_id" jdbcType="VARCHAR"
property="inlongStreamId"/>
<result column="sink_type" jdbcType="VARCHAR" property="sinkType"/>
<result column="sink_name" jdbcType="VARCHAR" property="sinkName"/>
- <result column="storage_period" jdbcType="INTEGER"
property="storagePeriod"/>
+ <result column="description" jdbcType="INTEGER"
property="description"/>
<result column="enable_create_resource" jdbcType="TINYINT"
property="enableCreateResource"/>
<result column="ext_params" jdbcType="VARCHAR" property="extParams"/>
<result column="operate_log" jdbcType="VARCHAR" property="operateLog"/>
@@ -42,8 +42,7 @@
</resultMap>
<sql id="Base_Column_List">
- id
- , inlong_group_id, inlong_stream_id, sink_type, sink_name,
storage_period, enable_create_resource, ext_params,
+ id, inlong_group_id, inlong_stream_id, sink_type, sink_name,
description, enable_create_resource, ext_params,
operate_log, status, previous_status, is_deleted, creator, modifier,
create_time, modify_time
</sql>
@@ -164,7 +163,7 @@
sink.inlong_group_id,
sink.inlong_stream_id,
sink.sink_type,
- sink.storage_period,
+ sink.description,
sink.enable_create_resource,
sink.ext_params,
sink.status,
@@ -173,7 +172,6 @@
stream.mq_resource_obj,
stream.data_source_type,
stream.data_type,
- stream.description,
stream.data_separator as sourceSeparator,
stream.data_escape_char
from inlong_stream stream, stream_sink sink
@@ -198,13 +196,13 @@
<insert id="insert" useGeneratedKeys="true" keyProperty="id"
parameterType="org.apache.inlong.manager.dao.entity.StreamSinkEntity">
insert into stream_sink (id, inlong_group_id, inlong_stream_id,
- sink_type, sink_name, storage_period,
+ sink_type, sink_name, description,
enable_create_resource, ext_params,
operate_log, status, previous_status,
is_deleted, creator, modifier,
create_time, modify_time)
values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR},
#{inlongStreamId,jdbcType=VARCHAR},
- #{sinkType,jdbcType=VARCHAR}, #{sinkName,jdbcType=VARCHAR},
#{storagePeriod,jdbcType=INTEGER},
+ #{sinkType,jdbcType=VARCHAR}, #{sinkName,jdbcType=VARCHAR},
#{description,jdbcType=VARCHAR},
#{enableCreateResource,jdbcType=TINYINT},
#{extParams,jdbcType=VARCHAR},
#{operateLog,jdbcType=VARCHAR}, #{status,jdbcType=INTEGER},
#{previousStatus,jdbcType=INTEGER},
#{isDeleted,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR},
#{modifier,jdbcType=VARCHAR},
@@ -229,8 +227,8 @@
<if test="sinkName != null">
sink_name,
</if>
- <if test="storagePeriod != null">
- storage_period,
+ <if test="description != null">
+ description,
</if>
<if test="enableCreateResource != null">
enable_create_resource,
@@ -279,8 +277,8 @@
<if test="sinkName != null">
#{sinkName,jdbcType=VARCHAR},
</if>
- <if test="storagePeriod != null">
- #{storagePeriod,jdbcType=INTEGER},
+ <if test="description != null">
+ #{description,jdbcType=VARCHAR},
</if>
<if test="enableCreateResource != null">
#{enableCreateResource,jdbcType=TINYINT},
@@ -325,8 +323,8 @@
<if test="inlongStreamId != null">
inlong_stream_id = #{inlongStreamId,jdbcType=VARCHAR},
</if>
- <if test="storagePeriod != null">
- storage_period = #{storagePeriod,jdbcType=INTEGER},
+ <if test="description != null">
+ description = #{description,jdbcType=VARCHAR},
</if>
<if test="sinkType != null">
sink_type = #{sinkType,jdbcType=VARCHAR},
@@ -373,7 +371,7 @@
inlong_stream_id = #{inlongStreamId,jdbcType=VARCHAR},
sink_type = #{sinkType,jdbcType=VARCHAR},
sink_name = #{sinkName,jdbcType=VARCHAR},
- storage_period = #{storagePeriod,jdbcType=INTEGER},
+ description = #{description,jdbcType=VARCHAR},
enable_create_resource = #{enableCreateResource,jdbcType=TINYINT},
ext_params = #{extParams,jdbcType=VARCHAR},
operate_log = #{operateLog,jdbcType=VARCHAR},
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkFieldEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkFieldEntityMapper.xml
index 132c356bf..3b2f7ff43 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkFieldEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkFieldEntityMapper.xml
@@ -31,28 +31,41 @@
<result column="field_comment" jdbcType="VARCHAR"
property="fieldComment"/>
<result column="source_field_name" jdbcType="VARCHAR"
property="sourceFieldName"/>
<result column="source_field_type" jdbcType="VARCHAR"
property="sourceFieldType"/>
+ <result column="field_length" jdbcType="INTEGER"
property="fieldLength"/>
+ <result column="field_precision" jdbcType="INTEGER"
property="fieldPrecision"/>
+ <result column="field_scale" jdbcType="INTEGER" property="fieldScale"/>
+ <result column="partition_strategy" jdbcType="VARCHAR"
property="partitionStrategy"/>
<result column="is_meta_field" jdbcType="SMALLINT"
property="isMetaField"/>
<result column="field_format" jdbcType="VARCHAR"
property="fieldFormat"/>
<result column="rank_num" jdbcType="SMALLINT" property="rankNum"/>
<result column="is_deleted" jdbcType="INTEGER" property="isDeleted"/>
</resultMap>
<sql id="Base_Column_List">
- id, sink_id, field_name, field_type, field_comment,
- source_field_name, source_field_type, is_meta_field, field_format,
rank_num, is_deleted
+ id, sink_id, field_name, field_type, field_comment, source_field_name,
source_field_type,
+ field_length, field_precision, field_scale, partition_strategy,
is_meta_field, field_format,
+ rank_num, is_deleted
</sql>
<insert id="insert" useGeneratedKeys="true" keyProperty="id"
parameterType="org.apache.inlong.manager.dao.entity.StreamSinkFieldEntity">
- insert into stream_sink_field (id, inlong_group_id, inlong_stream_id,
- sink_id, sink_type, field_name,
- field_type, field_comment,
source_field_name,
- source_field_type, is_meta_field,
field_format,
+ insert into stream_sink_field (id, inlong_group_id,
+ inlong_stream_id, sink_id,
+ sink_type, field_name,
+ field_type, field_comment,
+ source_field_name, source_field_type,
+ field_length, field_precision,
+ field_scale, partition_strategy,
+ is_meta_field, field_format,
rank_num, is_deleted)
- values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR},
#{inlongStreamId,jdbcType=VARCHAR},
- #{sinkId,jdbcType=INTEGER}, #{sinkType,jdbcType=VARCHAR},
#{fieldName,jdbcType=VARCHAR},
- #{fieldType,jdbcType=VARCHAR},
#{fieldComment,jdbcType=VARCHAR}, #{sourceFieldName,jdbcType=VARCHAR},
- #{sourceFieldType,jdbcType=VARCHAR},
#{isMetaField,jdbcType=SMALLINT},
- #{fieldFormat,jdbcType=VARCHAR}, #{rankNum,jdbcType=SMALLINT},
#{isDeleted,jdbcType=INTEGER})
+ values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR},
+ #{inlongStreamId,jdbcType=VARCHAR}, #{sinkId,jdbcType=INTEGER},
+ #{sinkType,jdbcType=VARCHAR}, #{fieldName,jdbcType=VARCHAR},
+ #{fieldType,jdbcType=VARCHAR},
#{fieldComment,jdbcType=VARCHAR},
+ #{sourceFieldName,jdbcType=VARCHAR},
#{sourceFieldType,jdbcType=VARCHAR},
+ #{fieldLength,jdbcType=INTEGER},
#{fieldPrecision,jdbcType=INTEGER},
+ #{fieldScale,jdbcType=INTEGER},
#{partitionStrategy,jdbcType=VARCHAR},
+ #{isMetaField,jdbcType=SMALLINT},
#{fieldFormat,jdbcType=VARCHAR},
+ #{rankNum,jdbcType=SMALLINT}, #{isDeleted,jdbcType=INTEGER})
</insert>
<insert id="insertAll">
insert into stream_sink_field (
@@ -61,7 +74,10 @@
sink_type, field_name,
field_type, field_comment,
source_field_name, source_field_type,
- is_meta_field, field_format, rank_num, is_deleted
+ field_length, field_precision,
+ field_scale, partition_strategy,
+ is_meta_field, field_format,
+ rank_num, is_deleted
)
values
<foreach collection="list" index="index" item="item" separator=",">
@@ -71,9 +87,10 @@
#{item.sinkType,jdbcType=VARCHAR},
#{item.fieldName,jdbcType=VARCHAR},
#{item.fieldType,jdbcType=VARCHAR},
#{item.fieldComment,jdbcType=VARCHAR},
#{item.sourceFieldName,jdbcType=VARCHAR},
#{item.sourceFieldType,jdbcType=VARCHAR},
+ #{item.fieldLength,jdbcType=INTEGER},
#{item.fieldPrecision,jdbcType=INTEGER},
+ #{item.fieldScale,jdbcType=INTEGER},
#{item.partitionStrategy,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}
)
</foreach>
</insert>
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/hive/HiveStreamSinkOperation.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/hive/HiveStreamSinkOperation.java
index 8b3e6d869..92eea579c 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/hive/HiveStreamSinkOperation.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/hive/HiveStreamSinkOperation.java
@@ -109,7 +109,7 @@ public class HiveStreamSinkOperation implements
StreamSinkOperation {
@Override
public void saveFieldOpt(SinkRequest request) {
List<SinkFieldRequest> fieldList = request.getFieldList();
- LOGGER.info("begin to save field={}", fieldList);
+ LOGGER.info("begin to save hive field={}", fieldList);
if (CollectionUtils.isEmpty(fieldList)) {
return;
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/iceberg/IcebergStreamSinkOperation.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/iceberg/IcebergStreamSinkOperation.java
index 56a3d8ee5..284cbf7ad 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/iceberg/IcebergStreamSinkOperation.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/iceberg/IcebergStreamSinkOperation.java
@@ -103,7 +103,7 @@ public class IcebergStreamSinkOperation implements
StreamSinkOperation {
@Override
public void saveFieldOpt(SinkRequest request) {
List<SinkFieldRequest> fieldList = request.getFieldList();
- LOGGER.info("begin to save field={}", fieldList);
+ LOGGER.info("begin to save iceberg field={}", fieldList);
if (CollectionUtils.isEmpty(fieldList)) {
return;
}
@@ -128,7 +128,7 @@ public class IcebergStreamSinkOperation implements
StreamSinkOperation {
}
sinkFieldMapper.insertAll(entityList);
- LOGGER.info("success to save hive field");
+ LOGGER.info("success to save iceberg field");
}
@Override
diff --git
a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/IcebergStreamSinkServiceTest.java
b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/IcebergStreamSinkServiceTest.java
index 0eb91d8e2..b3ad88ca9 100644
---
a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/IcebergStreamSinkServiceTest.java
+++
b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/sink/IcebergStreamSinkServiceTest.java
@@ -43,13 +43,12 @@ public class IcebergStreamSinkServiceTest extends
ServiceBaseTest {
public Integer saveSink() {
streamServiceTest.saveInlongStream(globalGroupId, globalStreamId,
globalOperator);
-
IcebergSinkRequest sinkInfo = new IcebergSinkRequest();
sinkInfo.setInlongGroupId(globalGroupId);
sinkInfo.setInlongStreamId(globalStreamId);
sinkInfo.setSinkType(Constant.SINK_ICEBERG);
sinkInfo.setEnableCreateResource(Constant.DISABLE_CREATE_RESOURCE);
- sinkInfo.setTableLocation("hdfs://127.0.0.1:8020/data");
+ sinkInfo.setDataPath("hdfs://127.0.0.1:8020/data");
sinkInfo.setSinkName(sinkName);
return sinkService.save(sinkInfo, globalOperator);
}
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 220e5f6f6..052726244 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
@@ -489,7 +489,7 @@ CREATE TABLE `stream_sink`
`inlong_stream_id` varchar(256) NOT NULL COMMENT 'Owning inlong
stream id',
`sink_type` varchar(15) DEFAULT 'HIVE' COMMENT
'Sink type, including: HIVE, ES, etc',
`sink_name` varchar(128) NOT NULL DEFAULT '' COMMENT 'Sink
name',
- `storage_period` int(11) DEFAULT '10' COMMENT 'Data
storage period, unit: day',
+ `description` varchar(500) NULL COMMENT 'Sink description',
`enable_create_resource` tinyint(2) DEFAULT '1' COMMENT
'Whether to enable create sink resource? 0: disable, 1: enable. default is 1',
`ext_params` text COMMENT 'Another fields, will saved as JSON
type',
`operate_log` varchar(5000) DEFAULT NULL COMMENT
'Background operate log',
@@ -527,20 +527,24 @@ CREATE TABLE `stream_sink_ext`
DROP TABLE IF EXISTS `stream_sink_field`;
CREATE TABLE `stream_sink_field`
(
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT
'Incremental primary key',
- `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
- `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id',
- `sink_id` int(11) NOT NULL COMMENT 'Sink id',
- `sink_type` varchar(15) NOT NULL COMMENT 'Sink type',
- `source_field_name` varchar(50) DEFAULT NULL COMMENT 'Source field name',
- `source_field_type` varchar(50) DEFAULT NULL COMMENT 'Source field type',
- `field_name` varchar(50) NOT NULL COMMENT 'Field name',
- `field_type` varchar(50) NOT NULL COMMENT 'Field type',
- `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',
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT
'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
+ `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id',
+ `sink_id` int(11) NOT NULL COMMENT 'Sink id',
+ `sink_type` varchar(15) NOT NULL COMMENT 'Sink type',
+ `source_field_name` varchar(50) DEFAULT NULL COMMENT 'Source field
name',
+ `source_field_type` varchar(50) DEFAULT NULL COMMENT 'Source field
type',
+ `field_name` varchar(50) NOT NULL COMMENT 'Field name',
+ `field_type` varchar(50) NOT NULL COMMENT 'Field type',
+ `field_comment` varchar(2000) DEFAULT NULL COMMENT 'Field
description',
+ `field_length` int(4) DEFAULT NULL COMMENT 'Field length',
+ `field_precision` int(4) DEFAULT NULL COMMENT 'Field precision',
+ `field_scale` int(4) DEFAULT NULL COMMENT 'Field scale',
+ `partition_strategy` varchar(20) DEFAULT NULL COMMENT 'Field partition
strategy',
+ `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',
PRIMARY KEY (`id`)
);
diff --git a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
index 47848a947..987ec4820 100644
--- a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
@@ -514,7 +514,7 @@ CREATE TABLE `stream_sink`
`inlong_stream_id` varchar(256) NOT NULL COMMENT 'Owning inlong
stream id',
`sink_type` varchar(15) DEFAULT 'HIVE' COMMENT
'Sink type, including: HIVE, ES, etc',
`sink_name` varchar(128) NOT NULL DEFAULT '' COMMENT 'Sink
name',
- `storage_period` int(11) DEFAULT '10' COMMENT 'Data
storage period, unit: day',
+ `description` varchar(500) NULL COMMENT 'Sink description',
`enable_create_resource` tinyint(1) DEFAULT '1' COMMENT
'Whether to enable create sink resource? 0: disable, 1: enable. default is 1',
`ext_params` text NULL COMMENT 'Another fields, will
saved as JSON type',
`operate_log` text DEFAULT NULL COMMENT
'Background operate log',
@@ -554,20 +554,24 @@ CREATE TABLE `stream_sink_ext`
DROP TABLE IF EXISTS `stream_sink_field`;
CREATE TABLE `stream_sink_field`
(
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT
'Incremental primary key',
- `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
- `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id',
- `sink_id` int(11) NOT NULL COMMENT 'Sink id',
- `sink_type` varchar(15) NOT NULL COMMENT 'Sink type',
- `source_field_name` varchar(50) DEFAULT NULL COMMENT 'Source field name',
- `source_field_type` varchar(50) DEFAULT NULL COMMENT 'Source field type',
- `field_name` varchar(50) NOT NULL COMMENT 'Field name',
- `field_type` varchar(50) NOT NULL COMMENT 'Field type',
- `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',
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT
'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
+ `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id',
+ `sink_id` int(11) NOT NULL COMMENT 'Sink id',
+ `sink_type` varchar(15) NOT NULL COMMENT 'Sink type',
+ `source_field_name` varchar(50) DEFAULT NULL COMMENT 'Source field
name',
+ `source_field_type` varchar(50) DEFAULT NULL COMMENT 'Source field
type',
+ `field_name` varchar(50) NOT NULL COMMENT 'Field name',
+ `field_type` varchar(50) NOT NULL COMMENT 'Field type',
+ `field_comment` varchar(2000) DEFAULT NULL COMMENT 'Field
description',
+ `field_length` int(4) DEFAULT NULL COMMENT 'Field length',
+ `field_precision` int(4) DEFAULT NULL COMMENT 'Field precision',
+ `field_scale` int(4) DEFAULT NULL COMMENT 'Field scale',
+ `partition_strategy` varchar(20) DEFAULT NULL COMMENT 'Field partition
strategy',
+ `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',
PRIMARY KEY (`id`)
) ENGINE = InnoDB
DEFAULT CHARSET = utf8mb4 COMMENT ='Stream sink field table';
