feat: add bulkPatch support for CosmosDB/PostgreSQL/Mongo with per-item result classification (#210)

* feat: add bulk patch methods

* feat: add bulk patch methods. add comments and tests

---------

Co-authored-by: zhang.lei <zhang.lei@onehr.jp>

Author: thunderz99

README.md

@@ -240,6 +240,33 @@ The main difference between Partial update and Patch is that:
 * Partial update is focused at insert/replace fields' values. And is able to support updating 10 more fields in one call.
 * Patch is focused to implementing a method similar to JSON Patch, which supports more complicated ops like "Add, Set, Replace, Remove, Increment". And is not able to support ops exceeding 10 in one patch call.
 
+### Bulk Patch
+
+Bulk patch is a non-transactional bulk operation for patch updates.
+
+```java
+// 1) Apply the same patch operations to multiple ids
+var ids = List.of("id001", "id002", "id003");
+var operations = PatchOperations.create()
+        .set("/status", "ENABLED")
+        .increment("/version", 1);
+
+var result1 = db.bulkPatch("Collection1", ids, operations, "Users");
+
+// 2) Apply different patch operations per id
+var patchList = List.of(
+        BulkPatchOperation.of("id001", PatchOperations.create().set("/status", "ENABLED")),
+        BulkPatchOperation.of("id002", PatchOperations.create().set("/status", "DISABLED"))
+);
+
+var result2 = db.bulkPatch("Collection1", patchList, "Users");
+```
+
+Return value is `CosmosBulkResult`:
+
+* `successList`: patched documents (or document ids depending on implementation)
+* `fatalList`: failed items with `CosmosException` (for example, target id not found)
+
 ### Complex queries
 
 ```java

src/main/java/io/github/thunderz99/cosmos/CosmosDatabase.java

@@ -4,6 +4,7 @@
 
 import io.github.thunderz99.cosmos.condition.Aggregate;
 import io.github.thunderz99.cosmos.condition.Condition;
+import io.github.thunderz99.cosmos.dto.BulkPatchOperation;
 import io.github.thunderz99.cosmos.dto.CosmosBulkResult;
 import io.github.thunderz99.cosmos.dto.PartialUpdateOption;
 import io.github.thunderz99.cosmos.v4.PatchOperations;
@@ -512,6 +513,67 @@ default public CosmosDocumentList aggregate(String coll, Aggregate aggregate, Co
      */
     public CosmosBulkResult bulkDelete(String coll, List<?> data, String partition) throws Exception;
 
+    /**
+     * Bulk patch documents with the same patch operations.
+     *
+     * <p>
+     * Default implementation converts ids to BulkPatchOperation list and delegates to
+     * {@link #bulkPatch(String, List, String)}.
+     * </p>
+     *
+     * @param coll       collection name
+     * @param ids        target document ids
+     * @param operations patch operations applied to each target id
+     * @param partition  partition name
+     * @return CosmosBulkResult
+     * @throws Exception cosmos exception
+     */
+    default CosmosBulkResult bulkPatch(String coll, List<String> ids, PatchOperations operations, String partition) throws Exception {
+        if (ids == null) {
+            return this.bulkPatch(coll, (List<BulkPatchOperation>) null, partition);
+        }
+
+        var data = ids.stream()
+                .map(id -> BulkPatchOperation.of(id, operations == null ? null : operations.copy()))
+                .toList();
+        return this.bulkPatch(coll, data, partition);
+    }
+
+    /**
+     * Bulk patch documents.
+     * Note: Non-transaction. Have no number limit in theoretically.
+     *
+     * <p>
+     * Default implementation executes one-by-one patch for compatibility.
+     * </p>
+     *
+     * @param coll      collection name
+     * @param data      bulk patch operations (id + PatchOperations)
+     * @param partition partition name
+     * @return CosmosBulkResult
+     */
+    default CosmosBulkResult bulkPatch(String coll, List<BulkPatchOperation> data, String partition) throws Exception {
+        var ret = new CosmosBulkResult();
+        if (data == null) {
+            return ret;
+        }
+
+        for (var operation : data) {
+            try {
+                if (operation == null) {
+                    throw new CosmosException(400, "BAD_REQUEST", "bulkPatch operation should not be null");
+                }
+                ret.successList.add(this.patch(coll, operation.id, operation.operations, partition));
+            } catch (CosmosException ce) {
+                ret.fatalList.add(ce);
+            } catch (Exception e) {
+                ret.fatalList.add(new CosmosException(500, "UNKNOWN", e.getMessage(), e));
+            }
+        }
+
+        return ret;
+    }
+
     /**
      * Ping a collection to test whether it is accessible.
      *

src/main/java/io/github/thunderz99/cosmos/dto/BulkPatchOperation.java

@@ -0,0 +1,47 @@
+package io.github.thunderz99.cosmos.dto;
+
+import io.github.thunderz99.cosmos.v4.PatchOperations;
+
+/**
+ * A patch task used by bulkPatch.
+ */
+public class BulkPatchOperation {
+
+    /**
+     * Target document id.
+     */
+    public String id;
+
+    /**
+     * Patch operations for the target document.
+     */
+    public PatchOperations operations;
+
+    /**
+     * Create an empty operation.
+     */
+    public BulkPatchOperation() {
+    }
+
+    /**
+     * Create an operation with id and patch operations.
+     *
+     * @param id         target document id
+     * @param operations patch operations
+     */
+    public BulkPatchOperation(String id, PatchOperations operations) {
+        this.id = id;
+        this.operations = operations;
+    }
+
+    /**
+     * Build a bulk patch operation.
+     *
+     * @param id         target document id
+     * @param operations patch operations
+     * @return BulkPatchOperation
+     */
+    public static BulkPatchOperation of(String id, PatchOperations operations) {
+        return new BulkPatchOperation(id, operations);
+    }
+}

src/main/java/io/github/thunderz99/cosmos/impl/cosmosdb/CosmosDatabaseImpl.java

@@ -10,6 +10,7 @@
 import io.github.thunderz99.cosmos.*;
 import io.github.thunderz99.cosmos.condition.Aggregate;
 import io.github.thunderz99.cosmos.condition.Condition;
+import io.github.thunderz99.cosmos.dto.BulkPatchOperation;
 import io.github.thunderz99.cosmos.dto.CosmosBatchResponseWrapper;
 import io.github.thunderz99.cosmos.dto.CosmosBulkResult;
 import io.github.thunderz99.cosmos.dto.CosmosSqlQuerySpec;
@@ -36,7 +37,8 @@ public class CosmosDatabaseImpl implements CosmosDatabase {
 
     private static Logger log = LoggerFactory.getLogger(CosmosDatabaseImpl.class);
 
-    static final int MAX_BATCH_NUMBER_OF_OPERATION = 100;
+    static final int MAX_BATCH_NUMBER_OF_OPERATION = CosmosLimits.BATCH_OPERATION_LIMIT;
+    static final int BULK_PATCH_CHUNK_SIZE = CosmosLimits.BULK_CHUNK_SIZE;
 
     static final int FIND_PREFERRED_PAGE_SIZE = 10;
 
@@ -1279,6 +1281,85 @@ public CosmosBulkResult bulkDelete(String coll, List<?> data, String partition)
         return result;
     }
 
+    /**
+     * Bulk patch documents with the same patch operations.
+     *
+     * @param coll       collection name
+     * @param ids        target document ids
+     * @param operations patch operations
+     * @param partition  partition name
+     * @return CosmosBulkResult
+     * @throws Exception cosmos exception
+     */
+    public CosmosBulkResult bulkPatch(String coll, List<String> ids, PatchOperations operations, String partition) throws Exception {
+        doCheckBeforeBulkPatch(coll, ids, operations, partition);
+
+        var partitionKey = new PartitionKey(partition);
+        var container = this.clientV4.getDatabase(db).getContainer(coll);
+
+        log.info("begin bulkPatch(same operations) coll:{}, partition:{}, account:{}", coll, partition, getAccount());
+
+        var ret = new CosmosBulkResult();
+        var retryList = new ArrayList<Object>();
+        for (int from = 0; from < ids.size(); from += BULK_PATCH_CHUNK_SIZE) {
+            var to = Math.min(from + BULK_PATCH_CHUNK_SIZE, ids.size());
+            var chunkIds = ids.subList(from, to);
+            var itemOperations = chunkIds.stream()
+                    .map(id -> CosmosBulkOperations.getPatchItemOperation(id, partitionKey, operations.getCosmosPatchOperations()))
+                    .collect(Collectors.toList());
+
+            var chunkResult = RetryUtil.executeBulkWithRetry(coll, itemOperations,
+                    (ops) -> container.executeBulkOperations(ops));
+
+            ret.successList.addAll(chunkResult.successList);
+            ret.fatalList.addAll(chunkResult.fatalList);
+            retryList.addAll(chunkResult.retryList);
+        }
+        ret.retryList = retryList;
+
+        log.info("end bulkPatch(same operations) coll:{}, partition:{}, account:{}", coll, partition, getAccount());
+        return ret;
+    }
+
+    /**
+     * Bulk patch documents.
+     * Note: Non-transaction. Have no number limit in theoretically.
+     *
+     * @param coll      collection name
+     * @param data      bulk patch operations (id + PatchOperations)
+     * @param partition partition name
+     * @return CosmosBulkResult
+     */
+    public CosmosBulkResult bulkPatch(String coll, List<BulkPatchOperation> data, String partition) throws Exception {
+        doCheckBeforeBulkPatch(coll, data, partition);
+
+        var partitionKey = new PartitionKey(partition);
+        var container = this.clientV4.getDatabase(db).getContainer(coll);
+
+        log.info("begin bulkPatch coll:{}, partition:{}, account:{}", coll, partition, getAccount());
+
+        var ret = new CosmosBulkResult();
+        var retryList = new ArrayList<Object>();
+        for (int from = 0; from < data.size(); from += BULK_PATCH_CHUNK_SIZE) {
+            var to = Math.min(from + BULK_PATCH_CHUNK_SIZE, data.size());
+            var chunkData = data.subList(from, to);
+            var operations = chunkData.stream()
+                    .map(it -> CosmosBulkOperations.getPatchItemOperation(it.id, partitionKey, it.operations.getCosmosPatchOperations()))
+                    .collect(Collectors.toList());
+
+            var chunkResult = RetryUtil.executeBulkWithRetry(coll, operations,
+                    (ops) -> container.executeBulkOperations(ops));
+
+            ret.successList.addAll(chunkResult.successList);
+            ret.fatalList.addAll(chunkResult.fatalList);
+            retryList.addAll(chunkResult.retryList);
+        }
+        ret.retryList = retryList;
+
+        log.info("end bulkPatch coll:{}, partition:{}, account:{}", coll, partition, getAccount());
+        return ret;
+    }
+
     @Override
     public boolean ping(String coll) throws Exception {
         var docs = this.find(coll, Condition.filter().limit(1), "_ping");
@@ -1290,10 +1371,37 @@ static void checkBatchMaxOperations(List<?> data) {
         // There's a current limit of 100 operations per TransactionalBatch to ensure the performance is as expected and within SLAs:
         // https://learn.microsoft.com/en-us/azure/cosmos-db/nosql/transactional-batch?tabs=dotnet#limitations
         if (data.size() > MAX_BATCH_NUMBER_OF_OPERATION) {
-            throw new IllegalArgumentException("The number of data operations should not exceed 100.");
+            throw new IllegalArgumentException("The number of data operations should not exceed %d.".formatted(MAX_BATCH_NUMBER_OF_OPERATION));
         }
     }
 
+    static void doCheckBeforeBulkPatch(String coll, List<BulkPatchOperation> data, String partition) {
+        Checker.checkNotBlank(coll, "coll");
+        Checker.checkNotBlank(partition, "partition");
+        Checker.checkNotEmpty(data, "bulkPatch data " + coll + " " + partition);
+
+        for (var operation : data) {
+            Checker.checkNotNull(operation, "bulkPatch operation");
+            Checker.checkNotBlank(operation.id, "bulkPatch operation id");
+            checkValidId(operation.id);
+            Checker.checkNotNull(operation.operations, "bulkPatch operation patch operations");
+            Preconditions.checkArgument(operation.operations.size() <= PatchOperations.LIMIT,
+                    "Size of operations should be less or equal to 10. We got: %d, which exceed the limit 10",
+                    operation.operations.size());
+        }
+    }
+
+    static void doCheckBeforeBulkPatch(String coll, List<String> ids, PatchOperations operations, String partition) {
+        Checker.checkNotBlank(coll, "coll");
+        Checker.checkNotBlank(partition, "partition");
+        Checker.checkNotEmpty(ids, "bulkPatch ids " + coll + " " + partition);
+        Checker.checkNotNull(operations, "bulkPatch operations");
+        Preconditions.checkArgument(operations.size() <= PatchOperations.LIMIT,
+                "Size of operations should be less or equal to 10. We got: %d, which exceed the limit 10",
+                operations.size());
+        checkValidId(ids);
+    }
+
     /**
      * Get cosmos db account id associated with this instance.
      *