This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 3c18efd [INLONG-716] set terminated symbol when create hive table
3c18efd is described below
commit 3c18efd77fc6811735596f6e87ff287e36e05fd4
Author: healzhou <[email protected]>
AuthorDate: Sun Jul 11 00:52:40 2021 +0800
[INLONG-716] set terminated symbol when create hive table
---
.../manager/service/core/impl/StorageHiveOperation.java | 14 ++++++++++++++
.../manager/service/thirdpart/hive/HiveTableOperator.java | 6 ++++++
.../thirdpart/hive/builder/HiveTableCreateSqlBuilder.java | 7 +++++--
.../thirdpart/sort/PushHiveConfigToSortEventListener.java | 9 ++++-----
4 files changed, 29 insertions(+), 7 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/StorageHiveOperation.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/StorageHiveOperation.java
index 9bdd6bb..2b79351 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/StorageHiveOperation.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/StorageHiveOperation.java
@@ -20,6 +20,7 @@ package org.apache.inlong.manager.service.core.impl;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
+import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
@@ -36,9 +37,11 @@ import
org.apache.inlong.manager.common.pojo.datastorage.StorageHiveListVO;
import org.apache.inlong.manager.common.pojo.datastorage.StoragePageRequest;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.DataStreamEntity;
import org.apache.inlong.manager.dao.entity.StorageExtEntity;
import org.apache.inlong.manager.dao.entity.StorageHiveEntity;
import org.apache.inlong.manager.dao.entity.StorageHiveFieldEntity;
+import org.apache.inlong.manager.dao.mapper.DataStreamEntityMapper;
import org.apache.inlong.manager.dao.mapper.StorageExtEntityMapper;
import org.apache.inlong.manager.dao.mapper.StorageHiveEntityMapper;
import org.apache.inlong.manager.dao.mapper.StorageHiveFieldEntityMapper;
@@ -62,6 +65,8 @@ public class StorageHiveOperation extends
StorageBaseOperation {
private StorageExtEntityMapper storageExtMapper;
@Autowired
private StorageHiveFieldEntityMapper hiveFieldMapper;
+ @Autowired
+ private DataStreamEntityMapper dataStreamMapper;
/**
* Save HIVE storage information
@@ -80,6 +85,15 @@ public class StorageHiveOperation extends
StorageBaseOperation {
StorageHiveInfo hiveInfo = (StorageHiveInfo) storageInfo;
StorageHiveEntity entity = CommonBeanUtils.copyProperties(hiveInfo,
StorageHiveEntity::new);
+ // Set the encoding type and field splitter
+ DataStreamEntity streamEntity =
dataStreamMapper.selectByIdentifier(bid, entity.getDataStreamIdentifier());
+ String encodingType = streamEntity.getDataEncoding() == null
+ ? StandardCharsets.UTF_8.displayName() :
streamEntity.getDataEncoding();
+ entity.setEncodingType(encodingType);
+ if (entity.getFieldSplitter() == null) {
+ entity.setFieldSplitter(streamEntity.getFileDelimiter());
+ }
+
entity.setStatus(EntityStatus.DATA_STORAGE_NEW.getCode());
entity.setCreator(operator);
entity.setModifier(operator);
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/HiveTableOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/HiveTableOperator.java
index 8be5502..d3313bc 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/HiveTableOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/HiveTableOperator.java
@@ -33,6 +33,7 @@ import
org.apache.inlong.manager.dao.mapper.StorageHiveFieldEntityMapper;
import org.apache.inlong.manager.service.core.DataSourceService;
import org.apache.inlong.manager.service.core.StorageService;
import org.apache.inlong.manager.workflow.exception.WorkflowException;
+import org.apache.inlong.sort.protocol.sink.HiveSinkInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -115,6 +116,11 @@ public class HiveTableOperator {
HiveTableQueryBean queryBean = new HiveTableQueryBean();
queryBean.setColumns(columnQueryBeans);
+ // set terminated symbol
+ if (hiveConfig.getFieldSplitter() != null) {
+ char ch = (char) Integer.parseInt(hiveConfig.getFieldSplitter());
+ queryBean.setFieldTerSymbol(String.valueOf(ch));
+ }
queryBean.setUsername(hiveConfig.getUserName());
queryBean.setPassword(hiveConfig.getPassword());
queryBean.setTableName(hiveConfig.getTableName());
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/builder/HiveTableCreateSqlBuilder.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/builder/HiveTableCreateSqlBuilder.java
index 3282ad4..d04515a 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/builder/HiveTableCreateSqlBuilder.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/hive/builder/HiveTableCreateSqlBuilder.java
@@ -32,10 +32,14 @@ public class HiveTableCreateSqlBuilder extends
SqlBuilder<HiveTableQueryBean> {
ddl.append("CREATE TABLE ").append(dbTableName);
// Construct columns and partition columns
ddl.append(this.buildColumnsAndComments(table.getColumns(),
table.getTableDesc()));
+ // Set TERMINATED symbol
+ if (table.getFieldTerSymbol() != null) {
+ ddl.append(" ROW FORMAT DELIMITED FIELDS TERMINATED BY
'").append(table.getFieldTerSymbol()).append("'");
+ }
return ddl.toString();
}
- // (col_name data_type [COMMENT col_comment],col_name data_type [COMMENT
col_comment]....)
+ // (col_name data_type [COMMENT col_comment], col_name data_type [COMMENT
col_comment]....)
private String buildColumnsAndComments(List<HiveColumnQueryBean> columns,
String tableComment) {
List<String> columnInfoList = new ArrayList<>();
List<String> partitionList = new ArrayList<>();
@@ -69,5 +73,4 @@ public class HiveTableCreateSqlBuilder extends
SqlBuilder<HiveTableQueryBean> {
return "CREATE_TABLE_HIVE";
}
-
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/sort/PushHiveConfigToSortEventListener.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/sort/PushHiveConfigToSortEventListener.java
index 4b35ea7..a7831b5 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/sort/PushHiveConfigToSortEventListener.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/thirdpart/sort/PushHiveConfigToSortEventListener.java
@@ -146,12 +146,9 @@ public class PushHiveConfigToSortEventListener implements
TaskEventListener {
.toString();
// Encapsulate the deserialization information in the source
- DataStreamInfo dataStream =
dataStreamService.get(hiveStorage.getBusinessIdentifier(),
- hiveStorage.getDataStreamIdentifier());
-
HiveSinkInfo.HiveFileFormat fileFormat = new
HiveSinkInfo.TextFileFormat(',');
- if (dataStream.getFileDelimiter() != null) {
- char c = (char) Integer.parseInt(dataStream.getFileDelimiter());
+ if (hiveStorage.getFieldSplitter() != null) {
+ char c = (char) Integer.parseInt(hiveStorage.getFieldSplitter());
fileFormat = new HiveSinkInfo.TextFileFormat(c);
}
@@ -170,6 +167,8 @@ public class PushHiveConfigToSortEventListener implements
TaskEventListener {
);
// data stream fields
+ DataStreamInfo dataStream =
dataStreamService.get(hiveStorage.getBusinessIdentifier(),
+ hiveStorage.getDataStreamIdentifier());
Stream<FieldInfo> streamFields =
dataStream.getFieldList().stream().map(field -> {
FormatInfo formatInfo =
SortFieldFormatUtils.convertFieldFormat(field.getFieldType().toLowerCase());
return new FieldInfo(field.getFieldName(), formatInfo);
