feat: mongo parse strong schema (#20100)

Signed-off-by: cl <[email protected]>
Signed-off-by: cailue <[email protected]>
Signed-off-by: 蔡略 <[email protected]>

Author: ClSlaid

e2e_test/source_legacy/basic/kafka.slt

@@ -386,6 +386,20 @@ WITH (
 	topic = 'debezium_mongo_json_customers')
 FORMAT DEBEZIUM_MONGO ENCODE JSON
 
+statement ok
+CREATE TABLE mongo_customers_strong_schema (
+	_id BIGINT PRIMARY KEY,
+	email VARCHAR,
+	first_name VARCHAR,
+	last_name VARCHAR
+)
+WITH (
+	connector = 'kafka',
+	properties.bootstrap.server = 'message_queue:29092',
+	topic = 'debezium_mongo_json_customers'
+)
+FORMAT DEBEZIUM_MONGO ENCODE JSON ( strong_schema = true )
+
 statement ok
 CREATE TABLE mongo_customers_no_schema_field (
 	_id BIGINT PRIMARY KEY,
@@ -397,6 +411,20 @@ WITH (
 	topic = 'debezium_mongo_json_customers_no_schema_field')
 FORMAT DEBEZIUM_MONGO ENCODE JSON
 
+statement ok
+CREATE TABLE mongo_customers_no_schema_field_strong_schema (
+	_id BIGINT PRIMARY KEY,
+	email VARCHAR,
+	first_name VARCHAR,
+	last_name VARCHAR
+)
+WITH (
+	connector = 'kafka',
+	properties.bootstrap.server = 'message_queue:29092',
+	topic = 'debezium_mongo_json_customers_no_schema_field'
+)
+FORMAT DEBEZIUM_MONGO ENCODE JSON ( strong_schema = true )
+
 statement ok
 CREATE TABLE upsert_students_default_key (
     "ID" INT,
@@ -749,6 +777,20 @@ ORDER BY
 1003    {"_id": {"$numberLong": "1003"}, "email": "[email protected]", "first_name": "Edward", "last_name": "Walker"}
 1004    {"_id": {"$numberLong": "1004"}, "email": "[email protected]", "first_name": "Anne", "last_name": "Kretchmar"}
 
+query II
+SELECT
+	*
+FROM
+	mongo_customers_strong_schema
+ORDER BY
+	_id;
+----
+1001 [email protected] Sally Thomas
+1002 [email protected] George Bailey
+1003 [email protected] Edward Walker
+1004 [email protected] Anne Kretchmar
+
+
 query II
 SELECT
 	*
@@ -763,6 +805,19 @@ ORDER BY
 1004    {"_id": {"$numberLong": "1004"}, "email": "[email protected]", "first_name": "Anne", "last_name": "Kretchmar"}
 
 
+query II
+SELECT
+	*
+FROM
+	mongo_customers_no_schema_field_strong_schema
+ORDER BY
+	_id;
+----
+1001 [email protected] Sally Thomas
+1002 [email protected] George Bailey
+1003 [email protected] Edward Walker
+1004 [email protected] Anne Kretchmar
+
 query II
 SELECT
 	"ID", "firstName", "lastName", "age", "height", "weight"

src/connector/src/parser/config.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
+use std::collections::{BTreeMap, HashSet};
 
 use risingwave_common::bail;
 use risingwave_common::secret::LocalSecretManager;
@@ -27,6 +27,7 @@ use crate::error::ConnectorResult;
 use crate::parser::PROTOBUF_MESSAGES_AS_JSONB;
 use crate::schema::AWS_GLUE_SCHEMA_ARN_KEY;
 use crate::schema::schema_registry::SchemaRegistryAuth;
+use crate::source::cdc::CDC_MONGODB_STRONG_SCHEMA_KEY;
 use crate::source::{SourceColumnDesc, SourceEncode, SourceFormat, extract_source_struct};
 
 /// Note: this is created in `SourceReader::build_stream`
@@ -60,7 +61,7 @@ pub enum EncodingProperties {
     Protobuf(ProtobufProperties),
     Csv(CsvProperties),
     Json(JsonProperties),
-    MongoJson,
+    MongoJson(MongoProperties),
     Bytes(BytesProperties),
     Parquet,
     Native,
@@ -262,10 +263,8 @@ impl SpecificParserConfig {
                 )?,
             }),
             (SourceFormat::DebeziumMongo, SourceEncode::Json) => {
-                EncodingProperties::Json(JsonProperties {
-                    use_schema_registry: false,
-                    timestamptz_handling: None,
-                })
+                let props = MongoProperties::from(&format_encode_options_with_secret);
+                EncodingProperties::MongoJson(props)
             }
             (SourceFormat::Plain, SourceEncode::Bytes) => {
                 EncodingProperties::Bytes(BytesProperties { column_name: None })
@@ -350,3 +349,22 @@ pub struct JsonProperties {
 pub struct BytesProperties {
     pub column_name: Option<String>,
 }
+
+#[derive(Debug, Default, Clone)]
+pub struct MongoProperties {
+    pub strong_schema: bool,
+}
+
+impl MongoProperties {
+    pub fn new(strong_schema: bool) -> Self {
+        Self { strong_schema }
+    }
+}
+impl From<&BTreeMap<String, String>> for MongoProperties {
+    fn from(config: &BTreeMap<String, String>) -> Self {
+        let strong_schema = config
+            .get(CDC_MONGODB_STRONG_SCHEMA_KEY)
+            .is_some_and(|k| k.eq_ignore_ascii_case("true"));
+        Self { strong_schema }
+    }
+}