Streaming added full document before change support

src/main/java/com/mongodb/spark/sql/connector/config/ReadConfig.java

@@ -23,6 +23,7 @@
 import static java.util.Collections.unmodifiableList;
 
 import com.mongodb.client.model.changestream.FullDocument;
+import com.mongodb.client.model.changestream.FullDocumentBeforeChange;
 import com.mongodb.spark.sql.connector.exceptions.ConfigException;
 import com.mongodb.spark.sql.connector.read.partitioner.Partitioner;
 import java.util.HashMap;
@@ -288,6 +289,34 @@ static ParseMode fromString(final String userParseMode) {
 
   private static final String STREAM_LOOKUP_FULL_DOCUMENT_DEFAULT = FullDocument.DEFAULT.getValue();
 
+  /**
+   * Streaming full document <b>before change</b> configuration.
+   *
+   * <p>Determines what to return as the pre-image of the document during replace, update, or delete operations
+   * when using a MongoDB Change Stream.
+   *
+   * <p>Only applies if the MongoDB server is configured to capture pre-images.
+   * See: <a href="https://www.mongodb.com/docs/manual/changeStreams/#change-streams-with-document-pre--and-post-images">
+   * Change streams lookup full document before change</a> for further details.
+   *
+   * <p>Possible values:
+   * <ul>
+   *   <li><b>"default"</b> – Uses the server's default behavior for the <code>fullDocumentBeforeChange</code> field.</li>
+   *   <li><b>"off"</b> – Do not include the pre-image of the document in the change stream event.</li>
+   *   <li><b>"whenAvailable"</b> – Include the pre-image of the modified document if available; otherwise, omit it.</li>
+   *   <li><b>"required"</b> – Include the pre-image, and raise an error if it is not available.</li>
+   * </ul>
+   *
+   * <p>Configuration: {@value}
+   *
+   * <p>Default: "default" – the server's default behavior for the <code>fullDocumentBeforeChange</code> field.
+   */
+  public static final String STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_CONFIG =
+      "change.stream.lookup.full.document.before.change";
+
+  private static final String STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_DEFAULT =
+      FullDocumentBeforeChange.DEFAULT.getValue();
+
   enum StreamingStartupMode {
     LATEST,
     TIMESTAMP;
@@ -492,6 +521,16 @@ public FullDocument getStreamFullDocument() {
     }
   }
 
+  /** @return the stream full document before change configuration or 'default' if not set. */
+  public FullDocumentBeforeChange getStreamFullDocumentBeforeChange() {
+    try {
+      return FullDocumentBeforeChange.fromString(
+          getOrDefault(STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_CONFIG, STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_DEFAULT));
+    } catch (IllegalArgumentException e) {
+      throw new ConfigException(e);
+    }
+  }
+
   /** @return true if should drop any malformed rows */
   public boolean dropMalformed() {
     return parseMode == ParseMode.DROPMALFORMED;

src/main/java/com/mongodb/spark/sql/connector/read/MongoContinuousPartitionReader.java

@@ -196,6 +196,7 @@ private MongoChangeStreamCursor<BsonDocument> getCursor() {
       }
       changeStreamIterable
           .fullDocument(readConfig.getStreamFullDocument())
+          .fullDocumentBeforeChange(readConfig.getStreamFullDocumentBeforeChange())
           .comment(readConfig.getComment());
       changeStreamIterable = lastOffset.applyToChangeStreamIterable(changeStreamIterable);
 

src/main/java/com/mongodb/spark/sql/connector/read/MongoMicroBatchPartitionReader.java

@@ -185,6 +185,7 @@ private MongoChangeStreamCursor<BsonDocument> getCursor() {
       }
       changeStreamIterable
           .fullDocument(readConfig.getStreamFullDocument())
+          .fullDocumentBeforeChange(readConfig.getStreamFullDocumentBeforeChange())
           .comment(readConfig.getComment());
       if (partition.getStartOffsetTimestamp().getTime() >= 0) {
         changeStreamIterable.startAtOperationTime(partition.getStartOffsetTimestamp());

src/test/java/com/mongodb/spark/sql/connector/config/MongoConfigTest.java

@@ -28,6 +28,7 @@
 
 import com.mongodb.WriteConcern;
 import com.mongodb.client.model.changestream.FullDocument;
+import com.mongodb.client.model.changestream.FullDocumentBeforeChange;
 import com.mongodb.spark.sql.connector.exceptions.ConfigException;
 import java.util.HashMap;
 import java.util.Map;
@@ -309,6 +310,29 @@ void testReadConfigStreamFullDocument() {
     assertEquals(readConfig.getStreamFullDocument(), FullDocument.UPDATE_LOOKUP);
   }
 
+  @Test
+  void testReadConfigStreamFullDocumentBeforeChange() {
+    ReadConfig readConfig = MongoConfig.readConfig(CONFIG_MAP);
+    assertEquals(readConfig.getStreamFullDocumentBeforeChange(), FullDocumentBeforeChange.DEFAULT);
+
+    readConfig =
+        readConfig.withOption(ReadConfig.STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_CONFIG, "off");
+    assertEquals(readConfig.getStreamFullDocumentBeforeChange(), FullDocumentBeforeChange.OFF);
+
+    readConfig = readConfig.withOption(
+        ReadConfig.STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_CONFIG, "whenAvailable");
+    assertEquals(
+        readConfig.getStreamFullDocumentBeforeChange(), FullDocumentBeforeChange.WHEN_AVAILABLE);
+
+    readConfig = readConfig.withOption(
+        ReadConfig.STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_CONFIG, "required");
+    assertEquals(readConfig.getStreamFullDocumentBeforeChange(), FullDocumentBeforeChange.REQUIRED);
+
+    readConfig = readConfig.withOption(
+        ReadConfig.STREAM_LOOKUP_FULL_DOCUMENT_BEFORE_CHANGE_CONFIG, "INVALID");
+    assertThrows(ConfigException.class, readConfig::getStreamFullDocumentBeforeChange);
+  }
+
   @Test
   void testReadConfigSchemaHints() {
     ReadConfig readConfig =