[FLINK-38236][mongodb] Add metadata support for full document in MongoDB CDC connector

Author: Hisoka-X

docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md

@@ -383,6 +383,11 @@ MongoDB 的更改事件记录在消息之前没有更新。因此，我们只能
       <td>TIMESTAMP_LTZ(3) NOT NULL</td>
       <td>它指示在数据库中进行更改的时间。 <br>如果记录是从表的快照而不是改变流中读取的，该值将始终为0。</td>
     </tr>
+    <tr>
+      <td>full_document</td>
+      <td>STRING</td>
+      <td>变更事件的完整文档 JSON 字符串原始数据。对于 insert 事件，这是新文档。对于 update 事件，这是更新后的完整文档。对于 delete 事件，该值为 null。</td>
+    </tr>
     <tr>
       <td>row_kind</td>
       <td>STRING NOT NULL</td>
@@ -398,6 +403,7 @@ CREATE TABLE products (
     db_name         STRING METADATA FROM 'database_name' VIRTUAL,
     collection_name STRING METADATA  FROM 'collection_name' VIRTUAL,
     operation_ts    TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
+    raw_data        STRING METADATA FROM 'full_document' VIRTUAL,
     operation       STRING METADATA FROM 'row_kind' VIRTUAL,
     _id             STRING, // 必须声明
     name            STRING,

docs/content/docs/connectors/flink-sources/mongodb-cdc.md

@@ -408,6 +408,11 @@ The following format metadata can be exposed as read-only (VIRTUAL) columns in a
       <td>TIMESTAMP_LTZ(3) NOT NULL</td>
       <td>It indicates the time that the change was made in the database. <br>If the record is read from snapshot of the table instead of the change stream, the value is always 0.</td>
     </tr>
+    <tr>
+      <td>full_document</td>
+      <td>STRING</td>
+      <td>The full document of the change event as a JSON string raw data. For insert events, this is the new document. For update events, this is the full document after the update. For delete events, this is null.</td>
+    </tr>
     <tr>
       <td>row_kind</td>
       <td>STRING NOT NULL</td>
@@ -424,6 +429,7 @@ CREATE TABLE products (
     db_name         STRING METADATA FROM 'database_name' VIRTUAL,
     collection_name STRING METADATA  FROM 'collection_name' VIRTUAL,
     operation_ts    TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
+    raw_data        STRING METADATA FROM 'full_document' VIRTUAL,
     operation       STRING METADATA FROM 'row_kind' VIRTUAL,
     _id             STRING, // must be declared
     name            STRING,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBReadableMetadata.java

@@ -85,6 +85,21 @@ public Object read(SourceRecord record) {
                 }
             }),
 
+    /** It indicates the full document as string raw data. */
+    FULL_DOCUMENT(
+            "full_document",
+            DataTypes.STRING().nullable(),
+            new MetadataConverter() {
+                private static final long serialVersionUID = 1L;
+
+                @Override
+                public Object read(SourceRecord record) {
+                    Struct value = (Struct) record.value();
+                    String fullDocString = value.getString(MongoDBEnvelope.FULL_DOCUMENT_FIELD);
+                    return fullDocString != null ? StringData.fromString(fullDocString) : null;
+                }
+            }),
+
     /**
      * It indicates the row kind of the changelog. '+I' means INSERT message, '-D' means DELETE
      * message, '-U' means UPDATE_BEFORE message and '+U' means UPDATE_AFTER message

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java

@@ -89,6 +89,8 @@ class MongoDBTableFactoryTest {
                             Column.metadata("time", DataTypes.TIMESTAMP_LTZ(3), "op_ts", true),
                             Column.metadata(
                                     "_database_name", DataTypes.STRING(), "database_name", true),
+                            Column.metadata(
+                                    "_full_document", DataTypes.STRING(), "full_document", true),
                             Column.metadata("_row_kind", DataTypes.STRING(), "row_kind", true)),
                     Collections.emptyList(),
                     UniqueConstraint.primaryKey("pk", Collections.singletonList("_id")));
@@ -227,7 +229,7 @@ void testMetadataColumns() {
         DynamicTableSource actualSource = createTableSource(SCHEMA_WITH_METADATA, properties);
         MongoDBTableSource mongoDBSource = (MongoDBTableSource) actualSource;
         mongoDBSource.applyReadableMetadata(
-                Arrays.asList("op_ts", "database_name", "row_kind"),
+                Arrays.asList("op_ts", "database_name", "full_document", "row_kind"),
                 SCHEMA_WITH_METADATA.toSourceRowDataType());
         actualSource = mongoDBSource.copy();
 
@@ -262,7 +264,8 @@ void testMetadataColumns() {
                         SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue());
 
         expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
-        expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name", "row_kind");
+        expectedSource.metadataKeys =
+                Arrays.asList("op_ts", "database_name", "full_document", "row_kind");
 
         Assertions.assertThat(actualSource).isEqualTo(expectedSource);