diff --git a/sdk/cosmos/azure-cosmos-spark_3-3_2-12/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_3-3_2-12/CHANGELOG.md index cdb056260233..064b61558efc 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-3_2-12/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_3-3_2-12/CHANGELOG.md @@ -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 diff --git a/sdk/cosmos/azure-cosmos-spark_3-4_2-12/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_3-4_2-12/CHANGELOG.md index dfa6c427152f..ec7c6b392242 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-4_2-12/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_3-4_2-12/CHANGELOG.md @@ -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 diff --git a/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md index 95ca876a520a..e82336246fbb 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md @@ -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 diff --git a/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md index 4d902f73dea0..174e0ae4eae6 100644 --- a/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_3-5_2-13/CHANGELOG.md @@ -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 diff --git a/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md b/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md index 166d0a0c9f75..9db5619aca3e 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md +++ b/sdk/cosmos/azure-cosmos-spark_3/docs/configuration-reference.md @@ -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 | diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala index a674e6d5a256..f3f67f021d48 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala @@ -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) diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala index 3e6461b94c89..2fc2867db449 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala @@ -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]))) diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/PointWriter.scala b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/PointWriter.scala index 8f07bf5339d5..31bc20240652 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/PointWriter.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/PointWriter.scala @@ -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 " diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/BulkWriterITest.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/BulkWriterITest.scala index 43538a033c36..f04de52dab7b 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/BulkWriterITest.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/BulkWriterITest.scala @@ -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) diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala index f9281d106362..1b842b4c1b5c 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/CosmosConfigSpec.scala @@ -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() diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/PointWriterITest.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/PointWriterITest.scala index fcbc2a926ec5..eaad3770a28a 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/PointWriterITest.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/PointWriterITest.scala @@ -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 diff --git a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/utils/CosmosPatchTestHelper.scala b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/utils/CosmosPatchTestHelper.scala index c6876bef3b1c..1661a60949c1 100644 --- a/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/utils/CosmosPatchTestHelper.scala +++ b/sdk/cosmos/azure-cosmos-spark_3/src/test/scala/com/azure/cosmos/spark/utils/CosmosPatchTestHelper.scala @@ -166,8 +166,9 @@ def getPatchFullTestSchemaWithSubpartitions(): StructType = { containerConfig: CosmosContainerConfig, partitionKeyDefinition: PartitionKeyDefinition, patchPredicateFilter: Option[String] = None, - metricsPublisher: OutputMetricsPublisherTrait = new TestOutputMetricsPublisher): BulkWriter = { - val patchConfigs = CosmosPatchConfigs(columnConfigsMap, patchPredicateFilter) + metricsPublisher: OutputMetricsPublisherTrait = new TestOutputMetricsPublisher, + filterPredicateIgnorePreconditionFailures: Boolean = false): BulkWriter = { + val patchConfigs = CosmosPatchConfigs(columnConfigsMap, patchPredicateFilter, filterPredicateIgnorePreconditionFailures) val writeConfigForPatch = CosmosWriteConfig( ItemWriteStrategy.ItemPatch, 5, @@ -215,9 +216,10 @@ def getPatchFullTestSchemaWithSubpartitions(): StructType = { container: CosmosAsyncContainer, partitionKeyDefinition: PartitionKeyDefinition, patchPredicateFilter: Option[String] = None, - metricsPublisher: OutputMetricsPublisherTrait = new TestOutputMetricsPublisher): PointWriter = { + metricsPublisher: OutputMetricsPublisherTrait = new TestOutputMetricsPublisher, + filterPredicateIgnorePreconditionFailures: Boolean = false): PointWriter = { - val patchConfigs = CosmosPatchConfigs(columnConfigsMap, patchPredicateFilter) + val patchConfigs = CosmosPatchConfigs(columnConfigsMap, patchPredicateFilter, filterPredicateIgnorePreconditionFailures) val writeConfigForPatch = CosmosWriteConfig( ItemWriteStrategy.ItemPatch, 5, diff --git a/sdk/cosmos/azure-cosmos-spark_4-0_2-13/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_4-0_2-13/CHANGELOG.md index 309ffc511d7c..e485e9885c96 100644 --- a/sdk/cosmos/azure-cosmos-spark_4-0_2-13/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_4-0_2-13/CHANGELOG.md @@ -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 diff --git a/sdk/cosmos/azure-cosmos-spark_4-1_2-13/CHANGELOG.md b/sdk/cosmos/azure-cosmos-spark_4-1_2-13/CHANGELOG.md index cdd208435172..ecfa3506e7ef 100644 --- a/sdk/cosmos/azure-cosmos-spark_4-1_2-13/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-spark_4-1_2-13/CHANGELOG.md @@ -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