[Feature][connector][access] Support source/sink for access

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

@@ -0,0 +1,90 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-access</artifactId>
+
+    <name>connector-access</name>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <ucanaccess.version>4.0.4</ucanaccess.version>
+        <db2.version>db2jcc4</db2.version>
+        <mysql.version>8.0.16</mysql.version>
+        <java-driver-core.version>4.14.0</java-driver-core.version>
+        <flink.version>1.18-SNAPSHOT</flink.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>3.8.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>net.sf.ucanaccess</groupId>
+            <artifactId>ucanaccess</artifactId>
+            <version>${ucanaccess.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-jdbc</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-flink-15-starter</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.datastax.oss</groupId>
+            <artifactId>java-driver-core</artifactId>
+            <version>${java-driver-core.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-connectors-v2/connector-access/src/main/java/org/apache/seatunnel/connectors/seatunnel/access/client/AccessClient.java

@@ -0,0 +1,67 @@
+/*
+ * 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.access.client;
+
+import java.io.Serializable;
+import java.sql.*;
+
+public class AccessClient implements Serializable {
+    private String driver;
+    private String url;
+    private String username;
+    private String password;
+    private String query;
+    private Connection connection;
+
+    public AccessClient(String driver, String url, String username, String password, String query) {
+        this.driver = driver;
+        this.url = url;
+        this.username = username;
+        this.password = password;
+        this.query = query;
+    }
+
+    public Connection getAccessConnection(String url, String username, String password) {
+        try {
+            Class.forName(driver);
+            this.connection = DriverManager.getConnection(url, username, password);
+        } catch (Exception e) {
+            throw new RuntimeException(e.getMessage());
+        }
+        return this.connection;
+    }
+
+    public ResultSetMetaData selectMetaData() throws Exception {
+        connection = this.getAccessConnection(url, username, password);
+        Statement statement = connection.createStatement();
+        ResultSet result = statement.executeQuery(query);
+        ResultSetMetaData metaData = result.getMetaData();
+        statement.close();
+        return metaData;
+    }
+
+    public ResultSetMetaData getTableSchema(String tableName) throws Exception {
+        connection = this.getAccessConnection(url, username, password);
+        Statement statement = connection.createStatement();
+        ResultSet result =
+                statement.executeQuery(String.format("select * from %s limit 1", tableName));
+        ResultSetMetaData metaData = result.getMetaData();
+        statement.close();
+        return metaData;
+    }
+}

seatunnel-connectors-v2/connector-access/src/main/java/org/apache/seatunnel/connectors/seatunnel/access/config/AccessConfig.java

@@ -0,0 +1,51 @@
+/*
+ * 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.access.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public class AccessConfig {
+    public static final Integer DEFAULT_BATCH_SIZE = 5000;
+
+    public static final Option<String> DRIVER =
+            Options.key("driver").stringType().noDefaultValue().withDescription("driver");
+
+    public static final Option<String> URL =
+            Options.key("url").stringType().noDefaultValue().withDescription("url");
+
+    public static final Option<String> USERNAME =
+            Options.key("username").stringType().noDefaultValue().withDescription("username");
+
+    public static final Option<String> PASSWORD =
+            Options.key("password").stringType().noDefaultValue().withDescription("password");
+
+    public static final Option<String> QUERY =
+            Options.key("query").stringType().noDefaultValue().withDescription("query");
+
+    public static final Option<String> TABLE =
+            Options.key("table").stringType().noDefaultValue().withDescription("table");
+
+    public static final Option<String> FIELDS =
+            Options.key("fields").stringType().defaultValue("LOCAL_ONE").withDescription("fields");
+    public static final Option<Integer> BATCH_SIZE =
+            Options.key("batch_size")
+                    .intType()
+                    .defaultValue(DEFAULT_BATCH_SIZE)
+                    .withDescription("");
+}

seatunnel-connectors-v2/connector-access/src/main/java/org/apache/seatunnel/connectors/seatunnel/access/config/AccessParameters.java

@@ -0,0 +1,64 @@
+/*
+ * 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.access.config;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import java.io.Serializable;
+import java.util.List;
+
+@Setter
+@Getter
+public class AccessParameters implements Serializable {
+    private String driver;
+    private String url;
+    private String username;
+    private String password;
+    private String query;
+    private List<String> fields;
+    private String table;
+    private Integer batchSize;
+
+    public void buildWithConfig(Config config) {
+        this.driver = config.getString(AccessConfig.DRIVER.key());
+        this.url = config.getString(AccessConfig.URL.key());
+
+        if (config.hasPath(AccessConfig.USERNAME.key())) {
+            this.username = config.getString(AccessConfig.USERNAME.key());
+        }
+        if (config.hasPath(AccessConfig.PASSWORD.key())) {
+            this.password = config.getString(AccessConfig.PASSWORD.key());
+        }
+        if (config.hasPath(AccessConfig.QUERY.key())) {
+            this.query = config.getString(AccessConfig.QUERY.key());
+        }
+        if (config.hasPath(AccessConfig.FIELDS.key())) {
+            this.fields = config.getStringList(AccessConfig.FIELDS.key());
+        }
+        if (config.hasPath(AccessConfig.TABLE.key())) {
+            this.table = config.getString(AccessConfig.TABLE.key());
+        }
+        if (config.hasPath(AccessConfig.BATCH_SIZE.key())) {
+            this.batchSize = config.getInt(AccessConfig.BATCH_SIZE.key());
+        } else {
+            this.batchSize = AccessConfig.BATCH_SIZE.defaultValue();
+        }
+    }
+}

seatunnel-connectors-v2/connector-access/src/main/java/org/apache/seatunnel/connectors/seatunnel/access/exception/AccessConnectorErrorCode.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.access.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+
+public enum AccessConnectorErrorCode implements SeaTunnelErrorCode {
+    FIELD_NOT_IN_TABLE("ACCESS-01", "Field is not existed in target table"),
+
+    CLOSE_CQL_CONNECT_FAILED("ACCESS-03", "Close connect of access failed");
+    private final String code;
+    private final String description;
+
+    AccessConnectorErrorCode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+    @Override
+    public String getCode() {
+        return null;
+    }
+
+    @Override
+    public String getDescription() {
+        return null;
+    }
+}

seatunnel-connectors-v2/connector-access/src/main/java/org/apache/seatunnel/connectors/seatunnel/access/exception/AccessConnectorException.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.access.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class AccessConnectorException extends SeaTunnelRuntimeException {
+    public AccessConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
+        super(seaTunnelErrorCode, errorMessage);
+    }
+
+    public AccessConnectorException(
+            SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
+        super(seaTunnelErrorCode, errorMessage, cause);
+    }
+
+    public AccessConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode, cause);
+    }
+}