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 2cc8532 [INLONG-713] add partition filed when create hive table (#543)
2cc8532 is described below
commit 2cc853278d1f7995d22cff2e26b92b50e80fbb55
Author: healchow <[email protected]>
AuthorDate: Sat Jul 10 22:53:55 2021 +0800
[INLONG-713] add partition filed when create hive table (#543)
Co-authored-by: healzhou <[email protected]>
---
.../service/core/impl/HiveSourceServiceImpl.java | 18 ++++++++----------
.../service/thirdpart/hive/HiveTableOperator.java | 22 +++++++++++++++++++---
2 files changed, 27 insertions(+), 13 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/HiveSourceServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/HiveSourceServiceImpl.java
index d87e6f9..6a9ca0d 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/HiveSourceServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/HiveSourceServiceImpl.java
@@ -72,7 +72,7 @@ public class HiveSourceServiceImpl implements
DataSourceService<DatabaseQueryBea
public void createDb(HiveTableQueryBean bean) throws Exception {
HiveCreateDbSqlBuilder builder = new HiveCreateDbSqlBuilder();
String createDbSql = builder.buildDDL(bean);
- LOGGER.info("create database sql=" + createDbSql);
+ LOGGER.info("create database sql={}", createDbSql);
hiveServerDao.executeDDL(createDbSql, bean.getJdbcUrl(),
bean.getUsername(), bean.getPassword());
}
@@ -80,7 +80,7 @@ public class HiveSourceServiceImpl implements
DataSourceService<DatabaseQueryBea
public void dropDb(DatabaseQueryBean bean) throws Exception {
HiveDropDbSqlBuilder builder = new HiveDropDbSqlBuilder();
String dropDbSql = builder.buildDDL(bean);
- LOGGER.info("drop database sql=" + dropDbSql);
+ LOGGER.info("drop database sql={}", dropDbSql);
hiveServerDao.executeDDL(dropDbSql, bean.getJdbcUrl(),
bean.getUserName(), bean.getPassword());
}
@@ -88,16 +88,15 @@ public class HiveSourceServiceImpl implements
DataSourceService<DatabaseQueryBea
public void createTable(HiveTableQueryBean bean) throws Exception {
HiveTableCreateSqlBuilder builder = new HiveTableCreateSqlBuilder();
String createTableSql = builder.buildDDL(bean);
- LOGGER.info("create table sql=" + createTableSql);
- hiveServerDao.executeDDL(createTableSql, bean.getJdbcUrl(),
- bean.getUsername(), bean.getPassword());
+ LOGGER.info("create table sql={}", createTableSql);
+ hiveServerDao.executeDDL(createTableSql, bean.getJdbcUrl(),
bean.getUsername(), bean.getPassword());
}
@Override
public void dropTable(HiveTableQueryBean bean) throws Exception {
HiveDropTableSqlBuilder builder = new HiveDropTableSqlBuilder();
String dropTableSql = builder.buildDDL(bean);
- LOGGER.info("drop table sql=" + dropTableSql);
+ LOGGER.info("drop table sql={}", dropTableSql);
hiveServerDao.executeDDL(dropTableSql, bean.getJdbcUrl(),
bean.getUsername(), bean.getPassword());
}
@@ -114,7 +113,7 @@ public class HiveSourceServiceImpl implements
DataSourceService<DatabaseQueryBea
}
HiveAddColumnSqlBuilder builder = new HiveAddColumnSqlBuilder();
String addColumnSql = builder.buildDDL(bean);
- LOGGER.info("add table Column sql=" + addColumnSql);
+ LOGGER.info("add table Column sql={}", addColumnSql);
hiveServerDao.executeDDL(addColumnSql, bean.getJdbcUrl(),
bean.getUsername(), bean.getPassword());
}
@@ -138,7 +137,7 @@ public class HiveSourceServiceImpl implements
DataSourceService<DatabaseQueryBea
HiveChangeColumnBuilder builder = new HiveChangeColumnBuilder();
String updateColumnSql = builder.buildDDL(tableQueryBean);
- LOGGER.info("update table Column sql=" + updateColumnSql);
+ LOGGER.info("update table Column sql={}", updateColumnSql);
hiveServerDao.executeDDL(updateColumnSql, tableQueryBean.getJdbcUrl(),
tableQueryBean.getUsername(), tableQueryBean.getPassword());
}
@@ -156,11 +155,10 @@ public class HiveSourceServiceImpl implements
DataSourceService<DatabaseQueryBea
HiveQueryTableSqlBuilder tableBuilder = new HiveQueryTableSqlBuilder();
String querySql = tableBuilder.buildDDL(queryBean);
- LOGGER.info("query sql=" + querySql);
+ LOGGER.info("query sql={}", querySql);
if (querySql != null && hiveServerDao.isExistTable(jdbcUrl, username,
password,
queryBean.getDbName(), queryBean.getTableName())) {
-
return hiveServerDao.queryStructure(querySql, jdbcUrl, username,
password);
}
return Collections.emptyList();
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 d2a4b8f..8be5502 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
@@ -88,10 +88,13 @@ public class HiveTableOperator {
}
protected HiveTableQueryBean getTableQueryBean(DataStreamInfoToHiveConfig
hiveConfig) {
- List<StorageHiveFieldEntity> fieldEntities = storageHiveFieldMapper
- .queryHiveFieldsByStreamId(hiveConfig.getBusinessIdentifier(),
hiveConfig.getDataStreamIdentifier());
- List<HiveColumnQueryBean> columnQueryBeans = new ArrayList<>();
+ String bid = hiveConfig.getBusinessIdentifier();
+ String dsid = hiveConfig.getDataStreamIdentifier();
+ log.info("begin to get table query bean for bid={}, dsid={}", bid,
dsid);
+
+ List<StorageHiveFieldEntity> fieldEntities =
storageHiveFieldMapper.queryHiveFieldsByStreamId(bid, dsid);
+ List<HiveColumnQueryBean> columnQueryBeans = new ArrayList<>();
for (StorageHiveFieldEntity field : fieldEntities) {
HiveColumnQueryBean columnBean = new HiveColumnQueryBean();
columnBean.setColumnName(field.getFieldName());
@@ -100,6 +103,16 @@ public class HiveTableOperator {
columnQueryBeans.add(columnBean);
}
+ // set partition field and type
+ String partitionField = hiveConfig.getPrimaryPartition();
+ if (partitionField != null) {
+ HiveColumnQueryBean partColumn = new HiveColumnQueryBean();
+ partColumn.setPartition(true);
+ partColumn.setColumnName(partitionField);
+ partColumn.setColumnType("string"); // currently only supports
'string' type
+ columnQueryBeans.add(partColumn);
+ }
+
HiveTableQueryBean queryBean = new HiveTableQueryBean();
queryBean.setColumns(columnQueryBeans);
queryBean.setUsername(hiveConfig.getUserName());
@@ -108,6 +121,9 @@ public class HiveTableOperator {
queryBean.setDbName(hiveConfig.getDbName());
queryBean.setJdbcUrl(hiveConfig.getJdbcUrl());
+ if (log.isDebugEnabled()) {
+ log.debug("success to get table query bean={}", queryBean);
+ }
return queryBean;
}
}
