Azure · kushagraThapar · Sep 29, 2022 · Jul 27, 2022 · Jul 27, 2022 · Jul 27, 2022
@@ -64,12 +64,12 @@ Used to influence the json serialization/deserialization behavior
 | `spark.cosmos.serialization.dateTimeConversionMode`     | `Default`    | The date/time conversion mode (`Default`, `AlwaysEpochMilliseconds`, `AlwaysEpochMillisecondsWithSystemDefaultTimezone`). With `Default` the standard Spark 3.* behavior is used (`java.sql.Date`/`java.time.LocalDate` are converted to EpochDay, `java.sql.Timestamp`/`java.time.Instant` are converted to MicrosecondsFromEpoch). With `AlwaysEpochMilliseconds` the same behavior the Cosmos DB connector for Spark 2.4 used is applied - `java.sql.Date`, `java.time.LocalDate`, `java.sql.Timestamp` and `java.time.Instant` are converted to MillisecondsFromEpoch. The behavior for `AlwaysEpochMillisecondsWithSystemDefaultTimezone` is identical with `AlwaysEpochMilliseconds` except that it will assume System default time zone / Spark session time zone (specified via `spark.sql.session.timezone`) instead of UTC when the date/time to be parsed has no explicit time zone.|
 
 #### Change feed (only for Spark-Streaming using `cosmos.oltp.changeFeed` data source, which is read-only) configuration
-| Config Property Name                            | Default       | Description |
-| :---                                            | :----         | :---        | 
-| `spark.cosmos.changeFeed.startFrom`               | `Beginning`   | ChangeFeed Start from settings (`Now`, `Beginning`  or a certain point in time (UTC) for example `2020-02-10T14:15:03`) - the default value is `Beginning`. If the write config contains a `checkpointLocation` and any checkpoints exist, the stream is always continued independent of the `spark.cosmos.changeFeed.startFrom` settings - you need to change `checkpointLocation` or delete checkpoints to restart the stream if that is the intention. | 
-| `spark.cosmos.changeFeed.mode`                    | `Incremental` | ChangeFeed mode (`Incremental` or `FullFidelity`) - NOTE: `FullFidelity` is in experimental state right now. It requires that the subscription/account has been enabled for the private preview and there are known breaking changes that will happen for `FullFidelity` (schema of the returned documents). It is recommended to only use `FullFidelity` for non-production scenarios at this point. | 
-| `spark.cosmos.changeFeed.itemCountPerTriggerHint` | None          | Approximate maximum number of items read from change feed for each micro-batch/trigger |
-| `spark.cosmos.changeFeed.batchCheckpointLocation` | None          | Can be used to generate checkpoints when using change feed queries in batch mode - and proceeding on the next iteration where the previous left off. |
+| Config Property Name                            | Default                     | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
+| :---                                            |:----------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| 
+| `spark.cosmos.changeFeed.startFrom`               | `Beginning`                 | ChangeFeed Start from settings (`Now`, `Beginning`  or a certain point in time (UTC) for example `2020-02-10T14:15:03`) - the default value is `Beginning`. If the write config contains a `checkpointLocation` and any checkpoints exist, the stream is always continued independent of the `spark.cosmos.changeFeed.startFrom` settings - you need to change `checkpointLocation` or delete checkpoints to restart the stream if that is the intention.                              | 
+| `spark.cosmos.changeFeed.mode`                    | `Incremental/LatestVersion` | ChangeFeed mode (`Incremental/LatestVersion` or `FullFidelity/AllVersionsAndDeletes`) - NOTE: `FullFidelity/AllVersionsAndDeletes` is in experimental state right now. It requires that the subscription/account has been enabled for the private preview and there are known breaking changes that will happen for `FullFidelity/AllVersionsAndDeletes` (schema of the returned documents). It is recommended to only use `FullFidelity/AllVersionsAndDeletes` for non-production scenarios at this point. | 
+| `spark.cosmos.changeFeed.itemCountPerTriggerHint` | None                        | Approximate maximum number of items read from change feed for each micro-batch/trigger                                                                                                                                                                                                                                                                                                                                                                                                 |
+| `spark.cosmos.changeFeed.batchCheckpointLocation` | None                        | Can be used to generate checkpoints when using change feed queries in batch mode - and proceeding on the next iteration where the previous left off.                                                                                                                                                                                                                                                                                                                                   |
 
 #### Json conversion configuration
 | Config Property Name      | Default | Description |

diff --git a/...12/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala b/...12/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala
@@ -5,10 +5,10 @@ package com.azure.cosmos.implementation
 
 import com.azure.cosmos.{CosmosAsyncClient, CosmosClientBuilder, DirectConnectionConfig, SparkBridgeInternal}
 import com.azure.cosmos.implementation.ImplementationBridgeHelpers.CosmosClientBuilderHelper
-import com.azure.cosmos.implementation.changefeed.common.{ChangeFeedMode, ChangeFeedStartFromInternal, ChangeFeedState, ChangeFeedStateV1}
+import com.azure.cosmos.implementation.changefeed.common.{ChangeFeedStartFromInternal, ChangeFeedState, ChangeFeedStateV1}
 import com.azure.cosmos.implementation.query.CompositeContinuationToken
 import com.azure.cosmos.implementation.routing.Range
-import com.azure.cosmos.models.{FeedRange, PartitionKey, SparkModelBridgeInternal}
+import com.azure.cosmos.models.{ChangeFeedMode, FeedRange, PartitionKey, SparkModelBridgeInternal}
 import com.azure.cosmos.spark.{ChangeFeedOffset, NormalizedRange}
 import com.azure.cosmos.spark.diagnostics.BasicLoggingTrait
 
@@ -283,7 +283,7 @@ private[cosmos] object SparkBridgeImplementationInternal extends BasicLoggingTra
     val changeFeedState: ChangeFeedState = new ChangeFeedStateV1(
       containerResourceId,
       FeedRangeEpkImpl.forFullRange,
-      ChangeFeedMode.INCREMENTAL,
+      ChangeFeedMode.LATEST_VERSION,
       ChangeFeedStartFromInternal.createFromLegacyContinuation(),
       feedRangeContinuation
     )

@@ -83,8 +83,12 @@ private case class ChangeFeedPartitionReader
       cosmosChangeFeedConfig.changeFeedMode match {
           case ChangeFeedModes.Incremental =>
               factoryMethod = (jsonNode: JsonNode) => changeFeedItemFactoryMethod(jsonNode)
+          case ChangeFeedModes.LatestVersion =>
+            factoryMethod = (jsonNode: JsonNode) => changeFeedItemFactoryMethod(jsonNode)
           case ChangeFeedModes.FullFidelity =>
               factoryMethod = (jsonNode: JsonNode) => changeFeedItemFactoryMethodV1(jsonNode)
+          case ChangeFeedModes.AllVersionsAndDeletes =>
+            factoryMethod = (jsonNode: JsonNode) => changeFeedItemFactoryMethodV1(jsonNode)
     }
 
     ImplementationBridgeHelpers

@@ -121,8 +121,12 @@ private class ChangeFeedTable(val session: SparkSession,
     val defaultSchema: StructType = changeFeedConfig.changeFeedMode match {
       case ChangeFeedModes.Incremental =>
         ChangeFeedTable.defaultIncrementalChangeFeedSchemaForInferenceDisabled
+      case ChangeFeedModes.LatestVersion =>
+        ChangeFeedTable.defaultIncrementalChangeFeedSchemaForInferenceDisabled
       case ChangeFeedModes.FullFidelity =>
         ChangeFeedTable.defaultFullFidelityChangeFeedSchemaForInferenceDisabled
+      case ChangeFeedModes.AllVersionsAndDeletes =>
+        ChangeFeedTable.defaultFullFidelityChangeFeedSchemaForInferenceDisabled
     }
 
     CosmosTableSchemaInferrer.inferSchema(

@@ -1072,6 +1072,8 @@ private object ChangeFeedModes extends Enumeration {
 
   val Incremental: ChangeFeedModes.Value = Value("Incremental")
   val FullFidelity: ChangeFeedModes.Value = Value("FullFidelity")
+  val LatestVersion: ChangeFeedModes.Value = Value("LatestVersion")
+  val AllVersionsAndDeletes: ChangeFeedModes.Value = Value("AllVersionsAndDeletes")
 }
 
 private object ChangeFeedStartFromModes extends Enumeration {
@@ -1103,7 +1105,9 @@ private case class CosmosChangeFeedConfig
 
     this.changeFeedMode match {
       case ChangeFeedModes.Incremental => options
-      case ChangeFeedModes.FullFidelity => options.fullFidelity()
+      case ChangeFeedModes.LatestVersion => options
+      case ChangeFeedModes.FullFidelity => options.allVersionsAndDeletes()
+      case ChangeFeedModes.AllVersionsAndDeletes => options.allVersionsAndDeletes()
     }
   }
 
@@ -1116,7 +1120,7 @@ private case class CosmosChangeFeedConfig
 }
 
 private object CosmosChangeFeedConfig {
-  private val DefaultChangeFeedMode: ChangeFeedMode = ChangeFeedModes.Incremental
+  private val DefaultChangeFeedMode: ChangeFeedMode = ChangeFeedModes.LatestVersion
   private val DefaultStartFromMode: ChangeFeedStartFromMode = ChangeFeedStartFromModes.Beginning
 
   private val startFrom = CosmosConfigEntry[ChangeFeedStartFromMode](
@@ -1140,9 +1144,9 @@ private object CosmosChangeFeedConfig {
   private val changeFeedMode = CosmosConfigEntry[ChangeFeedMode](
     key = CosmosConfigNames.ChangeFeedMode,
     mandatory = false,
-    defaultValue = Some(ChangeFeedModes.Incremental),
+    defaultValue = Some(ChangeFeedModes.LatestVersion),
     parseFromStringFunction = changeFeedModeString => CosmosConfigEntry.parseEnumeration(changeFeedModeString, ChangeFeedModes),
-    helpMessage = "ChangeFeed mode (Incremental or FullFidelity)")
+    helpMessage = "ChangeFeed mode (Incremental/LatestVersion or FullFidelity/AllVersionsAndDeletes)")
 
   private val maxItemCountPerTriggerHint = CosmosConfigEntry[Long](
     key = CosmosConfigNames.ChangeFeedItemCountPerTriggerHint,

@@ -517,7 +517,7 @@ class CosmosConfigSpec extends UnitSpec {
 
     val config = CosmosChangeFeedConfig.parseCosmosChangeFeedConfig(changeFeedConfig)
 
-    config.changeFeedMode shouldEqual ChangeFeedModes.Incremental
+    config.changeFeedMode shouldEqual ChangeFeedModes.LatestVersion
     config.startFrom shouldEqual ChangeFeedStartFromModes.Beginning
     config.startFromPointInTime shouldEqual None
     config.maxItemCountPerTrigger shouldEqual None
@@ -538,7 +538,22 @@ class CosmosConfigSpec extends UnitSpec {
     config.maxItemCountPerTrigger.get shouldEqual 54
   }
 
-  it should "parse change feed config with PIT start mode" in {
+  it should "parse change feed config for all versions and deletes with incorrect casing" in {
+    val changeFeedConfig = Map(
+      "spark.cosmos.changeFeed.mode" -> "AllVersionsANDDELETES",
+      "spark.cosmos.changeFeed.STARTfrom" -> "NOW",
+      "spark.cosmos.changeFeed.itemCountPerTriggerHint" -> "54"
+    )
+
+    val config = CosmosChangeFeedConfig.parseCosmosChangeFeedConfig(changeFeedConfig)
+
+    config.changeFeedMode shouldEqual ChangeFeedModes.AllVersionsAndDeletes
+    config.startFrom shouldEqual ChangeFeedStartFromModes.Now
+    config.startFromPointInTime shouldEqual None
+    config.maxItemCountPerTrigger.get shouldEqual 54
+  }
+
+  it should "parse change feed config (incremental) with PIT start mode" in {
     val changeFeedConfig = Map(
       "spark.cosmos.changeFeed.mode" -> "incremental",
       "spark.cosmos.changeFeed.STARTfrom" -> "2019-12-31T10:45:10Z",
@@ -556,6 +571,24 @@ class CosmosConfigSpec extends UnitSpec {
     config.maxItemCountPerTrigger.get shouldEqual 54
   }
 
+  it should "parse change feed config (latestversion) with PIT start mode" in {
+    val changeFeedConfig = Map(
+      "spark.cosmos.changeFeed.mode" -> "latestversion",
+      "spark.cosmos.changeFeed.STARTfrom" -> "2019-12-31T10:45:10Z",
+      "spark.cosmos.changeFeed.itemCountPerTriggerHint" -> "54"
+    )
+
+    val config = CosmosChangeFeedConfig.parseCosmosChangeFeedConfig(changeFeedConfig)
+
+    config.changeFeedMode shouldEqual ChangeFeedModes.LatestVersion
+    config.startFrom shouldEqual ChangeFeedStartFromModes.PointInTime
+    Instant.from(config.startFromPointInTime.get) shouldEqual
+      new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssXXX")
+        .parse("2019-12-31T10:45:10Z")
+        .toInstant
+    config.maxItemCountPerTrigger.get shouldEqual 54
+  }
+
   it should "complain when parsing invalid change feed mode" in {
     val changeFeedConfig = Map(
       "spark.cosmos.changeFeed.mode" -> "Whatever",
@@ -569,7 +602,7 @@ class CosmosConfigSpec extends UnitSpec {
     } catch {
       case e: Exception => e.getMessage shouldEqual
         "invalid configuration for spark.cosmos.changeFeed.mode:Whatever. Config description: " +
-          "ChangeFeed mode (Incremental or FullFidelity)"
+          "ChangeFeed mode (Incremental/LatestVersion or FullFidelity/AllVersionsAndDeletes)"
     }
   }
 

@@ -576,7 +576,7 @@ class PartitionMetadataSpec extends UnitSpec {
     val json = String.format(
       "{\"V\":1," +
         "\"Rid\":\"%s\"," +
-        "\"Mode\":\"INCREMENTAL\"," +
+        "\"Mode\":\"LATEST_VERSION\"," +
         "\"StartFrom\":{\"Type\":\"BEGINNING\"}," +
         "\"Continuation\":%s}",
       collectionRid,

@@ -82,6 +82,52 @@ class SparkE2EChangeFeedITest
       ChangeFeedTable.defaultIncrementalChangeFeedSchemaForInferenceDisabled) shouldEqual true
   }
 
+  "spark change feed query (LatestVersion)" can "use default schema" in {
+    val cosmosEndpoint = TestConfigurations.HOST
+    val cosmosMasterKey = TestConfigurations.MASTER_KEY
+
+    val container = cosmosClient.getDatabase(cosmosDatabase).getContainer(cosmosContainer)
+    for (state <- Array(true, false)) {
+      val objectNode = Utils.getSimpleObjectMapper.createObjectNode()
+      objectNode.put("name", "Shrodigner's cat")
+      objectNode.put("type", "cat")
+      objectNode.put("age", 20)
+      objectNode.put("isAlive", state)
+      objectNode.put("id", UUID.randomUUID().toString)
+      container.createItem(objectNode).block()
+    }
+    val cfg = Map(
+      "spark.cosmos.accountEndpoint" -> cosmosEndpoint,
+      "spark.cosmos.accountKey" -> cosmosMasterKey,
+      "spark.cosmos.database" -> cosmosDatabase,
+      "spark.cosmos.container" -> cosmosContainer,
+      "spark.cosmos.read.maxItemCount" -> "2",
+      "spark.cosmos.read.inferSchema.enabled" -> "false"
+    )
+
+    val df = spark.read.format("cosmos.oltp.changeFeed").options(cfg).load()
+    val rowsArray = df.collect()
+    rowsArray should have size 2
+    df.schema.equals(
+      ChangeFeedTable.defaultIncrementalChangeFeedSchemaForInferenceDisabled) shouldEqual true
+
+    val cfgExplicit = Map(
+      "spark.cosmos.accountEndpoint" -> cosmosEndpoint,
+      "spark.cosmos.accountKey" -> cosmosMasterKey,
+      "spark.cosmos.database" -> cosmosDatabase,
+      "spark.cosmos.container" -> cosmosContainer,
+      "spark.cosmos.read.inferSchema.enabled" -> "false",
+      "spark.cosmos.read.maxItemCount" -> "1",
+      "spark.cosmos.changeFeed.mode" -> "LatestVersion"
+    )
+
+    val dfExplicit = spark.read.format("cosmos.oltp.changeFeed").options(cfgExplicit).load()
+    val rowsArrayExplicit = dfExplicit.collect()
+    rowsArrayExplicit should have size 2
+    dfExplicit.schema.equals(
+      ChangeFeedTable.defaultIncrementalChangeFeedSchemaForInferenceDisabled) shouldEqual true
+  }
+
   "spark change feed query (incremental)" can "use user provided schema" in {
     val cosmosEndpoint = TestConfigurations.HOST
     val cosmosMasterKey = TestConfigurations.MASTER_KEY
@@ -151,6 +197,38 @@ class SparkE2EChangeFeedITest
       ChangeFeedTable.defaultFullFidelityChangeFeedSchemaForInferenceDisabled) shouldEqual true
   }
 
+  "spark change feed query (all versions and deletes)" can "use default schema" in {
+    val cosmosEndpoint = TestConfigurations.HOST
+    val cosmosMasterKey = TestConfigurations.MASTER_KEY
+
+    val container = cosmosClient.getDatabase(cosmosDatabase).getContainer(cosmosContainer)
+    for (state <- Array(true, false)) {
+      val objectNode = Utils.getSimpleObjectMapper.createObjectNode()
+      objectNode.put("name", "Shrodigner's cat")
+      objectNode.put("type", "cat")
+      objectNode.put("age", 20)
+      objectNode.put("isAlive", state)
+      objectNode.put("id", UUID.randomUUID().toString)
+      container.createItem(objectNode).block()
+    }
+    val cfg = Map(
+      "spark.cosmos.accountEndpoint" -> cosmosEndpoint,
+      "spark.cosmos.accountKey" -> cosmosMasterKey,
+      "spark.cosmos.database" -> cosmosDatabase,
+      "spark.cosmos.container" -> cosmosContainer,
+      "spark.cosmos.read.inferSchema.enabled" -> "false",
+      "spark.cosmos.changeFeed.mode" -> "AllVersionsAndDeletes",
+      "spark.cosmos.read.maxItemCount" -> "1",
+      "spark.cosmos.changeFeed.startFrom" -> "NOW"
+    )
+
+    val df = spark.read.format("cosmos.oltp.changeFeed").options(cfg).load()
+    val rowsArray = df.collect()
+    rowsArray should have size 0
+    df.schema.equals(
+      ChangeFeedTable.defaultFullFidelityChangeFeedSchemaForInferenceDisabled) shouldEqual true
+  }
+
   "spark change feed micro batch (incremental)" can "use default schema" in {
     val cosmosEndpoint = TestConfigurations.HOST
     val cosmosMasterKey = TestConfigurations.MASTER_KEY

@@ -211,7 +211,7 @@ trait CosmosContainerWithRetention extends CosmosContainer {
     val properties: CosmosContainerProperties =
       new CosmosContainerProperties(cosmosContainer, partitionKeyPath)
     properties.setChangeFeedPolicy(
-      ChangeFeedPolicy.createFullFidelityPolicy(Duration.ofMinutes(10)))
+      ChangeFeedPolicy.createAllVersionsAndDeletesPolicy(Duration.ofMinutes(10)))
 
     val throughputProperties = ThroughputProperties.createManualThroughput(Defaults.DefaultContainerThroughput)
 

@@ -3,6 +3,7 @@
 ### 4.37.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added new preview APIs to `ChangeFeedProcessor` for handling full fidelity / all changes - See [PR 30399](https://github.com/Azure/azure-sdk-for-java/pull/30399)
 
 #### Breaking Changes