feat(datastream-mongodb-to-firestore): implement high-throughput shadowless migration

v2/datastream-mongodb-to-firestore/src/main/java/com/google/cloud/teleport/v2/templates/datastream/MongoDbChangeEventContext.java

@@ -94,6 +94,9 @@ private boolean isDlqReconsumed(JsonNode changeEvent) {
           .asText()
           .equalsIgnoreCase("true");
     }
+    if (changeEvent.has("_metadata_dlq_reconsumed")) {
+      return changeEvent.get("_metadata_dlq_reconsumed").asText().equalsIgnoreCase("true");
+    }
     return false;
   }
 
@@ -143,11 +146,13 @@ public MongoDbChangeEventContext(
         this.documentId = docIdVal.asDouble();
       } else if (docIdVal.isTextual()) {
         this.documentId = docIdVal.asText();
-      } else if (docIdVal.isObject()) {
-        if (docIdVal.has(OID_FIELD_NAME) && docIdVal.get(OID_FIELD_NAME).isTextual()) {
+      } else if (docIdVal.isObject() || docIdVal.isArray()) {
+        if (docIdVal.isObject()
+            && docIdVal.has(OID_FIELD_NAME)
+            && docIdVal.get(OID_FIELD_NAME).isTextual()) {
           this.documentId = new ObjectId(docIdVal.get(OID_FIELD_NAME).asText());
         } else {
-          // Support for generic Object-typed IDs or other complex BSON types (e.g., Binary)
+          // Support for generic Document (Map), Array (List), Binary, and composite BSON _id types
           Document wrapper = Document.parse("{ \"val\": " + docIdVal.toString() + " }");
           this.documentId = wrapper.get("val");
         }
@@ -182,7 +187,7 @@ public MongoDbChangeEventContext(
     this.isUpdateEvent = isUpdateEvent(changeEvent);
 
     this.jsonStringData = dataAsJsonString();
-    this.shadowDocument = generateShadowDocument();
+    this.shadowDocument = null;
     this.isDlqReconsumed = isDlqReconsumed(changeEvent);
   }
 
@@ -245,6 +250,13 @@ public boolean isUpdateEvent() {
   }
 
   public Document getShadowDocument() {
+    if (this.shadowDocument == null && this.shadowCollection != null) {
+      try {
+        return generateShadowDocument();
+      } catch (JsonProcessingException e) {
+        LOG.warn("Failed to generate shadow document: {}", e.getMessage());
+      }
+    }
     return shadowDocument;
   }
 
@@ -317,10 +329,35 @@ public String toString() {
     }
   }
 
+  @Override
   public boolean equals(Object other) {
+    if (this == other) {
+      return true;
+    }
     if (other instanceof MongoDbChangeEventContext) {
-      return Objects.equals(this.toString(), other.toString());
+      MongoDbChangeEventContext o = (MongoDbChangeEventContext) other;
+      return Objects.equals(this.dataCollection, o.dataCollection)
+          && Objects.equals(this.documentId, o.documentId)
+          && Objects.equals(this.timestampDoc, o.timestampDoc)
+          && this.isDeleteEvent == o.isDeleteEvent
+          && this.isUpdateEvent == o.isUpdateEvent
+          && this.isDlqReconsumed == o.isDlqReconsumed
+          && this.retryCount == o.retryCount
+          && Objects.equals(this.changeEvent, o.changeEvent);
     }
     return false;
   }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        dataCollection,
+        documentId,
+        timestampDoc,
+        isDeleteEvent,
+        isUpdateEvent,
+        isDlqReconsumed,
+        retryCount,
+        changeEvent);
+  }
 }

v2/datastream-mongodb-to-firestore/src/main/java/com/google/cloud/teleport/v2/transforms/LatestChangeEventCombineFn.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright (C) 2026 Google LLC
+ *
+ * Licensed 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 com.google.cloud.teleport.v2.transforms;
+
+import com.google.cloud.teleport.v2.templates.datastream.MongoDbChangeEventContext;
+import org.apache.beam.sdk.transforms.Combine.CombineFn;
+import org.bson.Document;
+
+/**
+ * High-performance CombineFn that compacts bursts of change events for the same document key into
+ * the latest monotonic event in worker RAM before shuffle and state evaluation.
+ */
+public class LatestChangeEventCombineFn
+    extends CombineFn<
+        MongoDbChangeEventContext, MongoDbChangeEventContext, MongoDbChangeEventContext> {
+
+  @Override
+  public MongoDbChangeEventContext createAccumulator() {
+    return null;
+  }
+
+  @Override
+  public MongoDbChangeEventContext addInput(
+      MongoDbChangeEventContext accumulator, MongoDbChangeEventContext input) {
+    if (accumulator == null) {
+      return input;
+    }
+    if (input == null) {
+      return accumulator;
+    }
+
+    Document accTsDoc = accumulator.getTimestampDoc();
+    Document inputTsDoc = input.getTimestampDoc();
+
+    long accTs = Utils.getTimestampNanos(accTsDoc);
+    long inputTs = Utils.getTimestampNanos(inputTsDoc);
+
+    // If input is strictly newer, or equal with DLQ reconsumption, prefer input
+    if (inputTs > accTs || (inputTs == accTs && input.getIsDlqReconsumed())) {
+      return input;
+    }
+    return accumulator;
+  }
+
+  @Override
+  public MongoDbChangeEventContext mergeAccumulators(
+      Iterable<MongoDbChangeEventContext> accumulators) {
+    MongoDbChangeEventContext winner = null;
+    long maxTs = Long.MIN_VALUE;
+
+    for (MongoDbChangeEventContext acc : accumulators) {
+      if (acc != null) {
+        long ts = Utils.getTimestampNanos(acc.getTimestampDoc());
+        if (winner == null || ts > maxTs || (ts == maxTs && acc.getIsDlqReconsumed())) {
+          maxTs = ts;
+          winner = acc;
+        }
+      }
+    }
+    return winner;
+  }
+
+  @Override
+  public MongoDbChangeEventContext extractOutput(MongoDbChangeEventContext accumulator) {
+    return accumulator;
+  }
+}