[Feature][Connector-V2] MongodbSink Support SaveMode

docs/en/connector-v2/sink/MongoDB.md

@@ -62,19 +62,21 @@ The following table lists the field data type mapping from MongoDB BSON type to
 
 ## Sink Options
 
-|         Name          |   Type   | Required | Default |                                                         Description                                                          |
-|-----------------------|----------|----------|---------|------------------------------------------------------------------------------------------------------------------------------|
-| uri                   | String   | Yes      | -       | The MongoDB standard connection uri. eg. mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true. |
-| database              | String   | Yes      | -       | The name of MongoDB database to read or write.                                                                               |
-| collection            | String   | Yes      | -       | The name of MongoDB collection to read or write.                                                                             |
-| buffer-flush.max-rows | String   | No       | 1000    | Specifies the maximum number of buffered rows per batch request.                                                             |
-| buffer-flush.interval | String   | No       | 30000   | Specifies the maximum interval of buffered rows per batch request, the unit is millisecond.                                  |
-| retry.max             | String   | No       | 3       | Specifies the max number of retry if writing records to database failed.                                                     |
-| retry.interval        | Duration | No       | 1000    | Specifies the retry time interval if writing records to database failed, the unit is millisecond.                            |
-| upsert-enable         | Boolean  | No       | false   | Whether to write documents via upsert mode.                                                                                  |
-| primary-key           | List     | No       | -       | The primary keys for upsert/update. Keys are in `["id","name",...]` format for properties.                                   |
-| transaction           | Boolean  | No       | false   | Whether to use transactions in MongoSink (requires MongoDB 4.2+).                                                            |
-| common-options        |          | No       | -       | Source plugin common parameters, please refer to [Source Common Options](../sink-common-options.md) for details              |
+| Name                  | Type     | Required | Default | Description                                                                                                                                                                                                                            |
+|-----------------------|----------|----------|--------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| uri                   | String   | Yes      | -      | The MongoDB standard connection uri. eg. mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true.                                                                                                           |
+| database              | String   | Yes      | -      | The name of MongoDB database to read or write.                                                                                                                                                                                         |
+| collection            | String   | Yes      | -      | The name of MongoDB collection to read or write.                                                                                                                                                                                       |
+| buffer-flush.max-rows | String   | No       | 1000   | Specifies the maximum number of buffered rows per batch request.                                                                                                                                                                       |
+| buffer-flush.interval | String   | No       | 30000  | Specifies the maximum interval of buffered rows per batch request, the unit is millisecond.                                                                                                                                            |
+| retry.max             | String   | No       | 3      | Specifies the max number of retry if writing records to database failed.                                                                                                                                                               |
+| retry.interval        | Duration | No       | 1000   | Specifies the retry time interval if writing records to database failed, the unit is millisecond.                                                                                                                                      |
+| upsert-enable         | Boolean  | No       | false  | Whether to write documents via upsert mode.                                                                                                                                                                                            |
+| primary-key           | List     | No       | -      | The primary keys for upsert/update. Keys are in `["id","name",...]` format for properties.                                                                                                                                             |
+| transaction           | Boolean  | No       | false  | Whether to use transactions in MongoSink (requires MongoDB 4.2+).                                                                                                                                                                      |
+| common-options        |          | No       | -      | Source plugin common parameters, please refer to [Source Common Options](../sink-common-options.md) for details                                                                                                                        |
+| data_save_mode        | String   | No       | APPEND_DATA       | The data saving mode of mongodb，Option introduction,`DROP_DATA`:The collection will be cleared before inserting data;`APPEND_DATA`:Append data ;`ERROR_WHEN_DATA_EXISTS`:An error will be reported if there is data in the collection. |
+
 
 ### Tips
 

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/catalog/MongodbCatalog.java

@@ -0,0 +1,211 @@
+/*
+ * 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.mongodb.catalog;
+
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
+import org.apache.seatunnel.api.table.catalog.exception.DatabaseAlreadyExistException;
+import org.apache.seatunnel.api.table.catalog.exception.DatabaseNotExistException;
+import org.apache.seatunnel.api.table.catalog.exception.TableAlreadyExistException;
+import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
+import org.apache.seatunnel.common.exception.CommonError;
+
+import org.bson.Document;
+
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MongodbCatalog implements Catalog {
+
+    private final String catalogName;
+    private final String baseUrl;
+    private transient MongoClient mongoClient;
+    private final String defaultDatabase;
+
+    public MongodbCatalog(String catalogName, String baseUrl, String defaultDatabase) {
+        this.catalogName = catalogName;
+        this.baseUrl = baseUrl;
+        this.defaultDatabase = defaultDatabase;
+    }
+
+    @Override
+    public void open() throws CatalogException {
+        try {
+            if (mongoClient == null) {
+                mongoClient = MongoClients.create(baseUrl);
+            }
+        } catch (Exception e) {
+            throw new CatalogException("Failed to open MongoDB Catalog: " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public String name() {
+        return catalogName;
+    }
+
+    @Override
+    public String getDefaultDatabase() throws CatalogException {
+        return defaultDatabase;
+    }
+
+    @Override
+    public boolean databaseExists(String databaseName) throws CatalogException {
+        try {
+            return listDatabases().contains(databaseName);
+        } catch (Exception e) {
+            throw new CatalogException("Failed to check database existence: " + databaseName, e);
+        }
+    }
+
+    @Override
+    public List<String> listDatabases() throws CatalogException {
+        try {
+            List<String> dbs = new ArrayList<>();
+            for (String name : mongoClient.listDatabaseNames()) {
+                dbs.add(name);
+            }
+            return dbs;
+        } catch (Exception e) {
+            throw new CatalogException("Failed to list databases", e);
+        }
+    }
+
+    @Override
+    public List<String> listTables(String databaseName)
+            throws CatalogException, DatabaseNotExistException {
+        if (!databaseExists(databaseName)) {
+            throw new DatabaseNotExistException(name(), databaseName);
+        }
+        try {
+            MongoDatabase db = mongoClient.getDatabase(databaseName);
+            return db.listCollectionNames().into(new ArrayList<>());
+        } catch (Exception e) {
+            throw new CatalogException("Failed to list tables for database: " + databaseName, e);
+        }
+    }
+
+    @Override
+    public boolean tableExists(TablePath tablePath) throws CatalogException {
+        try {
+            return listTables(tablePath.getDatabaseName()).contains(tablePath.getTableName());
+        } catch (DatabaseNotExistException e) {
+            return false;
+        }
+    }
+
+    @Override
+    public CatalogTable getTable(TablePath tablePath)
+            throws CatalogException, TableNotExistException {
+        throw CommonError.unsupportedOperation(name(), "get table with tablePath ");
+    }
+
+    @Override
+    public void createTable(TablePath tablePath, CatalogTable table, boolean ignoreIfExists)
+            throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
+        if (!databaseExists(tablePath.getDatabaseName())) {
+            throw new DatabaseNotExistException(name(), tablePath.getDatabaseName());
+        }
+        if (tableExists(tablePath)) {
+            if (ignoreIfExists) return;
+            throw new TableAlreadyExistException(name(), tablePath);
+        }
+        try {
+            MongoDatabase db = mongoClient.getDatabase(tablePath.getDatabaseName());
+            db.createCollection(tablePath.getTableName());
+        } catch (Exception e) {
+            throw new CatalogException(
+                    "Failed to create collection: " + tablePath.getFullName(), e);
+        }
+    }
+
+    @Override
+    public void dropTable(TablePath tablePath, boolean ignoreIfNotExists)
+            throws TableNotExistException, CatalogException {
+        if (!tableExists(tablePath)) {
+            if (ignoreIfNotExists) return;
+            throw new TableNotExistException(name(), tablePath);
+        }
+        try {
+            MongoDatabase db = mongoClient.getDatabase(tablePath.getDatabaseName());
+            db.getCollection(tablePath.getTableName()).drop();
+        } catch (Exception e) {
+            throw new CatalogException("Failed to drop collection: " + tablePath.getFullName(), e);
+        }
+    }
+
+    @Override
+    public void createDatabase(TablePath tablePath, boolean ignoreIfExists)
+            throws DatabaseAlreadyExistException, CatalogException {
+        throw CommonError.unsupportedOperation(name(), "create database ");
+    }
+
+    @Override
+    public void dropDatabase(TablePath tablePath, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException, CatalogException {
+        throw CommonError.unsupportedOperation(name(), "drop database ");
+    }
+
+    @Override
+    public void truncateTable(TablePath tablePath, boolean ignoreIfNotExists)
+            throws TableNotExistException, CatalogException {
+        try {
+            if (!tableExists(tablePath)) {
+                if (ignoreIfNotExists) {
+                    return;
+                }
+                throw new TableNotExistException(name(), tablePath);
+            }
+            MongoDatabase db = mongoClient.getDatabase(tablePath.getDatabaseName());
+            MongoCollection<Document> collection = db.getCollection(tablePath.getTableName());
+            collection.deleteMany(new Document());
+        } catch (Exception e) {
+            throw new CatalogException(
+                    "Failed to truncate collection: " + tablePath.getFullName(), e);
+        }
+    }
+
+    @Override
+    public boolean isExistsData(TablePath tablePath) {
+        try {
+            if (!tableExists(tablePath)) {
+                return false;
+            }
+            MongoDatabase db = mongoClient.getDatabase(tablePath.getDatabaseName());
+            MongoCollection<Document> collection = db.getCollection(tablePath.getTableName());
+            return collection.estimatedDocumentCount() > 0;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    @Override
+    public void close() throws CatalogException {
+        if (mongoClient != null) {
+            mongoClient.close();
+            mongoClient = null;
+        }
+    }
+}

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/catalog/MongodbCatalogFactory.java

@@ -0,0 +1,48 @@
+/*
+ * 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.mongodb.catalog;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.factory.CatalogFactory;
+import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbConfig;
+
+import com.google.auto.service.AutoService;
+
+import static org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbConfig.CONNECTOR_IDENTITY;
+
+@AutoService(Factory.class)
+public class MongodbCatalogFactory implements CatalogFactory {
+    @Override
+    public Catalog createCatalog(String catalogName, ReadonlyConfig options) {
+        return new MongodbCatalog(
+                catalogName, options.get(MongodbConfig.URI), options.get(MongodbConfig.DATABASE));
+    }
+
+    @Override
+    public String factoryIdentifier() {
+        return CONNECTOR_IDENTITY;
+    }
+
+    @Override
+    public OptionRule optionRule() {
+        return OptionRule.builder().required(MongodbConfig.URI, MongodbConfig.DATABASE).build();
+    }
+}

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/config/MongodbConfig.java

@@ -19,12 +19,18 @@
 
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.sink.DataSaveMode;
 
 import org.bson.json.JsonMode;
 import org.bson.json.JsonWriterSettings;
 
+import java.util.Arrays;
 import java.util.List;
 
+import static org.apache.seatunnel.api.sink.DataSaveMode.APPEND_DATA;
+import static org.apache.seatunnel.api.sink.DataSaveMode.DROP_DATA;
+import static org.apache.seatunnel.api.sink.DataSaveMode.ERROR_WHEN_DATA_EXISTS;
+
 public class MongodbConfig {
 
     public static final String CONNECTOR_IDENTITY = "MongoDB";
@@ -153,4 +159,12 @@ public class MongodbConfig {
 
     public static final Option<Boolean> TRANSACTION =
             Options.key("transaction").booleanType().defaultValue(false).withDescription(".");
+
+    public static final Option<DataSaveMode> DATA_SAVE_MODE =
+            Options.key("data_save_mode")
+                    .singleChoice(
+                            DataSaveMode.class,
+                            Arrays.asList(DROP_DATA, APPEND_DATA, ERROR_WHEN_DATA_EXISTS))
+                    .defaultValue(APPEND_DATA)
+                    .withDescription("The save mode of collection data");
 }