[Feature][seatunnel-datasource]: Add doris data source plugin

seatunnel-datasource/seatunnel-datasource-client/src/main/java/org/apache/seatunnel/datasource/classloader/DatasourceLoadConfig.java

@@ -106,6 +106,8 @@ public class DatasourceLoadConfig {
         classLoaderFactoryName.put(
                 "CONSOLE",
                 "org.apache.seatunnel.datasource.plugin.console.ConsoleDataSourceFactory");
+        classLoaderFactoryName.put(
+                "DORIS", "org.apache.seatunnel.datasource.plugin.doris.DorisDataSourceFactory");
 
         classLoaderJarName.put("JDBC-ORACLE", "datasource-jdbc-oracle-");
         classLoaderJarName.put("JDBC-CLICKHOUSE", "datasource-jdbc-clickhouse-");
@@ -130,6 +132,7 @@ public class DatasourceLoadConfig {
         classLoaderJarName.put("JDBC-HIVE", "datasource-jdbc-hive-");
         classLoaderJarName.put("FAKESOURCE", "datasource-fakesource-");
         classLoaderJarName.put("CONSOLE", "datasource-console-");
+        classLoaderJarName.put("DORIS", "datasource-doris-");
     }
 
     public static final Set<String> pluginSet =
@@ -151,7 +154,8 @@ public class DatasourceLoadConfig {
                     "MongoDB",
                     "JDBC-Db2",
                     "FakeSource",
-                    "Console");
+                    "Console",
+                    "DORIS");
 
     public static Map<String, DatasourceClassLoader> datasourceClassLoaders = new HashMap<>();
 

seatunnel-datasource/seatunnel-datasource-plugins/datasource-all/pom.xml

@@ -102,6 +102,11 @@
             <artifactId>datasource-jdbc-db2</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>datasource-doris</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-datasource/seatunnel-datasource-plugins/datasource-doris/pom.xml

@@ -0,0 +1,56 @@
+<?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-datasource-plugins</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>datasource-doris</artifactId>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>datasource-plugins-api</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/com.google.auto.service/auto-service -->
+        <dependency>
+            <groupId>com.google.auto.service</groupId>
+            <artifactId>auto-service</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- driver -->
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>${mysql-connector.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-datasource/seatunnel-datasource-plugins/datasource-doris/src/main/java/org/apache/seatunnel/datasource/plugin/doris/DorisDataSourceChannel.java

@@ -0,0 +1,202 @@
+/*
+ * 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.datasource.plugin.doris;
+
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.datasource.plugin.api.DataSourceChannel;
+import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
+import org.apache.seatunnel.datasource.plugin.api.model.TableField;
+import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
+
+import org.apache.commons.lang3.StringUtils;
+
+import lombok.NonNull;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+public class DorisDataSourceChannel implements DataSourceChannel {
+
+    @Override
+    public OptionRule getDataSourceOptions(@NonNull String pluginName) {
+        return DorisDataSourceConfig.OPTION_RULE;
+    }
+
+    @Override
+    public OptionRule getDatasourceMetadataFieldsByDataSourceName(@NonNull String pluginName) {
+        return DorisDataSourceConfig.METADATA_RULE;
+    }
+
+    @Override
+    public List<String> getTables(
+            @NonNull String pluginName,
+            Map<String, String> requestParams,
+            String database,
+            Map<String, String> options) {
+        List<String> tableNames = new ArrayList<>();
+        String filterName = options.get("filterName");
+        String size = options.get("size");
+        boolean isSize = StringUtils.isNotEmpty(size);
+        if (StringUtils.isNotEmpty(filterName) && !filterName.contains("%")) {
+            filterName = "%" + filterName + "%";
+        } else if (StringUtils.equals(filterName, "")) {
+            filterName = null;
+        }
+        try (Connection connection = getConnection(requestParams);
+                ResultSet resultSet =
+                        connection
+                                .getMetaData()
+                                .getTables(database, null, filterName, new String[] {"TABLE"})) {
+            while (resultSet.next()) {
+                String tableName = resultSet.getString("TABLE_NAME");
+                if (StringUtils.isNotBlank(tableName)) {
+                    tableNames.add(tableName);
+                    if (isSize && tableNames.size() >= Integer.parseInt(size)) {
+                        break;
+                    }
+                }
+            }
+            return tableNames;
+        } catch (ClassNotFoundException | SQLException e) {
+            throw new DataSourcePluginException("get table names failed", e);
+        }
+    }
+
+    private Connection getConnection(Map<String, String> requestParams)
+            throws SQLException, ClassNotFoundException {
+        return getConnection(requestParams, null);
+    }
+
+    private Connection getConnection(Map<String, String> requestParams, String databaseName)
+            throws SQLException, ClassNotFoundException {
+        checkNotNull(requestParams.get(DorisOptionRule.DRIVER.key()));
+        checkNotNull(requestParams.get(DorisOptionRule.URL.key()), "doris url cannot be null");
+        String url =
+                JdbcUtils.replaceDatabase(
+                        requestParams.get(DorisOptionRule.URL.key()), databaseName);
+
+        Properties info = new java.util.Properties();
+        info.put("autoDeserialize", "false");
+        info.put("allowLoadLocalInfile", "false");
+        info.put("allowLoadLocalInfileInPath", "");
+        if (requestParams.containsKey(DorisOptionRule.USERNAME.key())) {
+            info.put("user", requestParams.get(DorisOptionRule.USERNAME.key()));
+            info.put("password", requestParams.get(DorisOptionRule.PASSWORD.key()));
+        }
+        return DriverManager.getConnection(url, info);
+    }
+
+    @Override
+    public List<String> getDatabases(
+            @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
+        List<String> dbNames = new ArrayList<>();
+        try (Connection connection = getConnection(requestParams);
+                PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;");
+                ResultSet re = statement.executeQuery()) {
+            // filter system databases
+            while (re.next()) {
+                String dbName = re.getString("database");
+                if (StringUtils.isNotBlank(dbName)
+                        && !DorisDataSourceConfig.DORIS_SYSTEM_DATABASES.contains(dbName)) {
+                    dbNames.add(dbName);
+                }
+            }
+            return dbNames;
+        } catch (SQLException | ClassNotFoundException e) {
+            throw new DataSourcePluginException("Get databases failed", e);
+        }
+    }
+
+    @Override
+    public boolean checkDataSourceConnectivity(
+            @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
+        try (Connection ignored = getConnection(requestParams)) {
+            return true;
+        } catch (Exception e) {
+            throw new DataSourcePluginException(
+                    "check doris connectivity failed; " + e.getMessage());
+        }
+    }
+
+    @Override
+    public List<TableField> getTableFields(
+            @NonNull String pluginName,
+            @NonNull Map<String, String> requestParams,
+            @NonNull String database,
+            @NonNull String table) {
+        List<TableField> tableFields = new ArrayList<>();
+        try (Connection connection = getConnection(requestParams, database)) {
+            DatabaseMetaData metaData = connection.getMetaData();
+            String primaryKey = getPrimaryKey(metaData, database, table);
+            try (ResultSet resultSet = metaData.getColumns(database, null, table, null)) {
+                while (resultSet.next()) {
+                    TableField tableField = new TableField();
+                    String columnName = resultSet.getString("COLUMN_NAME");
+                    tableField.setPrimaryKey(false);
+                    if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
+                        tableField.setPrimaryKey(true);
+                    }
+                    tableField.setName(columnName);
+                    tableField.setType(resultSet.getString("TYPE_NAME"));
+                    tableField.setComment(resultSet.getString("REMARKS"));
+                    Object nullable = resultSet.getObject("IS_NULLABLE");
+                    tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
+                    tableFields.add(tableField);
+                }
+            }
+        } catch (ClassNotFoundException | SQLException e) {
+            throw new DataSourcePluginException("get table fields failed", e);
+        }
+        return tableFields;
+    }
+
+    @Override
+    public Map<String, List<TableField>> getTableFields(
+            @NonNull String pluginName,
+            @NonNull Map<String, String> requestParams,
+            @NonNull String database,
+            @NonNull List<String> tables) {
+        return tables.parallelStream()
+                .collect(
+                        Collectors.toMap(
+                                Function.identity(),
+                                table ->
+                                        getTableFields(
+                                                pluginName, requestParams, database, table)));
+    }
+
+    private String getPrimaryKey(DatabaseMetaData metaData, String dbName, String tableName)
+            throws SQLException {
+        ResultSet primaryKeysInfo = metaData.getPrimaryKeys(dbName, "%", tableName);
+        while (primaryKeysInfo.next()) {
+            return primaryKeysInfo.getString("COLUMN_NAME");
+        }
+        return null;
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-doris/src/main/java/org/apache/seatunnel/datasource/plugin/doris/DorisDataSourceConfig.java

@@ -0,0 +1,50 @@
+/*
+ * 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.datasource.plugin.doris;
+
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginInfo;
+import org.apache.seatunnel.datasource.plugin.api.DatasourcePluginTypeEnum;
+
+import com.google.common.collect.Sets;
+
+import java.util.Set;
+
+public class DorisDataSourceConfig {
+
+    public static final String PLUGIN_NAME = "Doris";
+
+    public static final DataSourcePluginInfo DORIS_DATASOURCE_PLUGIN_INFO =
+            DataSourcePluginInfo.builder()
+                    .name(PLUGIN_NAME)
+                    .icon(PLUGIN_NAME)
+                    .version("1.0.0")
+                    .type(DatasourcePluginTypeEnum.DATABASE.getCode())
+                    .build();
+
+    public static final OptionRule OPTION_RULE =
+            OptionRule.builder()
+                    .required(DorisOptionRule.URL, DorisOptionRule.FE_NODES, DorisOptionRule.DRIVER)
+                    .optional(DorisOptionRule.QUERY_PORT)
+                    .required(DorisOptionRule.USERNAME, DorisOptionRule.PASSWORD)
+                    .build();
+
+    public static final OptionRule METADATA_RULE = OptionRule.builder().build();
+
+    public static final Set<String> DORIS_SYSTEM_DATABASES =
+            Sets.newHashSet("__internal_schema", "information_schema", "mysql");
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-doris/src/main/java/org/apache/seatunnel/datasource/plugin/doris/DorisDataSourceFactory.java

@@ -0,0 +1,47 @@
+/*
+ * 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.datasource.plugin.doris;
+
+import org.apache.seatunnel.datasource.plugin.api.DataSourceChannel;
+import org.apache.seatunnel.datasource.plugin.api.DataSourceFactory;
+import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginInfo;
+
+import com.google.auto.service.AutoService;
+import com.google.common.collect.Sets;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Set;
+
+@Slf4j
+@AutoService(DataSourceFactory.class)
+public class DorisDataSourceFactory implements DataSourceFactory {
+
+    @Override
+    public String factoryIdentifier() {
+        return DorisDataSourceConfig.PLUGIN_NAME;
+    }
+
+    @Override
+    public Set<DataSourcePluginInfo> supportedDataSources() {
+        return Sets.newHashSet(DorisDataSourceConfig.DORIS_DATASOURCE_PLUGIN_INFO);
+    }
+
+    @Override
+    public DataSourceChannel createChannel() {
+        return new DorisDataSourceChannel();
+    }
+}