CosmosDB-Mongodb IsUpsert无法用于批量更新

如何解决CosmosDB-Mongodb IsUpsert无法用于批量更新

最近几个月，我们通过.NET Core和最新的MongoDB.Driver Nuget软件包（2.11.0）在MonsDB和CosmosDB（Server v3.6）中广泛使用了MongoDB API。

批量插入和插入工作正常，但是很遗憾，我无法使批量操作与IsUpsert=true模式一起使用。

注意：

• 我们使用Polly来管理速率限制。在此过程中，我们处理MongoWriteException,MongoExecutionTimeoutException,MongoCommandException和MongoBulkWriteExceptions。
• 分片/非分片集合都可以观察到此问题。

具体来说，给定非分片输入文档List<T> documents的列表，以下方法可以正常工作：

1. 批量插入：

   await Collection.BulkWriteAsync(documents.Select(s => new InsertOneModel<T>(s)),...)
   

2. 批量更新：

   await Collection.BulkWriteAsync(documents.Select(s =>
     new ReplaceOneModel<T>(Builders<T>.Filter.Eq("Id",item.Id),item) { IsUpsert = false }),...)
   

不幸的是，如果某些文档是新文档，我们应该能够按原样使用上面的批量更新代码-只需将IsUpsert标志设置为true ...但是，这是行不通的。

具体来说，提供了50个现有文档和50个新文档：

• 如果文档的ID类型为ObjectId作为主键，则对于它处理的第一个新文档，CosmosDb将错误地使用Id=ObjectId("000000000000000000000000")插入它-届时将不再插入/更新其他文档。在这种情况下：
  • BulkWriteResult返回了MatchedCount=65,ModifiedCount=65,ProcessedRequests=100,RequestCount=100,Upserts=1,IsAcknowledged=true,IsModifiedCountAvailable=true,InsertedCount=0
  • 没有引发异常。
  • 注意-数据库中只有51个文档，因此不能依赖BulkWriteResult
• 如果文档的ID类型为int作为主键，那么cosmos db似乎可以
  • 在某些随机点放弃处理文档。这似乎是一种限制速率的方案... 除了不会引发异常。
  • 例如，更新所有50个文档，但仅插入8个。在这种情况下，BulkWriteResult返回MatchedCount=50,ModifiedCount=50,Upserts=8,InsertedCount=0。

我想念什么？ ObjectId场景似乎完全崩溃了；其他情况下可能会被编码，但是这里没有引发异常似乎并不正确。

解决方法

对于其他受此问题困扰的人-解决方法远非简单易懂，但这就是我最终要做的事情。

• 主键为ObjectId 的文档：只要您使用'InsertOneModel ，此方案就可以在CosmosDB和MongoDB中一致地工作或ReplaceOneModel<>基于标识符的值是否为ObjectId.Empty。但是，您可能仍然必须处理以下提到的 一次错误 。
• 主键不是ObjectId 的文档：肯定是CosmosDb 中的 bug，因为我无法在官方{{ 1}}实现。要解决此问题，我必须应用以下两种解决方法：
  • 抛出自定义异常并更新我现有的MongoDB策略以重试未处理的请求，就像我通常对Polly通常抛出的其他MongoDB速率限制异常所做的那样。示例代码：

  CosmosDB

• 关闭一个错误处理。不确定在纯BulkWriteResult<T> bulkWriteResult = await Collection .BulkWriteAsync( remainingWork,new BulkWriteOptions { BypassDocumentValidation = true },token); var actuallyProcessed = bulkWriteResult.DeletedCount + bulkWriteResult.InsertedCount + bulkWriteResult.ModifiedCount + bulkWriteResult.Upserts?.Count; if (actuallyProcessed < bulkWriteResult.ProcessedRequests.Count) { // Off by one error: OCCASIONALLY,the last one processed is not actually processed // No way to detect this,unfortunately - hence the adjustment by 1 actuallyProcessed = actuallyProcessed > 1 ? actuallyProcessed - 1 : 0; var processed = bulkWriteResult.ProcessedRequests.Take((int)actuallyProcessed) .ToList().AsReadOnly(); var unprocessed = bulkWriteResult.ProcessedRequests.Skip((int)actuallyProcessed) .ToList().AsReadOnly(); throw new CosmosDbRateLimitingBugException<T>(unprocessed,processed,bulkWriteResult); } 实现中是否需要这样做，但是就像上面一样，有时您还必须将处理的记录调整1。注意：无论使用'IsUpsert = true'，此问题都适用。当我使用MongoDB来跟踪异常和已处理/未处理的记录（未显示）时，下面的代码略有简化。这里的Polly.Context是必须向下一个remainingWork调用发出的WriteModel<T>请求。

BulkWriteAsync<>