fix(clickhouse): clickhouse支持jdbc方式读

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

@@ -50,6 +50,7 @@
         <kingbase8.version>8.6.0</kingbase8.version>
         <hive.jdbc.version>3.1.0</hive.jdbc.version>
         <oceanbase.jdbc.version>2.4.3</oceanbase.jdbc.version>
+        <clickhouse.jdbc.version>0.2.6</clickhouse.jdbc.version>
     </properties>
 
     <dependencyManagement>
@@ -192,6 +193,12 @@
                 <version>${oceanbase.jdbc.version}</version>
                 <scope>provided</scope>
             </dependency>
+            <dependency>
+                <groupId>ru.yandex.clickhouse</groupId>
+                <artifactId>clickhouse-jdbc</artifactId>
+                <version>${clickhouse.jdbc.version}</version>
+                <scope>provided</scope>
+            </dependency>
             <dependency>
                 <groupId>org.apache.hadoop</groupId>
                 <artifactId>hadoop-auth</artifactId>

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

@@ -38,4 +38,5 @@ public class DatabaseIdentifier {
     public static final String VERTICA = "Vertica";
     public static final String OCENABASE = "OceanBase";
     public static final String TIDB = "TiDB";
+    public static final String CLICKHOUSE = "Clickhouse";
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/clickhouse/ClickhouseDialect.java

@@ -0,0 +1,37 @@
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.clickhouse;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+import java.util.Optional;
+
+/**
+ * Description Copyright © 启明星辰 版权所有
+ *
+ * @author chaorongzhi
+ * @date 2024/1/5
+ */
+public class ClickhouseDialect implements JdbcDialect {
+    @Override
+    public String dialectName() {
+        return DatabaseIdentifier.CLICKHOUSE;
+    }
+
+    @Override
+    public JdbcRowConverter getRowConverter() {
+        return new ClickhouseJdbcRowConverter();
+    }
+
+    @Override
+    public JdbcDialectTypeMapper getJdbcDialectTypeMapper() {
+        return new ClickhouseTypeMapper();
+    }
+
+    @Override
+    public Optional<String> getUpsertStatement(
+            String database, String tableName, String[] fieldNames, String[] uniqueKeyFields) {
+        return Optional.empty();
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/clickhouse/ClickhouseDialectFactory.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.clickhouse;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql.MysqlDialect;
+
+import com.google.auto.service.AutoService;
+
+/** Factory for {@link MysqlDialect}. */
+@AutoService(JdbcDialectFactory.class)
+public class ClickhouseDialectFactory implements JdbcDialectFactory {
+    @Override
+    public boolean acceptsURL(String url) {
+        return url.startsWith("jdbc:clickhouse:");
+    }
+
+    @Override
+    public JdbcDialect create() {
+        return new ClickhouseDialect();
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/clickhouse/ClickhouseJdbcRowConverter.java

@@ -0,0 +1,28 @@
+/*
+ * 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.clickhouse;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
+
+public class ClickhouseJdbcRowConverter extends AbstractJdbcRowConverter {
+    @Override
+    public String converterName() {
+        return DatabaseIdentifier.CLICKHOUSE;
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/clickhouse/ClickhouseTypeMapper.java

@@ -0,0 +1,80 @@
+/*
+ * 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.clickhouse;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.DecimalType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+public class ClickhouseTypeMapper implements JdbcDialectTypeMapper {
+
+    @Override
+    public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex)
+            throws SQLException {
+        String clickhouseType = metadata.getColumnTypeName(colIndex);
+        int precision = metadata.getPrecision(colIndex);
+        int scale = metadata.getScale(colIndex);
+        switch (clickhouseType) {
+            case "UInt32":
+            case "UInt64":
+            case "Int64":
+            case "IntervalYear":
+            case "IntervalQuarter":
+            case "IntervalMonth":
+            case "IntervalWeek":
+            case "IntervalDay":
+            case "IntervalHour":
+            case "IntervalMinute":
+            case "IntervalSecond":
+                return BasicType.LONG_TYPE;
+            case "Float64":
+                return BasicType.DOUBLE_TYPE;
+            case "Float32":
+                return BasicType.FLOAT_TYPE;
+            case "Int8":
+            case "UInt8":
+            case "Int16":
+            case "UInt16":
+            case "Int32":
+                return BasicType.INT_TYPE;
+            case "Date":
+                return LocalTimeType.LOCAL_DATE_TYPE;
+            case "DateTime":
+                return LocalTimeType.LOCAL_DATE_TIME_TYPE;
+            case "Decimal":
+            case "Decimal64":
+            case "Decimal128":
+            case "Decimal256":
+                return new DecimalType(precision, scale);
+            case "String":
+            case "Int128":
+            case "UInt128":
+            case "Int256":
+            case "UInt256":
+            case "Point":
+            case "Polygon":
+            default:
+                return BasicType.STRING_TYPE;
+        }
+    }
+}

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

@@ -89,8 +89,6 @@ public Connection getOrEstablishConnection() throws SQLException, ClassNotFoundE
             } catch (IOException e) {
                 throw new JdbcConnectorException(KERBEROS_AUTHENTICATION_FAILED, e);
             }
-
-            //            super.getJdbcConfig().url = getHwHiveUrl(url);
         }
         Driver driver = getLoadedDriver();
         Properties info = new Properties();

seatunnel-examples/seatunnel-engine-examples/pom.xml

@@ -77,5 +77,10 @@
             <artifactId>hive-jdbc</artifactId>
             <version>3.1.0</version>
         </dependency>
+        <dependency>
+            <groupId>ru.yandex.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>0.2.6</version>
+        </dependency>
     </dependencies>
 </project>

seatunnel-examples/seatunnel-engine-examples/src/main/java/org/apache/seatunnel/example/engine/SeaTunnelEngineExample.java

@@ -31,7 +31,7 @@ public class SeaTunnelEngineExample {
 
     public static void main(String[] args)
             throws FileNotFoundException, URISyntaxException, CommandException {
-        String configurePath = args.length > 0 ? args[0] : "/examples/hive_hive_hw.conf";
+        String configurePath = args.length > 0 ? args[0] : "/examples/jdbc_clickhouse.conf";
         String configFile = getTestConfigFile(configurePath);
         ClientCommandArgs clientCommandArgs = new ClientCommandArgs();
         clientCommandArgs.setConfigFile(configFile);