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

Author: Hisoka-X

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

@@ -599,6 +599,7 @@ private void testMongoDBParallelSourceWithMetadataColumns(
                                 + " database_name STRING METADATA VIRTUAL,"
                                 + " collection_name STRING METADATA VIRTUAL,"
                                 + " row_kind STRING METADATA VIRTUAL,"
+                                + " full_document STRING METADATA FROM 'full_document' VIRTUAL,"
                                 + " primary key (_id) not enforced"
                                 + ") WITH ("
                                 + " 'connector' = 'mongodb-cdc',"
@@ -653,16 +654,32 @@ private void testMongoDBParallelSourceWithMetadataColumns(
         TableResult tableResult =
                 tEnv.executeSql(
                         "select database_name, collection_name, row_kind, "
-                                + "cid, name, address, phone_number from customers");
+                                + "cid, name, address, phone_number, full_document from customers");
         CloseableIterator<Row> iterator = tableResult.collect();
         JobID jobId = tableResult.getJobClient().get().getJobID();
         List<String> expectedSnapshotData = new ArrayList<>();
         for (int i = 0; i < captureCustomerCollections.length; i++) {
             expectedSnapshotData.addAll(snapshotForSingleTable);
         }
 
-        assertEqualsInAnyOrder(
-                expectedSnapshotData, fetchRows(iterator, expectedSnapshotData.size()));
+        List<String> rawSnapshotData = fetchRows(iterator, expectedSnapshotData.size());
+
+        // Strip full_document (last column) for precise matching of other metadata columns
+        List<String> snapshotWithoutFullDoc =
+                rawSnapshotData.stream()
+                        .map(MongoDBFullChangelogITCase::stripLastColumn)
+                        .collect(Collectors.toList());
+        assertEqualsInAnyOrder(expectedSnapshotData, snapshotWithoutFullDoc);
+
+        // Verify full_document is non-null and contains expected fields in snapshot records
+        for (String row : rawSnapshotData) {
+            Assertions.assertThat(row)
+                    .as("Snapshot full_document should contain 'cid'")
+                    .contains("\"cid\"");
+            Assertions.assertThat(row)
+                    .as("Snapshot full_document should contain 'name'")
+                    .contains("\"name\"");
+        }
 
         // second step: check the change stream data
         for (String collectionName : captureCustomerCollections) {
@@ -693,11 +710,50 @@ private void testMongoDBParallelSourceWithMetadataColumns(
         for (int i = 0; i < captureCustomerCollections.length; i++) {
             expectedChangeStreamData.addAll(changeEventsForSingleTable);
         }
-        List<String> actualChangeStreamData = fetchRows(iterator, expectedChangeStreamData.size());
-        assertEqualsInAnyOrder(expectedChangeStreamData, actualChangeStreamData);
+        List<String> rawChangeStreamData = fetchRows(iterator, expectedChangeStreamData.size());
+
+        // Strip full_document for precise matching of other metadata columns
+        List<String> changeStreamWithoutFullDoc =
+                rawChangeStreamData.stream()
+                        .map(MongoDBFullChangelogITCase::stripLastColumn)
+                        .collect(Collectors.toList());
+        assertEqualsInAnyOrder(expectedChangeStreamData, changeStreamWithoutFullDoc);
+
+        // Verify full_document in change stream: non-null for +I/+U, null for -D
+        for (String row : rawChangeStreamData) {
+            if (row.startsWith("+I") || row.startsWith("+U")) {
+                Assertions.assertThat(row)
+                        .as("Change stream full_document should contain 'name'")
+                        .contains("\"name\"");
+            }
+            if (row.startsWith("-D")) {
+                Assertions.assertThat(row)
+                        .as("Delete event full_document should be null")
+                        .endsWith(", null]");
+            }
+        }
+
         tableResult.getJobClient().get().cancel().get();
     }
 
+    /**
+     * Strip the last column (full_document) from a Row.toString() formatted string. Row format:
+     * "+I[col1, col2, ..., colN, full_document_json]" or "+I[col1, col2, ..., colN, null]".
+     */
+    private static String stripLastColumn(String row) {
+        // Find the last ", " before the full_document column value
+        // The full_document is a JSON string starting with {"  or null
+        int lastJsonStart = row.lastIndexOf(", {\"");
+        if (lastJsonStart > 0) {
+            return row.substring(0, lastJsonStart) + "]";
+        }
+        int lastNullStart = row.lastIndexOf(", null]");
+        if (lastNullStart > 0) {
+            return row.substring(0, lastNullStart) + "]";
+        }
+        return row;
+    }
+
     private void testMongoDBParallelSource(
             MongoDBTestUtils.FailoverType failoverType,
             MongoDBTestUtils.FailoverPhase failoverPhase,

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

@@ -471,6 +471,7 @@ void testMetadataColumns(boolean parallelismSnapshot) throws Exception {
                                 + " weight DECIMAL(10,3),"
                                 + " db_name STRING METADATA FROM 'database_name' VIRTUAL,"
                                 + " collection_name STRING METADATA VIRTUAL,"
+                                + " full_document STRING METADATA FROM 'full_document' VIRTUAL,"
                                 + " PRIMARY KEY (_id) NOT ENFORCED"
                                 + ") WITH ("
                                 + " 'connector' = 'mongodb-cdc',"
@@ -497,6 +498,7 @@ void testMetadataColumns(boolean parallelismSnapshot) throws Exception {
                         + " weight DECIMAL(10,3),"
                         + " database_name STRING,"
                         + " collection_name STRING,"
+                        + " full_document STRING,"
                         + " PRIMARY KEY (_id) NOT ENFORCED"
                         + ") WITH ("
                         + " 'connector' = 'values',"
@@ -549,6 +551,30 @@ void testMetadataColumns(boolean parallelismSnapshot) throws Exception {
 
         waitForSinkSize("meta_sink", 16);
 
+        List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("meta_sink");
+
+        // Verify original metadata columns (database_name, collection_name) by stripping
+        // the full_document column (last column) from each row for precise matching
+        List<String> actualWithoutFullDoc =
+                actual.stream()
+                        .map(
+                                row -> {
+                                    // Row format: +I(col1,col2,...,colN,full_document_json)
+                                    // full_document is the last column, strip it
+                                    int lastCommaBeforeJson = row.lastIndexOf(",{\"_id\"");
+                                    if (lastCommaBeforeJson > 0) {
+                                        return row.substring(0, lastCommaBeforeJson) + ")";
+                                    }
+                                    // For -D events, full_document is null
+                                    int lastCommaBeforeNull = row.lastIndexOf(",null)");
+                                    if (lastCommaBeforeNull > 0) {
+                                        return row.substring(0, lastCommaBeforeNull) + ")";
+                                    }
+                                    return row;
+                                })
+                        .sorted()
+                        .collect(Collectors.toList());
+
         List<String> expected =
                 Stream.of(
                                 "+I(100000000000000000000101,scooter,Small 2-wheel scooter,3.140,%s,products)",
@@ -571,8 +597,21 @@ void testMetadataColumns(boolean parallelismSnapshot) throws Exception {
                         .sorted()
                         .collect(Collectors.toList());
 
-        List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("meta_sink");
-        Assertions.assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
+        Assertions.assertThat(actualWithoutFullDoc).containsExactlyInAnyOrderElementsOf(expected);
+
+        // Verify full_document metadata: non-null for all events, contains expected fields
+        for (String row : actual) {
+            Assertions.assertThat(row)
+                    .as("full_document should contain 'name' field")
+                    .contains("\"name\"");
+            Assertions.assertThat(row)
+                    .as("full_document should contain 'description' field")
+                    .contains("\"description\"");
+            Assertions.assertThat(row)
+                    .as("full_document should contain 'weight' field")
+                    .contains("\"weight\"");
+        }
+
         result.getJobClient().get().cancel().get();
     }