[Feature][Connector-V2] Kylin source connector

docs/en/connector-v2/source/Jdbc.md

@@ -91,13 +91,14 @@ in parallel according to the concurrency of tasks.
 there are some reference value for params above.
 
 | datasource | driver                                       | url                                                          | maven                                                        |
-| ---------- | -------------------------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
+|------------| -------------------------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
 | mysql      | com.mysql.cj.jdbc.Driver                     | jdbc:mysql://localhost:3306/test                             | https://mvnrepository.com/artifact/mysql/mysql-connector-java |
 | postgresql | org.postgresql.Driver                        | jdbc:postgresql://localhost:5432/postgres                    | https://mvnrepository.com/artifact/org.postgresql/postgresql |
 | dm         | dm.jdbc.driver.DmDriver                      | jdbc:dm://localhost:5236                                     | https://mvnrepository.com/artifact/com.dameng/DmJdbcDriver18 |
 | phoenix    | org.apache.phoenix.queryserver.client.Driver | jdbc:phoenix:thin:url=http://localhost:8765;serialization=PROTOBUF | https://mvnrepository.com/artifact/com.aliyun.phoenix/ali-phoenix-shaded-thin-client |
 | sqlserver  | com.microsoft.sqlserver.jdbc.SQLServerDriver | jdbc:microsoft:sqlserver://localhost:1433                    | https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc |
 | oracle     | oracle.jdbc.OracleDriver                     | jdbc:oracle:thin:@localhost:1521/xepdb1                      | https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8 |
+| kylin      | org.apache.kylin.jdbc.Driver                 | jdbc:kylin://localhost:7070/kylin_project_name               | https://mvnrepository.com/artifact/org.apache.kylin/kylin-jdbc|
 
 ## Example
 

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

@@ -36,6 +36,7 @@
         <sqlserver.version>9.2.1.jre8</sqlserver.version>
         <phoenix.version>5.2.5-HBase-2.x</phoenix.version>
         <oracle.version>12.2.0.1</oracle.version>
+        <kylin.version>3.1.3</kylin.version>
     </properties>
 
     <dependencyManagement>
@@ -77,6 +78,13 @@
                 <scope>provided</scope>
             </dependency>
 
+            <dependency>
+                <groupId>org.apache.kylin</groupId>
+                <artifactId>kylin-jdbc</artifactId>
+                <version>${kylin.version}</version>
+                <scope>provided</scope>
+            </dependency>
+
         </dependencies>
     </dependencyManagement>
 
@@ -108,6 +116,10 @@
             <groupId>com.oracle.database.jdbc</groupId>
             <artifactId>ojdbc8</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kylin</groupId>
+            <artifactId>kylin-jdbc</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -0,0 +1,39 @@
+/*
+ * 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.kylin;
+
+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;
+
+public class KylinDialect implements JdbcDialect {
+    @Override
+    public String dialectName() {
+        return "Kylin";
+    }
+
+    @Override
+    public JdbcRowConverter getRowConverter() {
+        return new KylinJdbcRowConverter();
+    }
+
+    @Override
+    public JdbcDialectTypeMapper getJdbcDialectTypeMapper() {
+        return new KylinTypeMapper();
+    }
+}

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

@@ -0,0 +1,36 @@
+/*
+ * 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.kylin;
+
+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;
+
+@AutoService(JdbcDialectFactory.class)
+public class KylinDialectFactory implements JdbcDialectFactory {
+    @Override
+    public boolean acceptsURL(String url) {
+        return url.startsWith("jdbc:kylin:");
+    }
+
+    @Override
+    public JdbcDialect create() {
+        return new KylinDialect();
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/kylin/KylinJdbcRowConverter.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.kylin;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+public class KylinJdbcRowConverter extends AbstractJdbcRowConverter {
+    @Override
+    public String converterName() {
+        return "Kylin";
+    }
+
+    @Override
+    public SeaTunnelRow toInternal(ResultSet rs, ResultSetMetaData metaData, SeaTunnelRowType typeInfo) throws SQLException {
+        return super.toInternal(rs, metaData, typeInfo);
+    }
+}

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

@@ -0,0 +1,102 @@
+/*
+ * 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.kylin;
+
+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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+public class KylinTypeMapper implements JdbcDialectTypeMapper {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KylinTypeMapper.class);
+
+    // ============================data types=====================
+
+    private static final String KYLIN_UNKNOWN = "UNKNOWN";
+    private static final String KYLIN_BOOLEAN = "BOOLEAN";
+
+    // -------------------------number----------------------------
+    private static final String KYLIN_TINYINT = "TINYINT";
+    private static final String KYLIN_SMALLINT = "SMALLINT";
+    private static final String KYLIN_INT = "INT";
+    private static final String KYLIN_BIGINT = "BIGINT";
+    private static final String KYLIN_DECIMAL = "DECIMAL";
+    private static final String KYLIN_FLOAT = "FLOAT";
+    private static final String KYLIN_DOUBLE = "DOUBLE";
+
+    // -------------------------string----------------------------
+    private static final String KYLIN_CHAR = "CHAR";
+    private static final String KYLIN_VARCHAR = "VARCHAR";
+    private static final String KYLIN_STRING = "STRING";
+
+    // ------------------------------time-------------------------
+    private static final String KYLIN_DATE = "DATE";
+    private static final String KYLIN_TIME = "TIME";
+    private static final String KYLIN_TIMESTAMP = "TIMESTAMP";
+
+    @SuppressWarnings("checkstyle:MagicNumber")
+    @Override
+    public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex) throws SQLException {
+        String kylinType = metadata.getColumnTypeName(colIndex).toUpperCase();
+        int precision = metadata.getPrecision(colIndex);
+        int scale = metadata.getScale(colIndex);
+        switch (kylinType) {
+            case KYLIN_BOOLEAN:
+                return BasicType.BOOLEAN_TYPE;
+            case KYLIN_TINYINT:
+            case KYLIN_SMALLINT:
+                return BasicType.SHORT_TYPE;
+            case KYLIN_INT:
+                return BasicType.INT_TYPE;
+            case KYLIN_BIGINT:
+                return BasicType.LONG_TYPE;
+            case KYLIN_DECIMAL:
+                return new DecimalType(precision, scale);
+            case KYLIN_FLOAT:
+                return BasicType.FLOAT_TYPE;
+            case KYLIN_DOUBLE:
+                return BasicType.DOUBLE_TYPE;
+            case KYLIN_CHAR:
+            case KYLIN_VARCHAR:
+            case KYLIN_STRING:
+                return BasicType.STRING_TYPE;
+            case KYLIN_DATE:
+                return LocalTimeType.LOCAL_DATE_TYPE;
+            case KYLIN_TIME:
+                return LocalTimeType.LOCAL_TIME_TYPE;
+            case KYLIN_TIMESTAMP:
+                return LocalTimeType.LOCAL_DATE_TIME_TYPE;
+            //Doesn't support yet
+            case KYLIN_UNKNOWN:
+            default:
+                final String jdbcColumnName = metadata.getColumnName(colIndex);
+                throw new UnsupportedOperationException(
+                        String.format(
+                                "Doesn't support KYLIN type '%s' on column '%s'  yet.",
+                                kylinType, jdbcColumnName));
+        }
+    }
+}