[Improve] sink mongodb schema is not required (#8887)

Co-authored-by: zhouyq <[email protected]>

Author: uniding

Diff for: docs/en/connector-v2/sink/MongoDB.md

@@ -65,7 +65,6 @@ The following table lists the field data type mapping from MongoDB BSON type to
 | 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.                                                                             |
-| schema                | String   | Yes      | -       | MongoDB's BSON and seatunnel data structure mapping.                                                                         |
 | 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.                                                     |
@@ -113,12 +112,6 @@ sink {
     uri = mongodb://user:[email protected]:27017
     database = "test"
     collection = "test"
-    schema = {
-      fields {
-        _id = string
-        c_bigint = bigint
-      }
-    }
   }
 }
 ```
@@ -175,13 +168,6 @@ sink {
     collection = "users"
     buffer-flush.max-rows = 2000
     buffer-flush.interval = 1000
-    schema = {
-      fields {
-        _id = string
-        id = bigint
-        status = string
-      }
-    }
   }
 }
 ```
@@ -208,13 +194,6 @@ sink {
     collection = "users"
     upsert-enable = true
     primary-key = ["name","status"]
-    schema = {
-      fields {
-        _id = string
-        name = string
-        status = string
-      }
-    }
   }
 }
 ```

Diff for: seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSinkFactory.java

@@ -18,7 +18,6 @@
 package org.apache.seatunnel.connectors.seatunnel.mongodb.sink;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
-import org.apache.seatunnel.api.options.ConnectorCommonOptions;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
 import org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbConfig;
@@ -37,11 +36,7 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(
-                        MongodbConfig.URI,
-                        MongodbConfig.DATABASE,
-                        MongodbConfig.COLLECTION,
-                        ConnectorCommonOptions.SCHEMA)
+                .required(MongodbConfig.URI, MongodbConfig.DATABASE, MongodbConfig.COLLECTION)
                 .optional(
                         MongodbConfig.BUFFER_FLUSH_INTERVAL,
                         MongodbConfig.BUFFER_FLUSH_MAX_ROWS,

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/cdcIT/fake_cdc_sink_mongodb.conf

@@ -64,12 +64,5 @@ sink {
     database = "test_db"
     collection = "test_cdc_table"
     primary-key = ["c_int"]
-    schema = {
-      fields {
-        c_int = int
-        name = string
-        score = int
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/cdcIT/fake_cdc_upsert_sink_mongodb.conf

@@ -77,12 +77,5 @@ sink {
     collection = "test_cdc_table"
     upsert-enable = true
     primary-key = ["c_int"]
-    schema = {
-      fields {
-        c_int = int
-        name = string
-        score = int
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/compatibleParametersIT/fake_source_to_update_mongodb.conf

@@ -71,33 +71,5 @@ sink {
     // compatible parameters
     upsert-key = ["c_int"]
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_bytes = bytes
-        c_date = date
-        c_decimal = "decimal(33, 18)"
-        c_timestamp = timestamp
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-          c_bytes = bytes
-          c_date = date
-          c_decimal = "decimal(33, 18)"
-          c_timestamp = timestamp
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/compatibleParametersIT/mongodb_matchQuery_source_to_assert.conf

@@ -66,25 +66,5 @@ sink {
     database = "test_db"
     collection = "test_match_op_result_db"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/fake_source_to_mongodb.conf

@@ -67,33 +67,5 @@ sink {
     uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true"
     database = "test_db"
     collection = "test_source_sink_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_bytes = bytes
-        c_date = date
-        c_decimal = "decimal(33, 18)"
-        c_timestamp = timestamp
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-          c_bytes = bytes
-          c_date = date
-          c_decimal = "decimal(33, 18)"
-          c_timestamp = timestamp
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/flatIT/fake_source_to_flat_mongodb.conf

@@ -67,33 +67,5 @@ sink {
     database = "test_db"
     collection = "test_flat_table"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_bytes = bytes
-        c_date = date
-        c_decimal = "decimal(33, 18)"
-        c_timestamp = timestamp
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-          c_bytes = bytes
-          c_date = date
-          c_decimal = "decimal(33, 18)"
-          c_timestamp = timestamp
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/matchIT/mongodb_matchProjection_source_to_assert.conf

@@ -64,25 +64,5 @@ sink {
     database = "test_db"
     collection = "test_match_op_result_db"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
-
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/matchIT/mongodb_matchQuery_source_to_assert.conf

@@ -65,25 +65,5 @@ sink {
     database = "test_db"
     collection = "test_match_op_result_db"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_double_value.conf

@@ -64,25 +64,5 @@ sink {
     database = "test_db"
     collection = "test_double_op_db_result"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_null_value.conf

@@ -64,26 +64,5 @@ sink {
     uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true"
     database = "test_db"
     collection = "test_null_op_db_result"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
-
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/splitIT/mongodb_split_key_source_to_assert.conf

@@ -65,25 +65,5 @@ sink {
     database = "test_db"
     collection = "test_split_op_result_db"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/splitIT/mongodb_split_size_source_to_assert.conf

@@ -66,26 +66,6 @@ sink {
     database = "test_db"
     collection = "test_split_op_result_db"
     plugin_input = "mongodb_table"
-    schema = {
-      fields {
-        c_map = "map<string, string>"
-        c_array = "array<int>"
-        c_string = string
-        c_boolean = boolean
-        c_int = int
-        c_bigint = bigint
-        c_double = double
-        c_row = {
-          c_map = "map<string, string>"
-          c_array = "array<int>"
-          c_string = string
-          c_boolean = boolean
-          c_int = int
-          c_bigint = bigint
-          c_double = double
-        }
-      }
-    }
   }
 
 }