support hive

build.gradle

@@ -37,6 +37,7 @@ ext {
     commonsCollectionsVersion = "3.2.2"
     opencsvVersion = "5.9"
     poiVersion = "5.3.0"
+    hiveVersion = "2.3.10"
 
     httpClientVersion = "4.5.14"
     yitterIdGeneratorVersion = "1.0.6"
@@ -128,6 +129,7 @@ allprojects {
         exclude group: "org.springframework.boot", module: "spring-boot-starter-logging"
         exclude group: "org.slf4j", module: "log4j-over-slf4j"
         exclude group: "org.slf4j", module: "slf4j-log4j12"
+        exclude group: "ch.qos.logback", module: "logback-classic"
         exclude group: "org.apache.logging.log4j", module: "log4j-to-slf4j"
     }
 

wedpr-components/dataset/build.gradle

@@ -15,6 +15,12 @@ dependencies {
     compile project(":wedpr-components-storage")
     compile project(":wedpr-components-authorization")
 
+    // hive
+    compile("org.apache.hive:hive-jdbc:${hiveVersion}") {
+        exclude group: "org.apache.zookeeper", module: "zookeeper4"
+        exclude group: "org.eclipse.jetty.aggregate", module: "jetty-all"
+    }
+
     compile "org.apache.poi:poi:${poiVersion}"
     compile "org.apache.poi:poi-excelant:${poiVersion}"
     compile "com.opencsv:opencsv:${opencsvVersion}"

wedpr-components/dataset/src/main/java/com/webank/wedpr/components/dataset/config/HiveConfig.java

@@ -0,0 +1,19 @@
+package com.webank.wedpr.components.dataset.config;
+
+import lombok.Getter;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@Getter
+public class HiveConfig {
+
+    @Value("${wedpr.hive.jdbc.url:}")
+    String hiveJdbcUrl;
+
+    @Value("${wedpr.hive.user:}")
+    String hiveUserName;
+
+    @Value("${wedpr.hive.password:}")
+    String hiveUserPassword;
+}

wedpr-components/dataset/src/main/java/com/webank/wedpr/components/dataset/controller/DatasetController.java

@@ -400,14 +400,14 @@ public WeDPRResponse updateDatasetList(
 
             for (UpdateDatasetRequest updateDatasetRequest : updateDatasetRequestList) {
                 String datasetTitle = updateDatasetRequest.getDatasetTitle();
-                String datasetLabel = updateDatasetRequest.getDatasetLabel();
+                // String datasetLabel = updateDatasetRequest.getDatasetLabel();
                 String datasetDesc = updateDatasetRequest.getDatasetDesc();
                 String datasetId = updateDatasetRequest.getDatasetId();
                 String approvalChain = updateDatasetRequest.getApprovalChain();
 
                 Common.requireNonEmpty("datasetId", datasetId);
                 Common.requireNonEmpty("datasetTitle", datasetTitle);
-                Common.requireNonEmpty("datasetLabel", datasetLabel);
+                // Common.requireNonEmpty("datasetLabel", datasetLabel);
                 Common.requireNonEmpty("datasetDesc", datasetDesc);
 
                 Integer datasetVisibility = updateDatasetRequest.getDatasetVisibility();

wedpr-components/dataset/src/main/java/com/webank/wedpr/components/dataset/datasource/processor/DBDataSourceProcessor.java

@@ -6,6 +6,7 @@
 import com.webank.wedpr.components.dataset.datasource.DBType;
 import com.webank.wedpr.components.dataset.datasource.DataSourceMeta;
 import com.webank.wedpr.components.dataset.datasource.category.DBDataSource;
+import com.webank.wedpr.components.dataset.sqlutils.SQLExecutor;
 import com.webank.wedpr.components.dataset.sqlutils.SQLUtils;
 import com.webank.wedpr.components.dataset.utils.CsvUtils;
 import com.webank.wedpr.components.dataset.utils.FileUtils;
@@ -88,7 +89,20 @@ public void prepareData() throws DatasetException {
         String datasetBaseDir = datasetConfig.getDatasetBaseDir();
         String cvsFilePath = datasetBaseDir + File.separator + datasetId;
 
-        CsvUtils.convertDBDataToCsv(dbType, dbDataSource, cvsFilePath);
+        String jdbcUrl =
+                SQLExecutor.generateJdbcUrl(
+                        dbType,
+                        dbDataSource.getDbIp(),
+                        dbDataSource.getDbPort(),
+                        dbDataSource.getDatabase(),
+                        null);
+
+        CsvUtils.convertDBDataToCsv(
+                jdbcUrl,
+                dbDataSource.getUserName(),
+                dbDataSource.getPassword(),
+                dbDataSource.getSql(),
+                cvsFilePath);
 
         dataSourceProcessorContext.setCvsFilePath(cvsFilePath);
 

wedpr-components/dataset/src/main/java/com/webank/wedpr/components/dataset/datasource/processor/DataSourceProcessorContext.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.webank.wedpr.components.dataset.config.DatasetConfig;
+import com.webank.wedpr.components.dataset.config.HiveConfig;
 import com.webank.wedpr.components.dataset.datasource.DataSourceMeta;
 import com.webank.wedpr.components.dataset.service.ChunkUploadApi;
 import com.webank.wedpr.components.db.mapper.dataset.dao.Dataset;
@@ -24,6 +25,7 @@ public class DataSourceProcessorContext {
     private FileStorageInterface.FilePermissionInfo filePermissionInfo;
     private DatasetTransactionalWrapper datasetTransactionalWrapper;
     private DatasetConfig datasetConfig;
+    private HiveConfig hiveConfig;
     private UserInfo userInfo;
 
     // intermediate state

wedpr-components/dataset/src/main/java/com/webank/wedpr/components/dataset/datasource/processor/HiveDataSourceProcessor.java

@@ -1,24 +1,83 @@
 package com.webank.wedpr.components.dataset.datasource.processor;
 
+import com.webank.wedpr.common.utils.Common;
+import com.webank.wedpr.components.dataset.config.DatasetConfig;
+import com.webank.wedpr.components.dataset.config.HiveConfig;
 import com.webank.wedpr.components.dataset.datasource.DataSourceMeta;
+import com.webank.wedpr.components.dataset.datasource.category.HiveDataSource;
+import com.webank.wedpr.components.dataset.sqlutils.SQLUtils;
+import com.webank.wedpr.components.dataset.utils.CsvUtils;
+import com.webank.wedpr.components.dataset.utils.JsonUtils;
+import com.webank.wedpr.components.db.mapper.dataset.dao.Dataset;
 import com.webank.wedpr.components.db.mapper.dataset.exception.DatasetException;
+import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-public class HiveDataSourceProcessor implements DataSourceProcessor {
+public class HiveDataSourceProcessor extends DBDataSourceProcessor {
+
+    private static final Logger logger = LoggerFactory.getLogger(HiveDataSourceProcessor.class);
 
     @Override
     public DataSourceMeta parseDataSourceMeta(String strDataSourceMeta) throws DatasetException {
-        return null;
+        long startTimeMillis = System.currentTimeMillis();
+
+        HiveDataSource hiveDataSource =
+                (HiveDataSource)
+                        JsonUtils.jsonString2Object(strDataSourceMeta, HiveDataSource.class);
+
+        String sql = hiveDataSource.getSql();
+        Common.requireNonEmpty("sql", sql);
+        Boolean dynamicDataSource = hiveDataSource.getDynamicDataSource();
+        if (dynamicDataSource != null && dynamicDataSource) {
+            hiveDataSource.setDynamicDataSource(true);
+        }
+
+        // check if single select
+        SQLUtils.isSingleSelectStatement(sql);
+
+        long endTimeMillis = System.currentTimeMillis();
+        logger.info(
+                " => data source processor stage parse data source meta end, dbDataSource: {}, cost(ms): {}",
+                hiveDataSource,
+                endTimeMillis - startTimeMillis);
+
+        return hiveDataSource;
     }
 
     @Override
-    public void prepareData() throws DatasetException {}
+    public void prepareData() throws DatasetException {
+        long startTimeMillis = System.currentTimeMillis();
 
-    @Override
-    public void analyzeData() throws DatasetException {}
+        DatasetConfig datasetConfig = dataSourceProcessorContext.getDatasetConfig();
+        HiveConfig hiveConfig = dataSourceProcessorContext.getHiveConfig();
+        HiveDataSource dataSourceMeta =
+                (HiveDataSource) dataSourceProcessorContext.getDataSourceMeta();
 
-    @Override
-    public void uploadData() throws DatasetException {}
+        Dataset dataset = dataSourceProcessorContext.getDataset();
+        String datasetId = dataset.getDatasetId();
 
-    @Override
-    public void cleanupData() throws DatasetException {}
+        String datasetBaseDir = datasetConfig.getDatasetBaseDir();
+        String cvsFilePath = datasetBaseDir + File.separator + datasetId;
+
+        String hiveJdbcUrl = hiveConfig.getHiveJdbcUrl();
+        String hiveJdbcUserName = hiveConfig.getHiveUserName();
+        String hiveJdbcUserPassword = hiveConfig.getHiveUserPassword();
+
+        CsvUtils.convertDBDataToCsv(
+                hiveJdbcUrl,
+                hiveJdbcUserName,
+                hiveJdbcUserPassword,
+                dataSourceMeta.getSql(),
+                cvsFilePath);
+
+        dataSourceProcessorContext.setCvsFilePath(cvsFilePath);
+
+        long endTimeMillis = System.currentTimeMillis();
+        logger.info(
+                " ==> data source processor stage prepare data end, datasetId: {}, cvsFilePath: {}, cost(ms): {}",
+                datasetId,
+                cvsFilePath,
+                endTimeMillis - startTimeMillis);
+    }
 }

wedpr-components/dataset/src/main/java/com/webank/wedpr/components/dataset/service/DatasetServiceImpl.java

@@ -5,6 +5,7 @@
 import com.github.pagehelper.page.PageMethod;
 import com.webank.wedpr.components.dataset.common.DatasetStatus;
 import com.webank.wedpr.components.dataset.config.DatasetConfig;
+import com.webank.wedpr.components.dataset.config.HiveConfig;
 import com.webank.wedpr.components.dataset.datasource.DataSourceMeta;
 import com.webank.wedpr.components.dataset.datasource.dispatch.DataSourceProcessorDispatcher;
 import com.webank.wedpr.components.dataset.datasource.processor.DataSourceProcessor;
@@ -52,6 +53,7 @@ public class DatasetServiceImpl implements DatasetServiceApi {
 
     private static final Logger logger = LoggerFactory.getLogger(DatasetServiceImpl.class);
 
+    @Autowired private HiveConfig hiveConfig;
     @Autowired private DatasetConfig datasetConfig;
     @Autowired private DatasetMapper datasetMapper;
     @Autowired private DatasetPermissionMapper datasetPermissionMapper;
@@ -214,6 +216,7 @@ public CreateDatasetResponse createDataset(
                             DataSourceProcessorContext.builder()
                                     .dataset(dataset)
                                     .dataSourceMeta(dataSourceMeta)
+                                    .hiveConfig(hiveConfig)
                                     .datasetConfig(datasetConfig)
                                     .userInfo(userInfo)
                                     .datasetTransactionalWrapper(datasetTransactionalWrapper)