introduce DataWriteMode.Delete to undo inserts to the wrong container

Author: stefann42

Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs

@@ -151,7 +151,7 @@ private static Task<ItemResult> AddItemAsync(Container container, ExpandoObject
                         return t.Result;
                     }
 
-                    return new ItemResult(null, HttpStatusCode.InternalServerError);
+                    return new ItemResult(null, mode, HttpStatusCode.InternalServerError);
                 }, cancellationToken);
             return task;
         }
@@ -179,14 +179,24 @@ private static async Task<ItemResult> PopulateItem(Container container, ExpandoO
                     var upsertResponse = await container.UpsertItemAsync(item, cancellationToken: cancellationToken);
                     statusCode = upsertResponse.StatusCode;
                     break;
+                case DataWriteMode.DeleteStream:
+                    ArgumentNullException.ThrowIfNull(partitionKeyPath, nameof(partitionKeyPath));
+                    var deleteMessage = await container.DeleteItemStreamAsync(itemId, new PartitionKey(GetPropertyValue(item, partitionKeyPath.TrimStart('/'))), cancellationToken: cancellationToken);
+                    statusCode = deleteMessage.StatusCode;
+                    break;
+                case DataWriteMode.Delete:
+                    ArgumentNullException.ThrowIfNull(partitionKeyPath, nameof(partitionKeyPath));
+                    var deleteResponse = await container.DeleteItemAsync<ExpandoObject>(itemId, new PartitionKey(GetPropertyValue(item, partitionKeyPath.TrimStart('/'))), cancellationToken: cancellationToken);
+                    statusCode = deleteResponse.StatusCode;
+                    break;
             }
 
             if (statusCode == null)
             {
                 throw new ArgumentOutOfRangeException(nameof(mode), $"Invalid data write mode specified: {mode}");
             }
 
-            return new ItemResult(itemId, statusCode.Value);
+            return new ItemResult(itemId, mode, statusCode.Value);
         }
 
         private static MemoryStream CreateItemStream(ExpandoObject item)
@@ -205,9 +215,10 @@ public IEnumerable<IDataExtensionSettings> GetSettings()
             yield return new CosmosSinkSettings();
         }
 
-        public record ItemResult(string? Id, HttpStatusCode StatusCode)
+        public record ItemResult(string? Id, DataWriteMode DataWriteMode, HttpStatusCode StatusCode)
         {
-            public bool IsSuccess => StatusCode is HttpStatusCode.OK or HttpStatusCode.Created;
+            public bool IsSuccess => StatusCode is HttpStatusCode.OK or HttpStatusCode.Created ||
+                (StatusCode is HttpStatusCode.NoContent or HttpStatusCode.NotFound && DataWriteMode is DataWriteMode.Delete or DataWriteMode.DeleteStream);
             public int ItemCount => IsSuccess ? 1 : 0;
         }
     }

Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/DataWriteMode.cs

@@ -6,5 +6,7 @@ public enum DataWriteMode
         Insert,
         UpsertStream,
         Upsert,
+        DeleteStream,
+        Delete
     }
 }

Extensions/Cosmos/README.md

@@ -44,7 +44,17 @@ Or with RBAC:
 }
 ```
 
-Sink requires an additional `PartitionKeyPath` parameter which is used when creating the container if it does not exist. To use hierarchical partition keys, instead use the `PartitionKeyPaths` setting to supply an array of up to 3 paths. It also supports an optional `RecreateContainer` parameter (`false` by default) to delete and then recreate the container to ensure only newly imported data is present. The optional `BatchSize` parameter (100 by default) sets the number of items to accumulate before inserting. `ConnectionMode` can be set to either `Gateway` (default) or `Direct` to control how the client connects to the CosmosDB service. For situations where a container is created as part of the transfer operation `CreatedContainerMaxThroughput` (in RUs) and `UseAutoscaleForCreatedContainer` provide the initial throughput settings which will be in effect when executing the transfer. To instead use shared throughput that has been provisioned at the database level, set the `UseSharedThroughput` parameter to `true`. The optional `WriteMode` parameter specifies the type of data write to use: `InsertStream`, `Insert`, `UpsertStream`, or `Upsert`. The `IsServerlessAccount` parameter specifies whether the target account uses Serverless instead of Provisioned throughput, which affects the way containers are created. Additional parameters allow changing the behavior of the Cosmos client appropriate to your environment.
+Sink requires an additional `PartitionKeyPath` parameter which is used when creating the container if it does not exist. To use hierarchical partition keys, instead use the `PartitionKeyPaths` setting to supply an array of up to 3 paths. It also supports an optional `RecreateContainer` parameter (`false` by default) to delete and then recreate the container to ensure only newly imported data is present. 
+
+The optional `BatchSize` parameter (100 by default) sets the number of items to accumulate before inserting. 
+
+`ConnectionMode` can be set to either `Gateway` (default) or `Direct` to control how the client connects to the CosmosDB service. 
+
+For situations where a container is created as part of the transfer operation `CreatedContainerMaxThroughput` (in RUs) and `UseAutoscaleForCreatedContainer` provide the initial throughput settings which will be in effect when executing the transfer. To instead use shared throughput that has been provisioned at the database level, set the `UseSharedThroughput` parameter to `true`. 
+
+The optional `WriteMode` parameter specifies the type of data write to use: `InsertStream`, `Insert`, `UpsertStream`, `Upsert`, `DeleteStream` or `Delete`. The latter is useful to undo inserts that might have been written to the wrong container by mistake, if you cannot delete the entire container because it might contain other unrelated data.  
+
+The `IsServerlessAccount` parameter specifies whether the target account uses Serverless instead of Provisioned throughput, which affects the way containers are created. Additional parameters allow changing the behavior of the Cosmos client appropriate to your environment.
 
 ### Sink