[feature][datasource] Add Mongodb datasource

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

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

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

@@ -0,0 +1,64 @@
+<?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-mongodb</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>datasource-plugins-api</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>mongodb-driver-sync</artifactId>
+            <version>4.7.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </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>

seatunnel-datasource/seatunnel-datasource-plugins/datasource-mongodb/src/main/java/com/apache/seatunnel/datasource/plugin/mongodb/MongoDataSoueceChannel.java

@@ -0,0 +1,106 @@
+/*
+ * 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.apache.seatunnel.datasource.plugin.mongodb;
+
+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.commons.lang3.StringUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoIterable;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+@Slf4j
+public class MongoDataSoueceChannel implements DataSourceChannel {
+
+    private static final String DATABASE = "default";
+
+    @Override
+    public OptionRule getDataSourceOptions(@NonNull String pluginName) {
+        return MongoOptionRule.optionRule();
+    }
+
+    @Override
+    public OptionRule getDatasourceMetadataFieldsByDataSourceName(@NonNull String pluginName) {
+        return MongoOptionRule.metadataRule();
+    }
+
+    public List<String> getTables(
+            @NonNull String pluginName,
+            Map<String, String> requestParams,
+            String database,
+            Map<String, String> options) {
+        checkArgument(StringUtils.equalsIgnoreCase(database, DATABASE), "database must be default");
+
+        return Collections.emptyList();
+    }
+
+    @Override
+    public List<String> getDatabases(
+            @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
+        return ImmutableList.of(DATABASE);
+    }
+
+    @Override
+    public List<TableField> getTableFields(
+            @NonNull String pluginName,
+            @NonNull Map<String, String> requestParams,
+            @NonNull String database,
+            @NonNull String table) {
+        checkArgument(StringUtils.equalsIgnoreCase(database, DATABASE), "database must be default");
+        return Collections.emptyList();
+    }
+
+    @Override
+    public boolean checkDataSourceConnectivity(
+            @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
+
+        try (MongoClient mongoClient = createMongoClient(requestParams)) {
+            // Verify if the connection to mongodb was successful
+            MongoIterable<String> databaseNames = mongoClient.listDatabaseNames();
+            if (databaseNames.iterator().hasNext()) {
+                log.info("mongoDB connection successful");
+                return true;
+            } else {
+                return false;
+            }
+        } catch (Exception e) {
+            throw new DataSourcePluginException(
+                    "check MongoDB connectivity failed, " + e.getMessage(), e);
+        }
+    }
+
+    // Resolve the URI in requestParams of Map type
+    private MongoClient createMongoClient(Map<String, String> requestParams) {
+
+        return MongoClients.create(
+                MongoRequestParamsUtils.parseStringFromRequestParams(requestParams));
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-mongodb/src/main/java/com/apache/seatunnel/datasource/plugin/mongodb/MongoDataSourceFactory.java

@@ -0,0 +1,57 @@
+/*
+ * 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.apache.seatunnel.datasource.plugin.mongodb;
+
+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 org.apache.seatunnel.datasource.plugin.api.DatasourcePluginTypeEnum;
+
+import com.google.auto.service.AutoService;
+import com.google.common.collect.Sets;
+
+import java.util.Set;
+
+@AutoService(DataSourceFactory.class)
+public class MongoDataSourceFactory implements DataSourceFactory {
+    public static final String MONGO_PLUGIN_NAME = "MongoDB";
+    public static final String MONGO_PLUGIN_ICON = "MongoDB";
+    public static final String MONGO_PLUGIN_VERSION = "1.0.0";
+
+    @Override
+    public String factoryIdentifier() {
+        return MONGO_PLUGIN_NAME;
+    }
+
+    @Override
+    public Set<DataSourcePluginInfo> supportedDataSources() {
+        return Sets.newHashSet(
+                DataSourcePluginInfo.builder()
+                        .name(MONGO_PLUGIN_NAME)
+                        .icon(MONGO_PLUGIN_ICON)
+                        .version(MONGO_PLUGIN_VERSION)
+                        .supportVirtualTables(true)
+                        .type(DatasourcePluginTypeEnum.NO_STRUCTURED.getCode())
+                        .build());
+    }
+
+    @Override
+    public DataSourceChannel createChannel() {
+        return new MongoDataSoueceChannel();
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-mongodb/src/main/java/com/apache/seatunnel/datasource/plugin/mongodb/MongoOptionRule.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 com.apache.seatunnel.datasource.plugin.mongodb;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+
+public class MongoOptionRule {
+
+    public static final Option<String> URI =
+            Options.key("uri")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The MongoDB connection uri.");
+
+    public static final Option<String> DATABASE =
+            Options.key("database")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The name of MongoDB database to read or write.");
+
+    public static final Option<String> COLLECTION =
+            Options.key("collection")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The name of MongoDB collection to read or write.");
+
+    public static OptionRule optionRule() {
+        return OptionRule.builder().required(URI).build();
+    }
+
+    public static OptionRule metadataRule() {
+        return OptionRule.builder().required(DATABASE, COLLECTION).build();
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-mongodb/src/main/java/com/apache/seatunnel/datasource/plugin/mongodb/MongoRequestParamsUtils.java

@@ -0,0 +1,33 @@
+/*
+ * 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.apache.seatunnel.datasource.plugin.mongodb;
+
+import java.util.Map;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+public class MongoRequestParamsUtils {
+
+    public static String parseStringFromRequestParams(Map<String, String> requestParams) {
+        checkArgument(
+                requestParams.containsKey(MongoOptionRule.URI.key()),
+                String.format("Missing %s in requestParams", MongoOptionRule.URI.key()));
+
+        return requestParams.get(MongoOptionRule.URI.key());
+    }
+}

seatunnel-datasource/seatunnel-datasource-plugins/datasource-plugins-api/src/main/java/org/apache/seatunnel/datasource/plugin/api/DataSourceChannel.java

@@ -28,6 +28,8 @@
 import java.sql.Connection;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 public interface DataSourceChannel {
 
@@ -71,11 +73,19 @@ List<TableField> getTableFields(
             @NonNull String database,
             @NonNull String table);
 
-    Map<String, List<TableField>> getTableFields(
+    default Map<String, List<TableField>> getTableFields(
             @NonNull String pluginName,
             @NonNull Map<String, String> requestParams,
             @NonNull String database,
-            @NonNull List<String> tables);
+            @NonNull List<String> tables) {
+        return tables.parallelStream()
+                .collect(
+                        Collectors.toMap(
+                                Function.identity(),
+                                table ->
+                                        getTableFields(
+                                                pluginName, requestParams, database, table)));
+    }
 
     /**
      * just check metadata field is right and used by virtual table

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

@@ -44,6 +44,7 @@
         <module>datasource-s3</module>
         <module>datasource-sqlserver-cdc</module>
         <module>datasource-jdbc-tidb</module>
+        <module>datasource-mongodb</module>
     </modules>
 
 </project>

seatunnel-server/seatunnel-app/src/main/bin/download_datasource.sh

@@ -43,6 +43,7 @@ datasource_list=(
   "datasource-s3"
   "datasource-sqlserver-cdc"
   "datasource-starrocks"
+  "datasource-mongodb"
 )
 
 # the datasource default version is 1.0.0, you can also choose a custom version. eg: 1.1.2:  sh install-datasource.sh 2.1.2