Azure · tvaron3 · Jul 1, 2026 · Jul 1, 2026 · Jul 2, 2026 · Jul 2, 2026
@@ -3,6 +3,7 @@
 ### 4.50.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added a new config option `spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures` to allow ignoring `412 Precondition Failed` errors when using the `ItemPatch`/`ItemPatchIfExists` write strategy together with a conditional `spark.cosmos.write.patch.filter`, so documents excluded by the filter are skipped instead of failing the write. - See [PR 49700](https://github.com/Azure/azure-sdk-for-java/pull/49700)
 
 #### Breaking Changes
 

@@ -3,6 +3,7 @@
 ### 4.50.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added a new config option `spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures` to allow ignoring `412 Precondition Failed` errors when using the `ItemPatch`/`ItemPatchIfExists` write strategy together with a conditional `spark.cosmos.write.patch.filter`, so documents excluded by the filter are skipped instead of failing the write. - See [PR 49700](https://github.com/Azure/azure-sdk-for-java/pull/49700)
 
 #### Breaking Changes
 

@@ -3,6 +3,7 @@
 ### 4.50.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added a new config option `spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures` to allow ignoring `412 Precondition Failed` errors when using the `ItemPatch`/`ItemPatchIfExists` write strategy together with a conditional `spark.cosmos.write.patch.filter`, so documents excluded by the filter are skipped instead of failing the write. - See [PR 49700](https://github.com/Azure/azure-sdk-for-java/pull/49700)
 
 #### Breaking Changes
 

@@ -3,6 +3,7 @@
 ### 4.50.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added a new config option `spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures` to allow ignoring `412 Precondition Failed` errors when using the `ItemPatch`/`ItemPatchIfExists` write strategy together with a conditional `spark.cosmos.write.patch.filter`, so documents excluded by the filter are skipped instead of failing the write. - See [PR 49700](https://github.com/Azure/azure-sdk-for-java/pull/49700)
 
 #### Breaking Changes
 

@@ -83,6 +83,7 @@
 | `spark.cosmos.write.patch.defaultOperationType` | `Replace` | Default Cosmos DB patch operation type. Supported ones include none, add, set, replace, remove, increment. Choose none for no-op, for others please reference [here](https://docs.microsoft.com/azure/cosmos-db/partial-document-update#supported-operations) for full context.                                       |
 | `spark.cosmos.write.patch.columnConfigs`        | None      | Cosmos DB patch column configs. It can container multiple definitions matching the following patterns separated by comma. col(column).op(operationType) or col(column).path(patchInCosmosdb).op(operationType) - The difference of the second pattern is that it also allows you to define a different cosmosdb path. |
 | `spark.cosmos.write.patch.filter`               | None      | Used for [Conditional patch](https://docs.microsoft.com/azure/cosmos-db/partial-document-update-getting-started#java)                                                                                                                                                                                                 |
+| `spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures` | `false`   | Only applicable to the `ItemPatch`/`ItemPatchIfExists` write strategy when used together with `spark.cosmos.write.patch.filter`. When set to `true`, documents that do not match the conditional patch filter (which the service rejects with an HTTP 412 Precondition Failed) are silently skipped and treated as a successful no-op instead of failing the write. When `false` (default) a 412 fails the write, preserving the existing behavior. |
 
 ### Query Config
 | Config Property Name                                         | Default | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |

@@ -1295,9 +1295,16 @@ private class BulkWriter
   }
 
   private def shouldIgnore(statusCode: Int, subStatusCode: Int): Boolean = {
+    val ignorePatchPreconditionFailures =
+      writeConfig.patchConfigs.exists(patchConfigs =>
+        patchConfigs.filterPredicateIgnorePreconditionFailures && patchConfigs.filter.exists(_.nonEmpty))
     val returnValue = writeConfig.itemWriteStrategy match {
       case ItemWriteStrategy.ItemAppend => Exceptions.isResourceExistsException(statusCode)
-      case ItemWriteStrategy.ItemPatchIfExists => Exceptions.isNotFoundExceptionCore(statusCode, subStatusCode)
+      case ItemWriteStrategy.ItemPatch =>
+        ignorePatchPreconditionFailures && Exceptions.isPreconditionFailedException(statusCode)
+      case ItemWriteStrategy.ItemPatchIfExists =>
+        Exceptions.isNotFoundExceptionCore(statusCode, subStatusCode) ||
+          (ignorePatchPreconditionFailures && Exceptions.isPreconditionFailedException(statusCode))
       case ItemWriteStrategy.ItemDelete => Exceptions.isNotFoundExceptionCore(statusCode, subStatusCode)
       case ItemWriteStrategy.ItemDeleteIfNotModified => Exceptions.isNotFoundExceptionCore(statusCode, subStatusCode) ||
         Exceptions.isPreconditionFailedException(statusCode)

@@ -131,6 +131,7 @@ private[spark] object CosmosConfigNames {
   val WritePatchDefaultOperationType = "spark.cosmos.write.patch.defaultOperationType"
   val WritePatchColumnConfigs = "spark.cosmos.write.patch.columnConfigs"
   val WritePatchFilterPredicate = "spark.cosmos.write.patch.filter"
+  val WritePatchFilterPredicateIgnorePreconditionFailures = "spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures"
   val WriteBulkUpdateColumnConfigs = "spark.cosmos.write.bulkUpdate.columnConfigs"
   val WriteStrategy = "spark.cosmos.write.strategy"
   val WriteMaxRetryCount = "spark.cosmos.write.maxRetryCount"
@@ -272,6 +273,7 @@ private[spark] object CosmosConfigNames {
     WritePatchDefaultOperationType,
     WritePatchColumnConfigs,
     WritePatchFilterPredicate,
+    WritePatchFilterPredicateIgnorePreconditionFailures,
     WriteBulkUpdateColumnConfigs,
     WriteStrategy,
     WriteMaxRetryCount,
@@ -1581,7 +1583,8 @@ private case class CosmosPatchColumnConfig(columnName: String,
                                            isRawJson: Boolean)
 
 private case class CosmosPatchConfigs(columnConfigsMap: TrieMap[String, CosmosPatchColumnConfig],
-                                      filter: Option[String] = None)
+                                      filter: Option[String] = None,
+                                      filterPredicateIgnorePreconditionFailures: Boolean = false)
 
 private case class CosmosWriteConfig(itemWriteStrategy: ItemWriteStrategy,
                                      maxRetryCount: Int,
@@ -1755,6 +1758,17 @@ private object CosmosWriteConfig {
     helpMessage = "Used for conditional patch. Please see examples here: " +
      "https://docs.microsoft.com/en-us/azure/cosmos-db/partial-document-update-getting-started#java")
 
+  private val patchFilterPredicateIgnorePreconditionFailures = CosmosConfigEntry[Boolean](
+    key = CosmosConfigNames.WritePatchFilterPredicateIgnorePreconditionFailures,
+    mandatory = false,
+    defaultValue = Option.apply(false),
+    parseFromStringFunction = ignoreString => ignoreString.toBoolean,
+    helpMessage = "Flag to indicate whether precondition failures (HTTP 412) should be ignored when using " +
+     "the ItemPatch/ItemPatchIfExists write strategy together with a patch filter predicate " +
+     s"('${CosmosConfigNames.WritePatchFilterPredicate}'). When enabled, documents skipped by the filter " +
+     "predicate (which the service rejects with a 412) are treated as a successful no-op instead of failing " +
+     "the write. Defaults to false.")
+
   private val patchBulkUpdateColumnConfigs = CosmosConfigEntry[TrieMap[String, CosmosPatchColumnConfig]](key = CosmosConfigNames.WriteBulkUpdateColumnConfigs,
       mandatory = false,
       parseFromStringFunction = columnConfigsString => parsePatchBulkUpdateColumnConfigs(columnConfigsString),
@@ -1933,7 +1947,9 @@ private object CosmosWriteConfig {
       case ItemWriteStrategy.ItemPatch | ItemWriteStrategy.ItemPatchIfExists =>
         val patchColumnConfigMap = parsePatchColumnConfigs(cfg, inputSchema)
         val patchFilter = CosmosConfigEntry.parse(cfg, patchFilterPredicate)
-        patchConfigsOpt = Some(CosmosPatchConfigs(patchColumnConfigMap, patchFilter))
+        val ignorePreconditionFailures =
+          CosmosConfigEntry.parse(cfg, patchFilterPredicateIgnorePreconditionFailures).getOrElse(false)
+        patchConfigsOpt = Some(CosmosPatchConfigs(patchColumnConfigMap, patchFilter, ignorePreconditionFailures))
       case ItemWriteStrategy.ItemBulkUpdate =>
         val patchColumnConfigMapOpt = CosmosConfigEntry.parse(cfg, patchBulkUpdateColumnConfigs)
         patchConfigsOpt = Some(CosmosPatchConfigs(patchColumnConfigMapOpt.getOrElse(new TrieMap[String, CosmosPatchColumnConfig])))

@@ -375,6 +375,17 @@ private class PointWriter(container: CosmosAsyncContainer,
               case None => None
             })
           return
+        case e: CosmosException if cosmosWriteConfig.patchConfigs.exists(patchConfigs =>
+          patchConfigs.filterPredicateIgnorePreconditionFailures && patchConfigs.filter.exists(_.nonEmpty)) &&
+          Exceptions.isPreconditionFailedException(e.getStatusCode) =>
+          log.logItemWriteSkipped(patchOperation, "preConditionNotMet")
+          outputMetricsPublisher.trackWriteOperation(
+            0,
+            Option.apply(e.getDiagnostics) match {
+              case Some(diagnostics) => Option.apply(diagnostics.getDiagnosticsContext)
+              case None => None
+            })
+          return
         case e: CosmosException if Exceptions.canBeTransientFailure(e.getStatusCode, e.getSubStatusCode) =>
           log.logWarning(
             s"patch item $patchOperation attempt #$attempt max remaining retries "

@@ -1336,6 +1336,73 @@ class BulkWriterITest extends IntegrationSpec with CosmosClient with AutoCleanab
     objectMapper.writeValueAsString(updatedItem) shouldEqual  objectMapper.writeValueAsString(originalItem)
   }
 
+  "Bulk Writer" can "skip 412 for patch item with condition when filterPredicateIgnorePreconditionFailures is enabled" in {
+    val container = getContainer
+    val containerProperties = container.read().block().getProperties
+    val partitionKeyDefinition = containerProperties.getPartitionKeyDefinition
+    val strippedPartitionKeyPath = CosmosPatchTestHelper.getStrippedPartitionKeyPath(partitionKeyDefinition)
+    val containerConfig = CosmosContainerConfig(container.getDatabase.getId, container.getId, None)
+    val writeConfig = CosmosWriteConfig(
+      ItemWriteStrategy.ItemOverwrite,
+      5,
+      bulkEnabled = true,
+      bulkTransactional = false,
+      bulkExecutionConfigs = Some(CosmosWriteBulkExecutionConfigs()),
+      bulkMaxPendingOperations = Some(900)
+    )
+
+    val bulkWriter = new BulkWriter(
+      container,
+      containerConfig,
+      partitionKeyDefinition,
+      writeConfig,
+      DiagnosticsConfig(),
+      new TestOutputMetricsPublisher)
+
+    // First create one item, as patch can only operate on existing items
+    val itemWithFullSchema = CosmosPatchTestHelper.getPatchItemWithFullSchema(UUID.randomUUID().toString, strippedPartitionKeyPath)
+    val id = itemWithFullSchema.get("id").textValue()
+    val partitionKey = new PartitionKey(itemWithFullSchema.get(strippedPartitionKeyPath).textValue())
+
+    bulkWriter.scheduleWrite(partitionKey, itemWithFullSchema)
+    bulkWriter.flushAndClose()
+    val originalItem = container.readItem(id, partitionKey, classOf[ObjectNode]).block().getItem
+
+    val partialUpdateSchema = StructType(Seq(
+      StructField("propInt", IntegerType)
+    ))
+
+    val patchPartialUpdateItem =
+      CosmosPatchTestHelper.getPatchItemWithSchema(
+        strippedPartitionKeyPath,
+        partialUpdateSchema,
+        originalItem)
+
+    val columnConfigsMap = new TrieMap[String, CosmosPatchColumnConfig]
+    patchPartialUpdateItem.fields().asScala.foreach(field => {
+      columnConfigsMap += field.getKey -> CosmosPatchColumnConfig(
+        field.getKey, CosmosPatchOperationTypes.Set, s"/${field.getKey}", isRawJson = false)
+    })
+
+    val bulkWriterForPatch =
+      CosmosPatchTestHelper.getBulkWriterForPatch(
+        columnConfigsMap,
+        container,
+        containerConfig,
+        partitionKeyDefinition,
+        Some(s"from c where c.propInt > ${Integer.MAX_VALUE}"), // using an always false condition
+        filterPredicateIgnorePreconditionFailures = true)
+
+    // with the flag enabled, the 412 raised by the always-false filter should be skipped, not thrown
+    bulkWriterForPatch.scheduleWrite(partitionKey, patchPartialUpdateItem)
+    bulkWriterForPatch.flushAndClose()
+
+    val updatedItem: ObjectNode = container.readItem(id, partitionKey, classOf[ObjectNode]).block().getItem
+
+    // since the condition is always false, the item should not be updated even though no exception was thrown
+    objectMapper.writeValueAsString(updatedItem) shouldEqual objectMapper.writeValueAsString(originalItem)
+  }
+
   "Bulk Writer" should "throw exception if no valid operations are included in patch operation" in {
     val container = getContainer
     val containerConfig = CosmosContainerConfig(container.getDatabase.getId, container.getId, None)

@@ -1349,6 +1349,34 @@ class CosmosConfigSpec extends UnitSpec with BasicLoggingTrait {
 
   }
 
+  "Config Parser" should "parse patch filterPredicateIgnorePreconditionFailures config" in {
+    val schema = CosmosPatchTestHelper.getPatchConfigTestSchema()
+
+    // defaults to false when not specified
+    val defaultConfig = Map(
+      "spark.cosmos.write.strategy" -> "ItemPatch",
+      "spark.cosmos.write.patch.filter" -> "where c.booleanTypeColumn = true"
+    )
+    val defaultWriteConfig = CosmosWriteConfig.parseWriteConfig(defaultConfig, schema)
+    defaultWriteConfig.patchConfigs.isDefined shouldEqual true
+    defaultWriteConfig.patchConfigs.get.filterPredicateIgnorePreconditionFailures shouldEqual false
+
+    // parses true
+    val enabledConfig = defaultConfig +
+      ("spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures" -> "true")
+    val enabledWriteConfig = CosmosWriteConfig.parseWriteConfig(enabledConfig, schema)
+    enabledWriteConfig.patchConfigs.get.filterPredicateIgnorePreconditionFailures shouldEqual true
+
+    // parses false explicitly, also for ItemPatchIfExists
+    val disabledConfig = Map(
+      "spark.cosmos.write.strategy" -> "ItemPatchIfExists",
+      "spark.cosmos.write.patch.filter" -> "where c.booleanTypeColumn = true",
+      "spark.cosmos.write.patch.filterPredicateIgnorePreconditionFailures" -> "false"
+    )
+    val disabledWriteConfig = CosmosWriteConfig.parseWriteConfig(disabledConfig, schema)
+    disabledWriteConfig.patchConfigs.get.filterPredicateIgnorePreconditionFailures shouldEqual false
+  }
+
   "Config Parser" should "validate column configs for column does not exists in schema for patch configs" in {
     val schema = CosmosPatchTestHelper.getPatchConfigTestSchema()
 

@@ -948,6 +948,67 @@ class PointWriterITest extends IntegrationSpec with CosmosClient with AutoCleana
     objectMapper.writeValueAsString(updatedItem) shouldEqual objectMapper.writeValueAsString(originalItem)
   }
 
+  "Point Writer" can "skip 412 for patch item with condition when filterPredicateIgnorePreconditionFailures is enabled" in {
+    val container = getContainer
+    val containerProperties = container.read().block().getProperties
+    val partitionKeyDefinition = containerProperties.getPartitionKeyDefinition
+    val strippedPartitionKeyPath = CosmosPatchTestHelper.getStrippedPartitionKeyPath(partitionKeyDefinition)
+    val writeConfig = CosmosWriteConfig(
+      ItemWriteStrategy.ItemOverwrite,
+      5,
+      bulkEnabled = false,
+      bulkTransactional = false)
+
+    val pointWriter = new PointWriter(
+      container,
+      partitionKeyDefinition,
+      writeConfig,
+      DiagnosticsConfig(),
+      MockTaskContext.mockTaskContext(),
+      new TestOutputMetricsPublisher)
+
+    // First create one item, as patch can only operate on existing items
+    val itemWithFullSchema = CosmosPatchTestHelper.getPatchItemWithFullSchema(UUID.randomUUID().toString, strippedPartitionKeyPath)
+    val id = itemWithFullSchema.get("id").textValue()
+    val partitionKey = new PartitionKey(itemWithFullSchema.get(strippedPartitionKeyPath).textValue())
+
+    pointWriter.scheduleWrite(partitionKey, itemWithFullSchema)
+    pointWriter.flushAndClose()
+    val originalItem = container.readItem(id, partitionKey, classOf[ObjectNode]).block().getItem
+
+    val partialUpdateSchema = StructType(Seq(
+      StructField("propInt", IntegerType)
+    ))
+
+    val patchPartialUpdateItem =
+      CosmosPatchTestHelper.getPatchItemWithSchema(
+        strippedPartitionKeyPath,
+        partialUpdateSchema,
+        originalItem)
+
+    val columnConfigsMap = new TrieMap[String, CosmosPatchColumnConfig]
+    patchPartialUpdateItem.fields().asScala.foreach(field => {
+      columnConfigsMap += field.getKey -> CosmosPatchColumnConfig(
+        field.getKey, CosmosPatchOperationTypes.Set, s"/${field.getKey}", isRawJson = false)
+    })
+
+    val pointWriterForPatch =
+      CosmosPatchTestHelper.getPointWriterForPatch(
+        columnConfigsMap,
+        container,
+        partitionKeyDefinition,
+        Some(s"from c where c.propInt > ${Integer.MAX_VALUE}"), // using an always false condition
+        filterPredicateIgnorePreconditionFailures = true)
+
+    // with the flag enabled, the 412 raised by the always-false filter should be skipped, not thrown
+    pointWriterForPatch.scheduleWrite(partitionKey, patchPartialUpdateItem)
+    pointWriterForPatch.flushAndClose()
+
+    // since the condition is always false, the item should not be updated even though no exception was thrown
+    val updatedItem: ObjectNode = container.readItem(id, partitionKey, classOf[ObjectNode]).block().getItem
+    objectMapper.writeValueAsString(updatedItem) shouldEqual objectMapper.writeValueAsString(originalItem)
+  }
+
   "Point Writer" should "throw exception if no valid operations are included in patch operation" in {
     val container = getContainer
     val containerProperties = container.read().block().getProperties