Hive support multiple table

Author: ruanwenjun

Diff for: docs/en/connector-v2/sink/Hive.md

@@ -47,7 +47,7 @@ By default, we use 2PC commit to ensure `exactly-once`
 
 ### table_name [string]
 
-Target Hive table name eg: db1.table1
+Target Hive table name eg: db1.table1, and if the source is multiple mode, you can use `${database_name}.${table_name}` to generate the table name, it will replace the `${database_name}` and `${table_name}` with the value of the CatalogTable generate from the source.
 
 ### metastore_uri [string]
 
@@ -343,6 +343,56 @@ sink {
 }
 ```
 
+### example 2
+
+We have multiple source table like this:
+
+```bash
+create table test_1(
+)
+PARTITIONED BY (xx);
+
+create table test_2(
+)
+PARTITIONED BY (xx);
+...
+```
+
+We need read data from these source tables and write to another tables:
+
+The job config file can like this:
+
+```
+env {
+  # You can set flink configuration here
+  parallelism = 3
+  job.name="test_hive_source_to_hive"
+}
+
+source {
+  Hive {
+    tables_configs = [
+      {
+        table_name = "test_hive.test_1"
+        metastore_uri = "thrift://ctyun6:9083"
+      },
+      {
+        table_name = "test_hive.test_2"
+        metastore_uri = "thrift://ctyun7:9083"
+      }
+    ]
+  }
+}
+
+sink {
+  # choose stdout output plugin to output data to console
+  Hive {
+    table_name = "${database_name}.${table_name}"
+    metastore_uri = "thrift://ctyun7:9083"
+  }
+}
+```
+
 ## Changelog
 
 ### 2.2.0-beta 2022-09-26

Diff for: docs/en/connector-v2/source/Hive.md

@@ -59,10 +59,6 @@ Hive metastore uri
 
 The path of `hdfs-site.xml`, used to load ha configuration of namenodes
 
-### hive_site_path [string]
-
-The path of `hive-site.xml`, used to authentication hive metastore
-
 ### read_partitions [list]
 
 The target partitions that user want to read from hive table, if user does not set this parameter, it will read all the data from hive table.
@@ -102,6 +98,8 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 ## Example
 
+### Example 1: Single table
+
 ```bash
 
   Hive {
@@ -111,6 +109,25 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 ```
 
+### Example 2: Multiple tables
+
+```bash
+
+  Hive {
+    tables_configs = [
+        {
+          table_name = "default.seatunnel_orc_1"
+          metastore_uri = "thrift://namenode001:9083"
+        },
+        {
+          table_name = "default.seatunnel_orc_2"
+          metastore_uri = "thrift://namenode001:9083"
+        }
+    ]
+  }
+
+```
+
 ## Changelog
 
 ### 2.2.0-beta 2022-09-26

Diff for: seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java

@@ -42,6 +42,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /** Utils contains some common methods for construct CatalogTable. */
 @Slf4j
@@ -234,4 +236,41 @@ public static SeaTunnelRowType buildSimpleTextSchema() {
     public static CatalogTable buildSimpleTextTable() {
         return getCatalogTable("default", buildSimpleTextSchema());
     }
+
+    public static CatalogTable newCatalogTable(
+            CatalogTable catalogTable, SeaTunnelRowType seaTunnelRowType) {
+        TableSchema tableSchema = catalogTable.getTableSchema();
+
+        Map<String, Column> columnMap =
+                tableSchema.getColumns().stream()
+                        .collect(Collectors.toMap(Column::getName, Function.identity()));
+        String[] fieldNames = seaTunnelRowType.getFieldNames();
+        SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes();
+
+        List<Column> finalColumns = new ArrayList<>();
+        for (int i = 0; i < fieldNames.length; i++) {
+            Column column = columnMap.get(fieldNames[i]);
+            if (column != null) {
+                finalColumns.add(column);
+            } else {
+                finalColumns.add(
+                        PhysicalColumn.of(fieldNames[i], fieldTypes[i], 0, false, null, null));
+            }
+        }
+
+        TableSchema finalSchema =
+                TableSchema.builder()
+                        .columns(finalColumns)
+                        .primaryKey(tableSchema.getPrimaryKey())
+                        .constraintKey(tableSchema.getConstraintKeys())
+                        .build();
+
+        return CatalogTable.of(
+                catalogTable.getTableId(),
+                finalSchema,
+                catalogTable.getOptions(),
+                catalogTable.getPartitionKeys(),
+                catalogTable.getComment(),
+                catalogTable.getCatalogName());
+    }
 }

Diff for: seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/config/LocalFileSourceConfig.java

@@ -18,13 +18,20 @@
 package org.apache.seatunnel.connectors.seatunnel.file.local.source.config;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
-import org.apache.seatunnel.connectors.seatunnel.file.config.BaseFileSourceConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
 import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType;
 import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
 import org.apache.seatunnel.connectors.seatunnel.file.local.config.LocalFileHadoopConf;
 
 import lombok.Getter;
 
+import java.io.Serializable;
+import java.util.List;
+
 @Getter
 public class LocalFileSourceConfig extends BaseFileSourceConfig {
 
@@ -41,6 +48,77 @@ public String getPluginName() {
     }
 
     public LocalFileSourceConfig(ReadonlyConfig readonlyConfig) {
-        super(readonlyConfig);
+        validateConfig(readonlyConfig);
+        this.fileFormat = readonlyConfig.get(LocalFileSourceOptions.FILE_FORMAT_TYPE);
+        this.localFileHadoopConf = new LocalFileHadoopConf();
+        this.readStrategy = ReadStrategyFactory.of(readonlyConfig, localFileHadoopConf);
+        this.filePaths = parseFilePaths(readonlyConfig);
+        this.catalogTable = parseCatalogTable(readonlyConfig);
+    }
+
+    private void validateConfig(ReadonlyConfig readonlyConfig) {
+        if (!readonlyConfig.getOptional(LocalFileSourceOptions.FILE_PATH).isPresent()) {
+            throw new FileConnectorException(
+                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                    String.format(
+                            "PluginName: %s, PluginType: %s, Message: %s",
+                            FileSystemType.LOCAL.getFileSystemPluginName(),
+                            PluginType.SOURCE,
+                            LocalFileSourceOptions.FILE_PATH + " is required"));
+        }
+        if (!readonlyConfig.getOptional(LocalFileSourceOptions.FILE_FORMAT_TYPE).isPresent()) {
+            throw new FileConnectorException(
+                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                    String.format(
+                            "PluginName: %s, PluginType: %s, Message: %s",
+                            FileSystemType.LOCAL.getFileSystemPluginName(),
+                            PluginType.SOURCE,
+                            LocalFileSourceOptions.FILE_FORMAT_TYPE.key() + " is required"));
+        }
+    }
+
+    private List<String> parseFilePaths(ReadonlyConfig readonlyConfig) {
+        String rootPath = null;
+        try {
+            rootPath = readonlyConfig.get(LocalFileSourceOptions.FILE_PATH);
+            return readStrategy.getFileNamesByPath(localFileHadoopConf, rootPath);
+        } catch (Exception ex) {
+            String errorMsg = String.format("Get file list from this path [%s] failed", rootPath);
+            throw new FileConnectorException(
+                    FileConnectorErrorCode.FILE_LIST_GET_FAILED, errorMsg, ex);
+        }
+    }
+
+    private CatalogTable parseCatalogTable(ReadonlyConfig readonlyConfig) {
+        final CatalogTable catalogTable;
+        if (readonlyConfig.getOptional(TableSchemaOptions.SCHEMA).isPresent()) {
+            catalogTable =
+                    CatalogTableUtil.buildWithConfig(
+                            FileSystemType.LOCAL.getFileSystemPluginName(), readonlyConfig);
+        } else {
+            catalogTable = CatalogTableUtil.buildSimpleTextTable();
+        }
+        if (CollectionUtils.isEmpty(filePaths)) {
+            return catalogTable;
+        }
+        switch (fileFormat) {
+            case CSV:
+            case TEXT:
+            case JSON:
+            case EXCEL:
+                readStrategy.setSeaTunnelRowTypeInfo(catalogTable.getSeaTunnelRowType());
+                return CatalogTableUtil.newCatalogTable(
+                        catalogTable, readStrategy.getActualSeaTunnelRowTypeInfo());
+            case ORC:
+            case PARQUET:
+                return CatalogTableUtil.newCatalogTable(
+                        catalogTable,
+                        readStrategy.getSeaTunnelRowTypeInfo(
+                                localFileHadoopConf, filePaths.get(0)));
+            default:
+                throw new FileConnectorException(
+                        FileConnectorErrorCode.FORMAT_NOT_SUPPORT,
+                        "SeaTunnel does not supported this file format: [" + fileFormat + "]");
+        }
     }
 }

Diff for: seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/commit/HiveSinkAggregatedCommitter.java

@@ -17,13 +17,15 @@
 
 package org.apache.seatunnel.connectors.seatunnel.hive.commit;
 
-import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
-import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileAggregatedCommitInfo;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileSinkAggregatedCommitter;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.util.FileSystemUtils;
+import org.apache.seatunnel.connectors.seatunnel.hive.sink.HiveSinkOptions;
 import org.apache.seatunnel.connectors.seatunnel.hive.utils.HiveMetaStoreProxy;
 
+import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
+import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.thrift.TException;
 
 import lombok.extern.slf4j.Slf4j;
@@ -33,33 +35,31 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import static org.apache.seatunnel.connectors.seatunnel.hive.config.HiveConfig.ABORT_DROP_PARTITION_METADATA;
-
 @Slf4j
 public class HiveSinkAggregatedCommitter extends FileSinkAggregatedCommitter {
-    private final Config pluginConfig;
     private final String dbName;
     private final String tableName;
     private final boolean abortDropPartitionMetadata;
 
+    private final ReadonlyConfig readonlyConfig;
+
     public HiveSinkAggregatedCommitter(
-            Config pluginConfig, String dbName, String tableName, HadoopConf hadoopConf) {
-        super(hadoopConf);
-        this.pluginConfig = pluginConfig;
-        this.dbName = dbName;
-        this.tableName = tableName;
+            ReadonlyConfig readonlyConfig, Table table, FileSystemUtils fileSystemUtils) {
+        super(fileSystemUtils);
+        this.readonlyConfig = readonlyConfig;
+        this.dbName = table.getDbName();
+        this.tableName = table.getTableName();
         this.abortDropPartitionMetadata =
-                pluginConfig.hasPath(ABORT_DROP_PARTITION_METADATA.key())
-                        ? pluginConfig.getBoolean(ABORT_DROP_PARTITION_METADATA.key())
-                        : ABORT_DROP_PARTITION_METADATA.defaultValue();
+                readonlyConfig.get(HiveSinkOptions.ABORT_DROP_PARTITION_METADATA);
     }
 
     @Override
     public List<FileAggregatedCommitInfo> commit(
             List<FileAggregatedCommitInfo> aggregatedCommitInfos) throws IOException {
+
         List<FileAggregatedCommitInfo> errorCommitInfos = super.commit(aggregatedCommitInfos);
         if (errorCommitInfos.isEmpty()) {
-            HiveMetaStoreProxy hiveMetaStore = HiveMetaStoreProxy.getInstance(pluginConfig);
+            HiveMetaStoreProxy hiveMetaStore = HiveMetaStoreProxy.getInstance(readonlyConfig);
             try {
                 for (FileAggregatedCommitInfo aggregatedCommitInfo : aggregatedCommitInfos) {
                     Map<String, List<String>> partitionDirAndValuesMap =
@@ -87,7 +87,7 @@ public List<FileAggregatedCommitInfo> commit(
     public void abort(List<FileAggregatedCommitInfo> aggregatedCommitInfos) throws Exception {
         super.abort(aggregatedCommitInfos);
         if (abortDropPartitionMetadata) {
-            HiveMetaStoreProxy hiveMetaStore = HiveMetaStoreProxy.getInstance(pluginConfig);
+            HiveMetaStoreProxy hiveMetaStore = HiveMetaStoreProxy.getInstance(readonlyConfig);
             for (FileAggregatedCommitInfo aggregatedCommitInfo : aggregatedCommitInfos) {
                 Map<String, List<String>> partitionDirAndValuesMap =
                         aggregatedCommitInfo.getPartitionDirAndValuesMap();

Diff for: seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/config/BaseHiveOptions.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.hive.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfig;
+
+public class BaseHiveOptions extends BaseSourceConfig {
+
+    public static final Option<String> TABLE_NAME =
+            Options.key("table_name")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Hive table name");
+
+    public static final Option<String> METASTORE_URI =
+            Options.key("metastore_uri")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Hive metastore uri");
+
+    public static final Option<String> HIVE_SITE_PATH =
+            Options.key("hive_site_path")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The path of hive-site.xml");
+}

Diff for: seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/config/HiveConfig.java

@@ -1,3 +1,4 @@
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
@@ -91,4 +92,4 @@ public static Pair<String[], Table> getTableInfo(Config config) {
         hiveMetaStoreProxy.close();
         return Pair.of(splits, tableInformation);
     }
-}
+}