[Feature][Connector-V2][JDBC] support sqlite Source & Sink

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>
+        <sqlite.version>3.39.3.0</sqlite.version>
     </properties>
 
     <dependencyManagement>
@@ -76,7 +77,12 @@
                 <version>${oracle.version}</version>
                 <scope>provided</scope>
             </dependency>
-
+            <dependency>
+                <groupId>org.xerial</groupId>
+                <artifactId>sqlite-jdbc</artifactId>
+                <version>${sqlite.version}</version>
+                <scope>provided</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
@@ -108,6 +114,10 @@
             <groupId>com.oracle.database.jdbc</groupId>
             <artifactId>ojdbc8</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.xerial</groupId>
+            <artifactId>sqlite-jdbc</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcConfig.java

@@ -53,6 +53,7 @@ public class JdbcConfig implements Serializable {
 
     public static final String TRANSACTION_TIMEOUT_SEC = "transaction_timeout_sec";
 
+    public static final String TYPE_AFFINITY = "type_affinity";
 
     //source config
     public static final String PARTITION_COLUMN = "partition_column";
@@ -95,6 +96,10 @@ public static JdbcConnectionOptions buildJdbcConnectionOptions(Config config) {
                 jdbcOptions.transactionTimeoutSec = config.getInt(JdbcConfig.TRANSACTION_TIMEOUT_SEC);
             }
         }
+
+        if (config.hasPath(JdbcConfig.TYPE_AFFINITY)) {
+            jdbcOptions.typeAffinity = config.getBoolean(JdbcConfig.TYPE_AFFINITY);
+        }
         return jdbcOptions;
     }
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlite/SqliteDialect.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.sqlite;
+
+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 SqliteDialect implements JdbcDialect {
+    @Override
+    public String dialectName() {
+        return "Sqlite";
+    }
+
+    @Override
+    public JdbcRowConverter getRowConverter() {
+        return new SqliteJdbcRowConverter();
+    }
+
+    @Override
+    public JdbcDialectTypeMapper getJdbcDialectTypeMapper() {
+        return new SqliteTypeMapper();
+    }
+}

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

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

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlite/SqliteJdbcRowConverter.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.sqlite;
+
+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 SqliteJdbcRowConverter extends AbstractJdbcRowConverter {
+
+    @Override
+    public String converterName() {
+        return "Sqlite";
+    }
+
+    @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/sqlite/SqliteTypeMapper.java

@@ -0,0 +1,127 @@
+/*
+ * 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.sqlite;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
+
+@Slf4j
+public class SqliteTypeMapper implements JdbcDialectTypeMapper {
+
+    private static final String OBJECT_CLASS = "java.lang.Object";
+    private static final String CHARACTER_CLASS = "java.lang.Character";
+    private static final String STRING_CLASS = "java.lang.String";
+    private static final String INTEGER_CLASS = "java.lang.Integer";
+    private static final String SHORT_CLASS = "java.lang.Short";
+    private static final String LONG_CLASS = "java.lang.Long";
+    private static final String FLOAT_CLASS = "java.lang.Float";
+    private static final String DOUBLE_CLASS = "java.lang.Double";
+    private static final String BOOLEAN_CLASS = "java.lang.Boolean";
+    private static final String BYTE_CLASS = "java.lang.Byte";
+
+    /**
+     * because of sqlite's dynamic data type and affinity, use columnType(java.sql.Types) determine the SeaTunnel Data Types
+     */
+    @SuppressWarnings("checkstyle:MagicNumber")
+    @Override
+    public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex) throws SQLException {
+        int columnType = metadata.getColumnType(colIndex);
+        String columnClassName = metadata.getColumnClassName(colIndex);
+        // sqlite data type is dynamic and type affinity, see https://www.sqlite.org/datatype3.html
+        // 1. if columnClassName is java.lang.Object, this column has no data, use columnType determine the SeaTunnel Data Types
+        // 2. otherwise we just use columnClassName to determine the SeaTunnel Data Types
+        if (columnClassName.equalsIgnoreCase(OBJECT_CLASS)) {   // case 1.
+            switch (columnType) {
+                case Types.CHAR:
+                case Types.NCHAR:
+                case Types.VARCHAR:
+                case Types.NVARCHAR:
+                case Types.LONGVARCHAR:
+                case Types.LONGNVARCHAR:
+                    return BasicType.STRING_TYPE;
+                case Types.INTEGER:
+                    return BasicType.INT_TYPE;
+                case Types.TINYINT:
+                case Types.SMALLINT:
+                    return BasicType.SHORT_TYPE;
+                case Types.BIGINT:
+                case Types.DATE:
+                case Types.TIME:
+                case Types.TIMESTAMP:
+                case Types.TIME_WITH_TIMEZONE:
+                case Types.TIMESTAMP_WITH_TIMEZONE:
+                    return BasicType.LONG_TYPE;
+                case Types.FLOAT:
+                    return BasicType.FLOAT_TYPE;
+                case Types.DOUBLE:
+                case Types.REAL:
+                case Types.NUMERIC:
+                case Types.DECIMAL:
+                    return BasicType.DOUBLE_TYPE;
+                case Types.BOOLEAN:
+                case Types.BIT:
+                    return BasicType.BOOLEAN_TYPE;
+                case Types.BLOB:
+                case Types.BINARY:
+                case Types.VARBINARY:
+                case Types.LONGVARBINARY:
+                    return PrimitiveByteArrayType.INSTANCE;
+                default:
+                    final String jdbcColumnName = metadata.getColumnName(colIndex);
+                    throw new UnsupportedOperationException(
+                            String.format(
+                                    "Doesn't support sql type '%s' on column '%s'  yet.",
+                                    columnType, jdbcColumnName));
+            }
+        } else {    // case 2
+            switch (columnClassName) {
+                case INTEGER_CLASS:
+                    return BasicType.INT_TYPE;
+                case SHORT_CLASS:
+                    return BasicType.SHORT_TYPE;
+                case LONG_CLASS:
+                    return BasicType.LONG_TYPE;
+                case FLOAT_CLASS:
+                    return BasicType.FLOAT_TYPE;
+                case DOUBLE_CLASS:
+                    return BasicType.DOUBLE_TYPE;
+                case STRING_CLASS:
+                case CHARACTER_CLASS:
+                    return BasicType.STRING_TYPE;
+                case BOOLEAN_CLASS:
+                    return BasicType.BOOLEAN_TYPE;
+                case BYTE_CLASS:
+                    return BasicType.BYTE_TYPE;
+                default:
+                    final String jdbcColumnName = metadata.getColumnName(colIndex);
+                    throw new UnsupportedOperationException(
+                            String.format(
+                                    "Doesn't support sql type '%s' on column '%s'  yet.",
+                                    columnType, jdbcColumnName));
+            }
+        }
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/options/JdbcConnectionOptions.java

@@ -38,6 +38,8 @@ public class JdbcConnectionOptions
     public String username;
     public String password;
     public String query;
+    // since sqlite data type affinity, the specific data type cannot be determined only by column type name
+    public boolean typeAffinity;
 
     public int batchSize = DEFAULT_BATCH_SIZE;
     public int batchIntervalMs = DEFAULT_BATCH_INTERVAL_MS;
@@ -99,6 +101,10 @@ public Optional<Integer> getTransactionTimeoutSec() {
         return transactionTimeoutSec < 0 ? Optional.empty() : Optional.of(transactionTimeoutSec);
     }
 
+    public boolean isTypeAffinity() {
+        return typeAffinity;
+    }
+
     public static JdbcConnectionOptionsBuilder builder() {
         return new JdbcConnectionOptionsBuilder();
     }
@@ -116,6 +122,7 @@ public static final class JdbcConnectionOptionsBuilder {
         private String xaDataSourceClassName;
         private int maxCommitAttempts = DEFAULT_MAX_COMMIT_ATTEMPTS;
         private int transactionTimeoutSec = DEFAULT_TRANSACTION_TIMEOUT_SEC;
+        private boolean typeAffinity;
 
         private JdbcConnectionOptionsBuilder() {
         }
@@ -180,6 +187,11 @@ public JdbcConnectionOptionsBuilder withTransactionTimeoutSec(int transactionTim
             return this;
         }
 
+        public JdbcConnectionOptionsBuilder withTypeAffinity(boolean affinity) {
+            this.typeAffinity = affinity;
+            return this;
+        }
+
         public JdbcConnectionOptions build() {
             JdbcConnectionOptions jdbcConnectionOptions = new JdbcConnectionOptions();
             jdbcConnectionOptions.batchSize = this.batchSize;
@@ -194,6 +206,7 @@ public JdbcConnectionOptions build() {
             jdbcConnectionOptions.transactionTimeoutSec = this.transactionTimeoutSec;
             jdbcConnectionOptions.maxCommitAttempts = this.maxCommitAttempts;
             jdbcConnectionOptions.xaDataSourceClassName = this.xaDataSourceClassName;
+            jdbcConnectionOptions.typeAffinity = this.typeAffinity;
             return jdbcConnectionOptions;
         }
     }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSource.java

@@ -35,6 +35,7 @@
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectLoader;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.sqlite.SqliteDialect;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.state.JdbcSourceState;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
@@ -51,6 +52,7 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 
 @AutoService(SeaTunnelSource.class)
 public class JdbcSource implements SeaTunnelSource<SeaTunnelRow, JdbcSourceSplit, JdbcSourceState> {
@@ -137,15 +139,31 @@ private SeaTunnelRowType initTableField(Connection conn) {
         JdbcDialectTypeMapper jdbcDialectTypeMapper = jdbcDialect.getJdbcDialectTypeMapper();
         ArrayList<SeaTunnelDataType<?>> seaTunnelDataTypes = new ArrayList<>();
         ArrayList<String> fieldNames = new ArrayList<>();
+        ResultSet resultSet = null;
         try {
-            PreparedStatement ps = conn.prepareStatement(jdbcSourceOptions.getJdbcConnectionOptions().getQuery());
-            ResultSetMetaData resultSetMetaData = ps.getMetaData();
+            ResultSetMetaData resultSetMetaData;
+            if (jdbcSourceOptions.getJdbcConnectionOptions().isTypeAffinity() && jdbcDialect instanceof SqliteDialect) {
+                PreparedStatement ps = conn.prepareStatement("select t.* from (" + jdbcSourceOptions.getJdbcConnectionOptions().getQuery() + ") as t limit 1, 1");
+                resultSet = ps.executeQuery();
+                resultSetMetaData = resultSet.getMetaData();
+            } else {
+                PreparedStatement ps = conn.prepareStatement(jdbcSourceOptions.getJdbcConnectionOptions().getQuery());
+                resultSetMetaData = ps.getMetaData();
+            }
             for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
                 fieldNames.add(resultSetMetaData.getColumnName(i));
                 seaTunnelDataTypes.add(jdbcDialectTypeMapper.mapping(resultSetMetaData, i));
             }
         } catch (Exception e) {
             LOG.warn("get row type info exception", e);
+        } finally {
+            if (Objects.nonNull(resultSet)) {
+                try {
+                    resultSet.close();
+                } catch (SQLException e) {
+                    LOG.warn("get row type info exception", e);
+                }
+            }
         }
         return new SeaTunnelRowType(fieldNames.toArray(new String[fieldNames.size()]), seaTunnelDataTypes.toArray(new SeaTunnelDataType<?>[seaTunnelDataTypes.size()]));
     }