[Feature][Hive JDBC Source] Support Hive JDBC Source Connector

docs/en/connector-v2/source/Hive-jdbc.md

@@ -0,0 +1,156 @@
+# Hive
+
+> JDBC Hive Source Connector
+
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
+## Key Features
+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [ ] [stream](../../concept/connector-v2-features.md)
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [x] [column projection](../../concept/connector-v2-features.md)
+- [x] [parallelism](../../concept/connector-v2-features.md)
+- [x] [support user-defined split](../../concept/connector-v2-features.md)
+
+> supports query SQL and can achieve projection effect.
+
+## Description
+
+Read external data source data through JDBC.
+
+## Supported DataSource Info
+
+| Datasource | Supported versions | Driver | Url | Maven |
+|------------|----------------------------------------------------------|---------------------------------|--------------------------------------|--------------------------------------------------------------------------|
+| Hive | Different dependency version has different driver class. | org.apache.hive.jdbc.HiveDriver | jdbc:hive2://localhost:10000/default | [Download](https://mvnrepository.com/artifact/org.apache.hive/hive-jdbc) |
+
+## Database Dependency
+
+> Please download the support list corresponding to 'Maven' and copy it to the '$SEATNUNNEL_HOME/plugins/jdbc/lib/'
+> working directory<br/>
+> For example Hive datasource: cp hive-jdbc-xxx.jar $SEATNUNNEL_HOME/plugins/jdbc/lib/
+
+## Data Type Mapping
+
+| Hive Data type | SeaTunnel Data type |
+|-------------------------------------------------------------------------------------------|---------------------|
+| BOOLEAN | BOOLEAN |
+| TINYINT<br/> SMALLINT | SHORT |
+| INT<br/>INTEGER | INT |
+| BIGINT | LONG |
+| FLOAT | FLOAT |
+| DOUBLE<br/>DOUBLE PRECISION | DOUBLE |
+| DECIMAL(x,y)<br/>NUMERIC(x,y)<br/>(Get the designated column's specified column size.<38) | DECIMAL(x,y) |
+| DECIMAL(x,y)<br/>NUMERIC(x,y)<br/>(Get the designated column's specified column size.>38) | DECIMAL(38,18) |
+| CHAR<br/>VARCHAR<br/>STRING | STRING |
+| DATE | DATE |
+| DATETIME<br/>TIMESTAMP | TIMESTAMP |
+| BINARY<br/> ARRAY <br/>INTERVAL <br/>MAP <br/>STRUCT<br/>UNIONTYPE | Not supported yet |
+
+## Source Options
+
+| Name | Type | Required | Default | Description |
+|------------------------------|------------|----------|-----------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| url | String | Yes | - | The URL of the JDBC connection. Refer to a case: jdbc:hive2://localhost:10000/default |
+| driver | String | Yes | - | The jdbc class name used to connect to the remote data source,<br/> if you use Hive the value is `org.apache.hive.jdbc.HiveDriver`. |
+| user | String | No | - | Connection instance user name |
+| password | String | No | - | Connection instance password |
+| query | String | Yes | - | Query statement |
+| connection_check_timeout_sec | Int | No | 30 | The time in seconds to wait for the database operation used to validate the connection to complete |
+| partition_column | String | No | - | The column name for parallelism's partition, only support numeric type,Only support numeric type primary key, and only can config one column. |
+| partition_lower_bound | BigDecimal | No | - | The partition_column min value for scan, if not set SeaTunnel will query database get min value. |
+| partition_upper_bound | BigDecimal | No | - | The partition_column max value for scan, if not set SeaTunnel will query database get max value. |
+| partition_num | Int | No | job parallelism | The number of partition count, only support positive integer. default value is job parallelism |
+| fetch_size | Int | No | 0 | For queries that return a large number of objects,you can configure<br/> the row fetch size used in the query toimprove performance by<br/> reducing the number database hits required to satisfy the selection criteria.<br/> Zero means use jdbc default value. |
+| common-options | | No | - | Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details |
+
+### Tips
+
+> If partition_column is not set, it will run in single concurrency, and if partition_column is set, it will be executed
+> in parallel according to the concurrency of tasks , When your shard read field is a large number type such as bigint(
+> and above and the data is not evenly distributed, it is recommended to set the parallelism level to 1 to ensure that
+> the
+> data skew problem is resolved
+
+## Task Example
+
+### Simple:
+
+> This example queries type_bin 'table' 16 data in your test "database" in single parallel and queries all of its
+> fields. You can also specify which fields to query for final output to the console.
+
+```
+# Defining the runtime environment
+env {
+ # You can set flink configuration here
+ execution.parallelism = 2
+ job.mode = "BATCH"
+}
+source{
+ Jdbc {
+ url = "jdbc:hive2://localhost:10000/default"
+ driver = "org.apache.hive.jdbc.HiveDriver"
+ connection_check_timeout_sec = 100
+ query = "select * from type_bin limit 16"
+ }
+}
+
+transform {
+ # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+ # please go to https://seatunnel.apache.org/docs/transform-v2/sql
+}
+
+sink {
+ Console {}
+}
+```
+
+### Parallel:
+
+> Read your query table in parallel with the shard field you configured and the shard data You can do this if you want
+> to read the whole table
+
+```
+source {
+ Jdbc {
+ url = "jdbc:hive2://localhost:10000/default"
+ driver = "org.apache.hive.jdbc.HiveDriver"
+ connection_check_timeout_sec = 100
+ # Define query logic as required
+ query = "select * from type_bin"
+ # Parallel sharding reads fields
+ partition_column = "id"
+ # Number of fragments
+ partition_num = 10
+ }
+}
+```
+
+### Parallel Boundary:
+
+> It is more efficient to specify the data within the upper and lower bounds of the query It is more efficient to read
+> your data source according to the upper and lower boundaries you configured
+
+```
+source {
+ Jdbc {
+ url = "jdbc:hive2://localhost:10000/default"
+ driver = "org.apache.hive.jdbc.HiveDriver"
+ connection_check_timeout_sec = 100
+ # Define query logic as required
+ query = "select * from type_bin"
+ partition_column = "id"
+ # Read start boundary
+ partition_lower_bound = 1
+ # Read end boundary
+ partition_upper_bound = 500
+ partition_num = 10
+ }
+}
+```
+

seatunnel-connectors-v2/connector-jdbc/pom.xml

@@ -46,6 +46,7 @@
  <snowflake.version>3.13.29</snowflake.version>
  <vertica.version>12.0.3-0</vertica.version>
  <postgis.jdbc.version>2.5.1</postgis.jdbc.version>
+ <hive.jdbc.version>3.1.2</hive.jdbc.version>
  </properties>
 
  <dependencyManagement>
@@ -143,6 +144,12 @@
  <version>${vertica.version}</version>
  <scope>provided</scope>
  </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-jdbc</artifactId>
+ <version>${hive.jdbc.version}</version>
+ <scope>provided</scope>
+ </dependency>
  </dependencies>
  </dependencyManagement>
 
@@ -218,5 +225,9 @@
  <groupId>com.vertica.jdbc</groupId>
  <artifactId>vertica-jdbc</artifactId>
  </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-jdbc</artifactId>
+ </dependency>
  </dependencies>
 </project>

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/exception/JdbcConnectorErrorCode.java

@@ -27,8 +27,8 @@ public enum JdbcConnectorErrorCode implements SeaTunnelErrorCode {
  CONNECT_DATABASE_FAILED("JDBC-04", "Connector database failed"),
  TRANSACTION_OPERATION_FAILED(
  "JDBC-05", "transaction operation failed, such as (commit, rollback) etc.."),
- NO_SUITABLE_DIALECT_FACTORY("JDBC-06", "No suitable dialect factory found");
-
+ NO_SUITABLE_DIALECT_FACTORY("JDBC-06", "No suitable dialect factory found"),
+ DONT_SUPPORT_SINK("JDBC-07", "The jdbc type don't support sink");
  private final String code;
 
  private final String description;

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/hive/HiveDialect.java

@@ -0,0 +1,60 @@
+/*
+ * 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.jdbc.internal.dialect.hive;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.Optional;
+
+public class HiveDialect implements JdbcDialect {
+
+ @Override
+ public String dialectName() {
+ return "HIVE";
+ }
+
+ @Override
+ public JdbcRowConverter getRowConverter() {
+ return new HiveJdbcRowConverter();
+ }
+
+ @Override
+ public JdbcDialectTypeMapper getJdbcDialectTypeMapper() {
+ return new HiveTypeMapper();
+ }
+
+ @Override
+ public Optional<String> getUpsertStatement(
+ String database, String tableName, String[] fieldNames, String[] uniqueKeyFields) {
+ return Optional.empty();
+ }
+
+ @Override
+ public ResultSetMetaData getResultSetMetaData(
+ Connection conn, JdbcSourceConfig jdbcSourceConfig) throws SQLException {
+ PreparedStatement ps = conn.prepareStatement(jdbcSourceConfig.getQuery());
+ return ps.executeQuery().getMetaData();
+ }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/hive/HiveDialectFactory.java

@@ -0,0 +1,38 @@
+/*
+ * 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.jdbc.internal.dialect.hive;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory;
+
+import com.google.auto.service.AutoService;
+
+/** Factory for {@link HiveDialect}. */
+@AutoService(JdbcDialectFactory.class)
+public class HiveDialectFactory implements JdbcDialectFactory {
+
+ @Override
+ public boolean acceptsURL(String url) {
+ return url.startsWith("jdbc:hive2:");
+ }
+
+ @Override
+ public JdbcDialect create() {
+ return new HiveDialect();
+ }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/hive/HiveJdbcRowConverter.java

@@ -0,0 +1,42 @@
+/*
+ * 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.jdbc.internal.dialect.hive;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
+
+import java.sql.PreparedStatement;
+
+public class HiveJdbcRowConverter extends AbstractJdbcRowConverter {
+
+ @Override
+ public String converterName() {
+ return "Hive";
+ }
+
+ @Override
+ public PreparedStatement toExternal(
+ SeaTunnelRowType rowType, SeaTunnelRow row, PreparedStatement statement) {
+ throw new JdbcConnectorException(
+ JdbcConnectorErrorCode.DONT_SUPPORT_SINK,
+ "The Hive jdbc connector don't support sink");
+ }
+}