[feature][datasource] Main add doris

seatunnel-datasource/seatunnel-datasource-client/src/main/java/org/apache/seatunnel/datasource/AbstractDataSourceClient.java

@@ -143,8 +143,7 @@ public OptionRule queryDataSourceFieldByName(String pluginName) {
     public OptionRule queryMetadataFieldByName(String pluginName) {
         updateClassLoader(pluginName);
         OptionRule datasourceMetadataFieldsByDataSourceName =
-                getDataSourceChannel(pluginName)
-                        .getDatasourceMetadataFieldsByDataSourceName(pluginName);
+                getDataSourceChannel(pluginName).getVirtualTableOptions(pluginName);
         classLoaderRestore();
         return datasourceMetadataFieldsByDataSourceName;
     }

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

@@ -92,6 +92,9 @@ public class DatasourceLoadConfig {
         classLoaderFactoryName.put(
                 "JDBC-STARROCKS",
                 "org.apache.seatunnel.datasource.plugin.starrocks.jdbc.StarRocksJdbcDataSourceFactory");
+        classLoaderFactoryName.put(
+                "DORIS",
+                "com.apache.seatunnel.datasource.plugin.doris.jdbc.DorisJdbcDataSourceFactory");
 
         classLoaderJarName.put("JDBC-ORACLE", "datasource-jdbc-oracle-");
         classLoaderJarName.put("JDBC-CLICKHOUSE", "datasource-jdbc-clickhouse-");
@@ -111,6 +114,7 @@ public class DatasourceLoadConfig {
         classLoaderJarName.put("STARROCKS", "datasource-starrocks-");
         classLoaderJarName.put("S3-REDSHIFT", "datasource-s3redshift-");
         classLoaderJarName.put("JDBC-STARROCKS", "datasource-jdbc-starrocks-");
+        classLoaderJarName.put("DORIS", "datasource-jdbc-doris-");
     }
 
     public static final Set<String> pluginSet =
@@ -127,6 +131,7 @@ public class DatasourceLoadConfig {
                     "MySQL-CDC",
                     "S3",
                     "SqlServer-CDC",
+                    "Doris",
                     "StarRocks");
 
     public static Map<String, DatasourceClassLoader> datasourceClassLoaders = new HashMap<>();

seatunnel-datasource/seatunnel-datasource-plugins/datasource-elasticsearch/src/main/java/org/apache/seatunnel/datasource/plugin/elasticsearch/ElasticSearchDataSourceChannel.java

@@ -49,7 +49,7 @@ public OptionRule getDataSourceOptions(@NonNull String pluginName) {
     }
 
     @Override
-    public OptionRule getDatasourceMetadataFieldsByDataSourceName(@NonNull String pluginName) {
+    public OptionRule getVirtualTableOptions(@NonNull String pluginName) {
         return ElasticSearchOptionRule.metadataRule();
     }
 

seatunnel-datasource/seatunnel-datasource-plugins/datasource-elasticsearch/src/test/java/org/apache/seatunnel/datasource/plugin/elasticsearch/ElasticSearchDataSourceChannelTest.java

@@ -63,8 +63,7 @@ void getDataSourceOptions() {
     @Test
     void getDatasourceMetadataFieldsByDataSourceName() {
         Assertions.assertNotNull(
-                ELASTIC_SEARCH_DATA_SOURCE_CHANNEL.getDatasourceMetadataFieldsByDataSourceName(
-                        PLUGIN_NAME));
+                ELASTIC_SEARCH_DATA_SOURCE_CHANNEL.getVirtualTableOptions(PLUGIN_NAME));
     }
 
     //    @Test

seatunnel-datasource/seatunnel-datasource-plugins/datasource-hive/src/main/java/com/whaleops/datasource/datasource/plugin/hive/HiveDataSourceChannel.java

@@ -44,7 +44,7 @@ public OptionRule getDataSourceOptions(@NonNull String pluginName) {
     }
 
     @Override
-    public OptionRule getDatasourceMetadataFieldsByDataSourceName(@NonNull String pluginName) {
+    public OptionRule getVirtualTableOptions(@NonNull String pluginName) {
         return HiveOptionRule.metadataRule();
     }
 

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-clickhouse/src/main/java/org/apache/seatunnel/datasource/plugin/clickhouse/jdbc/ClickhouseJdbcDataSourceChannel.java

@@ -49,7 +49,7 @@ public OptionRule getDataSourceOptions(@NonNull String pluginName) {
     }
 
     @Override
-    public OptionRule getDatasourceMetadataFieldsByDataSourceName(@NonNull String pluginName) {
+    public OptionRule getVirtualTableOptions(@NonNull String pluginName) {
         return ClickhouseDataSourceConfig.METADATA_RULE;
     }
 

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-common/pom.xml

@@ -0,0 +1,42 @@
+<?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-jdbc-common</artifactId>
+
+    <properties>
+        <hikaricp.version>4.0.3</hikaricp.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP</artifactId>
+            <version>${hikaricp.version}</version>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-common/src/main/java/com/whaleops/datasource/plugin/jdbc/common/JdbcDataSourceLRUConnectionPool.java

@@ -0,0 +1,165 @@
+/*
+ * 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 com.whaleops.datasource.plugin.jdbc.common;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalListener;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+@Slf4j
+public abstract class JdbcDataSourceLRUConnectionPool implements AutoCloseable {
+    private static final Duration SLOW_CONNECTION_CREATE_LIMIT = Duration.ofSeconds(2);
+    private static final Duration CLEAN_CONNECTION_INTERVAL = Duration.ofMinutes(30);
+    private static Cache<String, HikariDataSource> datasourceCache;
+    private static ScheduledExecutorService cleanUpExecutor;
+
+    public Connection getConnection(Map<String, String> requestParams) throws SQLException {
+        initialize();
+
+        HikariDataSource dataSource;
+        try {
+            dataSource =
+                    datasourceCache.get(
+                            datasourceCacheKey(requestParams),
+                            createDataSourceLoader(requestParams));
+        } catch (ExecutionException e) {
+            throw new RuntimeException(e);
+        }
+
+        long start = System.currentTimeMillis();
+        Connection connection = dataSource.getConnection();
+        long end = System.currentTimeMillis();
+        if (end - start > SLOW_CONNECTION_CREATE_LIMIT.toMillis()) {
+            log.warn(
+                    "Slow jdbc connection creation, cost {}ms for {}",
+                    end - start,
+                    dataSource.getJdbcUrl());
+        }
+        return connection;
+    }
+
+    protected abstract HikariDataSource createDataSource(Map<String, String> requestParams);
+
+    protected abstract String datasourceCacheKey(Map<String, String> requestParams);
+
+    private Callable<HikariDataSource> createDataSourceLoader(Map<String, String> requestParams) {
+        return () -> {
+            log.info("Creating jdbc datasource for {}", desensitize(requestParams));
+            HikariDataSource dataSource = createDataSource(requestParams);
+            log.info("Created jdbc datasource for {}", desensitize(requestParams));
+            return dataSource;
+        };
+    }
+
+    protected Map<String, String> desensitize(Map<String, String> requestParams) {
+        Map<String, String> newRequestParams = new HashMap<>(requestParams);
+        newRequestParams.remove("password");
+        return newRequestParams;
+    }
+
+    public void initialize() {
+        if (cleanUpExecutor == null) {
+            synchronized (JdbcDataSourceLRUConnectionPool.class) {
+                if (cleanUpExecutor == null) {
+                    log.info("Initializing jdbc datasource connection pool");
+                    datasourceCache =
+                            CacheBuilder.newBuilder()
+                                    .expireAfterAccess(
+                                            CLEAN_CONNECTION_INTERVAL.toMillis(),
+                                            TimeUnit.MILLISECONDS)
+                                    .removalListener(
+                                            (RemovalListener<String, HikariDataSource>)
+                                                    notification -> {
+                                                        try {
+                                                            log.info(
+                                                                    "Closing expired jdbc datasource connection pool for {}",
+                                                                    notification.getKey());
+                                                            notification.getValue().close();
+                                                            log.info(
+                                                                    "Closed expired jdbc datasource connection pool for {}",
+                                                                    notification.getKey());
+                                                        } catch (Exception e) {
+                                                            log.error(
+                                                                    "Failed to close expired jdbc datasource connection pool for {}",
+                                                                    notification.getKey(),
+                                                                    e);
+                                                        }
+                                                    })
+                                    .build();
+                    cleanUpExecutor =
+                            Executors.newSingleThreadScheduledExecutor(
+                                    new ThreadFactoryBuilder()
+                                            .setNameFormat(
+                                                    "datasource-plugin-jdbc-connection-pool-clean-up-%d")
+                                            .build());
+                    cleanUpExecutor.scheduleWithFixedDelay(
+                            () -> {
+                                log.debug("Cleaning up jdbc datasource cache");
+                                datasourceCache.cleanUp();
+                                log.debug("Cleaned up jdbc datasource cache");
+                            },
+                            CLEAN_CONNECTION_INTERVAL.toMillis(),
+                            CLEAN_CONNECTION_INTERVAL.toMillis(),
+                            TimeUnit.MILLISECONDS);
+                    log.info("Initialized jdbc datasource connection pool");
+                }
+            }
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (cleanUpExecutor != null) {
+            synchronized (JdbcDataSourceLRUConnectionPool.class) {
+                log.info("Closing jdbc datasource connection pool");
+                cleanUpExecutor.shutdownNow();
+                datasourceCache.cleanUp();
+                cleanUpExecutor = null;
+                log.info("Closed jdbc datasource connection pool");
+            }
+        }
+    }
+
+    public static String extractDbName(String jdbcUrl) {
+        String dbName = null;
+        Pattern pattern = Pattern.compile("[:/](\\w+)(?:\\?|&|$)");
+        Matcher matcher = pattern.matcher(jdbcUrl);
+
+        while (matcher.find()) {
+            dbName = matcher.group(1);
+        }
+
+        return dbName;
+    }
+}

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

@@ -0,0 +1,83 @@
+<?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-jdbc-doris</artifactId>
+
+    <properties>
+        <mysql-connector.version>8.0.28</mysql-connector.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>datasource-plugins-api</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>datasource-jdbc-common</artifactId>
+            <version>${project.version}</version>
+        </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>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <configuration>
+                    <skip>${e2e.dependency.skip}</skip>
+                    <appendOutput>true</appendOutput>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>