Add IdFieldName parameter to MongoDB sink extensions for custom _id mapping

ExampleConfigs.md

@@ -44,6 +44,28 @@
 }
 ```
 
+## Cosmos-NoSQL to MongoDB (with custom _id mapping)
+
+```json
+{
+    "Source": "Cosmos-NoSql",
+    "Sink": "MongoDB",
+    "SourceSettings": {
+        "ConnectionString": "AccountEndpoint=https://...",
+        "Database": "cosmicworks",
+        "Container": "baskets"
+    },
+    "SinkSettings": {
+        "ConnectionString": "mongodb://localhost:27017",
+        "DatabaseName": "mydb",
+        "Collection": "baskets",
+        "IdFieldName": "id"
+    }
+}
+```
+
+> **Note**: The `IdFieldName` parameter specifies which field from the source should be mapped to MongoDB's `_id` field. In this example, the `id` field from Cosmos will be used as the `_id` in MongoDB while also keeping the original `id` field in the document.
+
 ## MongoDB to Cosmos-NoSQL
 
 ```json

Extensions/Mongo/Cosmos.DataTransfer.MongoExtension.UnitTests/MongoDataSinkIdMappingTests.cs

@@ -0,0 +1,121 @@
+using Cosmos.DataTransfer.Interfaces;
+using MongoDB.Bson;
+
+namespace Cosmos.DataTransfer.MongoExtension.UnitTests;
+
+[TestClass]
+public class MongoDataSinkIdMappingTests
+{
+    [TestMethod]
+    public void BsonDocument_ShouldMapCustomFieldToIdWhenCreated()
+    {
+        // Arrange - Simulate a data item with an "id" field
+        var testData = new Dictionary<string, object?>
+        {
+            { "id", "BSKT_1" },
+            { "CustomerId", 111 }
+        };
+
+        // Act - Create BsonDocument and manually set _id (simulating what the sink does)
+        var bsonDoc = new BsonDocument(testData);
+        var idValue = testData["id"];
+        if (idValue != null)
+        {
+            bsonDoc["_id"] = BsonValue.Create(idValue);
+        }
+
+        // Assert
+        Assert.IsNotNull(bsonDoc["_id"]);
+        Assert.AreEqual("BSKT_1", bsonDoc["_id"].AsString);
+        Assert.AreEqual("BSKT_1", bsonDoc["id"].AsString);
+        Assert.AreEqual(111, bsonDoc["CustomerId"].AsInt32);
+        Assert.AreEqual(3, bsonDoc.ElementCount); // _id, id, CustomerId
+    }
+
+    [TestMethod]
+    public void BsonDocument_ShouldHandleStringId()
+    {
+        // Arrange
+        var testId = "string-id";
+
+        // Act
+        var bsonDoc = new BsonDocument
+        {
+            { "type", "string" }
+        };
+        bsonDoc["_id"] = BsonValue.Create(testId);
+        bsonDoc["id"] = BsonValue.Create(testId);
+
+        // Assert
+        Assert.IsNotNull(bsonDoc["_id"]);
+        Assert.IsNotNull(bsonDoc["id"]);
+        Assert.AreEqual(bsonDoc["_id"].ToString(), bsonDoc["id"].ToString());
+    }
+
+    [TestMethod]
+    public void BsonDocument_ShouldHandleIntId()
+    {
+        // Arrange
+        var testId = 12345;
+
+        // Act
+        var bsonDoc = new BsonDocument
+        {
+            { "type", "int" }
+        };
+        bsonDoc["_id"] = BsonValue.Create(testId);
+        bsonDoc["id"] = BsonValue.Create(testId);
+
+        // Assert
+        Assert.IsNotNull(bsonDoc["_id"]);
+        Assert.IsNotNull(bsonDoc["id"]);
+        Assert.AreEqual(bsonDoc["_id"].AsInt32, bsonDoc["id"].AsInt32);
+    }
+
+    [TestMethod]
+    public void BsonDocument_ShouldWorkWithCaseInsensitiveFieldNames()
+    {
+        // Arrange - Test case insensitive matching
+        var fieldNames = new[] { "id", "Id", "ID", "iD" };
+
+        foreach (var fieldName in fieldNames)
+        {
+            var testData = new Dictionary<string, object?>
+            {
+                { fieldName, "TEST_ID" },
+                { "data", "value" }
+            };
+
+            // Act
+            var bsonDoc = new BsonDocument(testData);
+            // Simulate case-insensitive lookup
+            var idField = testData.Keys.FirstOrDefault(k => k.Equals(fieldName, StringComparison.CurrentCultureIgnoreCase));
+            if (idField != null)
+            {
+                bsonDoc["_id"] = BsonValue.Create(testData[idField]);
+            }
+
+            // Assert
+            Assert.IsNotNull(bsonDoc["_id"], $"Failed for field name: {fieldName}");
+            Assert.AreEqual("TEST_ID", bsonDoc["_id"].AsString, $"Failed for field name: {fieldName}");
+        }
+    }
+
+    [TestMethod]
+    public void BsonDocument_ShouldNotOverwriteExistingIdIfNotSpecified()
+    {
+        // Arrange - Test when no IdFieldName is specified
+        var testData = new Dictionary<string, object?>
+        {
+            { "id", "BSKT_1" },
+            { "CustomerId", 111 }
+        };
+
+        // Act - Create BsonDocument without setting _id
+        var bsonDoc = new BsonDocument(testData);
+
+        // Assert - _id should not exist yet (MongoDB will add it on insert)
+        Assert.IsFalse(bsonDoc.Contains("_id"));
+        Assert.AreEqual("BSKT_1", bsonDoc["id"].AsString);
+    }
+}

Extensions/Mongo/Cosmos.DataTransfer.MongoExtension.UnitTests/MongoSinkSettingsTests.cs

@@ -0,0 +1,72 @@
+using Cosmos.DataTransfer.MongoExtension.Settings;
+
+namespace Cosmos.DataTransfer.MongoExtension.UnitTests;
+
+[TestClass]
+public class MongoSinkSettingsTests
+{
+    [TestMethod]
+    public void MongoSinkSettings_ShouldHaveIdFieldNameProperty()
+    {
+        // Arrange & Act
+        var settings = new MongoSinkSettings();
+
+        // Assert
+        Assert.IsNull(settings.IdFieldName);
+    }
+
+    [TestMethod]
+    public void MongoSinkSettings_ShouldAllowIdFieldNameToBeSet()
+    {
+        // Arrange
+        var settings = new MongoSinkSettings();
+        var testFieldName = "id";
+
+        // Act
+        settings.IdFieldName = testFieldName;
+
+        // Assert
+        Assert.AreEqual(testFieldName, settings.IdFieldName);
+    }
+
+    [TestMethod]
+    public void MongoSinkSettings_ShouldAllowNullIdFieldName()
+    {
+        // Arrange
+        var settings = new MongoSinkSettings();
+
+        // Act
+        settings.IdFieldName = null;
+
+        // Assert
+        Assert.IsNull(settings.IdFieldName);
+    }
+
+    [TestMethod]
+    public void MongoSinkSettings_ShouldAllowEmptyIdFieldName()
+    {
+        // Arrange
+        var settings = new MongoSinkSettings();
+
+        // Act
+        settings.IdFieldName = string.Empty;
+
+        // Assert
+        Assert.AreEqual(string.Empty, settings.IdFieldName);
+    }
+
+    [TestMethod]
+    public void MongoSinkSettings_ShouldSupportCaseInsensitiveFieldName()
+    {
+        // Arrange
+        var settings = new MongoSinkSettings();
+        var testFieldNames = new[] { "id", "Id", "ID", "iD" };
+
+        // Act & Assert
+        foreach (var fieldName in testFieldNames)
+        {
+            settings.IdFieldName = fieldName;
+            Assert.AreEqual(fieldName, settings.IdFieldName);
+        }
+    }
+}

Extensions/Mongo/Cosmos.DataTransfer.MongoExtension/MongoDataSinkExtension.cs

@@ -28,7 +28,23 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
             await foreach (var item in dataItems.WithCancellation(cancellationToken))
             {
                 var dict = item.BuildDynamicObjectTree();
-                objects.Add(new BsonDocument(dict));
+                var bsonDoc = new BsonDocument(dict);
+
+                // Map the specified field to _id if IdFieldName is provided
+                if (!string.IsNullOrEmpty(settings.IdFieldName) && dict != null)
+                {
+                    var sourceField = item.GetFieldNames().FirstOrDefault(n => n.Equals(settings.IdFieldName, StringComparison.CurrentCultureIgnoreCase));
+                    if (sourceField != null)
+                    {
+                        var idValue = item.GetValue(sourceField);
+                        if (idValue != null)
+                        {
+                            bsonDoc["_id"] = BsonValue.Create(idValue);
+                        }
+                    }
+                }
+
+                objects.Add(bsonDoc);
                 itemCount++;
 
                 if (objects.Count == batchSize)

Extensions/Mongo/Cosmos.DataTransfer.MongoExtension/Settings/MongoSinkSettings.cs

@@ -7,4 +7,6 @@ public class MongoSinkSettings : MongoBaseSettings
     public string? Collection { get; set; }
 
     public int? BatchSize { get; set; }
+
+    public string? IdFieldName { get; set; }
 }

Extensions/Mongo/Cosmos.DataTransfer.MongoLegacyExtension/MongoLegacyDataSinkExtension.cs

@@ -39,7 +39,23 @@ public async Task WriteAsync(IAsyncEnumerable<IDataItem> dataItems, IConfigurati
             await foreach (var item in dataItems.WithCancellation(cancellationToken))
             {
                 var dict = item.BuildDynamicObjectTree();
-                objects.Add(new BsonDocument(dict));
+                var bsonDoc = new BsonDocument(dict);
+
+                // Map the specified field to _id if IdFieldName is provided
+                if (!string.IsNullOrEmpty(settings.IdFieldName) && dict != null)
+                {
+                    var sourceField = item.GetFieldNames().FirstOrDefault(n => n.Equals(settings.IdFieldName, StringComparison.CurrentCultureIgnoreCase));
+                    if (sourceField != null)
+                    {
+                        var idValue = item.GetValue(sourceField);
+                        if (idValue != null)
+                        {
+                            bsonDoc["_id"] = BsonValue.Create(idValue);
+                        }
+                    }
+                }
+
+                objects.Add(bsonDoc);
                 itemCount++;
 
                 if (objects.Count == batchSize)

Extensions/Mongo/Cosmos.DataTransfer.MongoLegacyExtension/Settings/MongoLegacySinkSettings.cs

@@ -7,4 +7,6 @@ public class MongoLegacySinkSettings : MongoLegacyBaseSettings
     public string? Collection { get; set; }
 
     public int? BatchSize { get; set; }
+
+    public string? IdFieldName { get; set; }
 }

Extensions/Mongo/README.md

@@ -8,7 +8,7 @@ The MongoDB data transfer extension provides source and sink capabilities for re
 > 
 ## Settings
 
-Source and sink settings require both `ConnectionString` and `DatabaseName` parameters. The source takes an optional `Collection` parameter (if this parameter is not set, it will read from all collections) and an optional `Query` parameter for filtering documents. The sink requires the `Collection` parameter and will insert all records received from a source into that collection, as well as an optional `BatchSize` parameter (default value is 100) to batch the writes into the collection.
+Source and sink settings require both `ConnectionString` and `DatabaseName` parameters. The source takes an optional `Collection` parameter (if this parameter is not set, it will read from all collections) and an optional `Query` parameter for filtering documents. The sink requires the `Collection` parameter and will insert all records received from a source into that collection, as well as optional `BatchSize` (default value is 1000) and `IdFieldName` parameters.
 
 ### Source
 
@@ -72,10 +72,50 @@ For more information on MongoDB query syntax, see the [MongoDB Query Documentati
 {
     "ConnectionString": "",
     "DatabaseName": "",
-    "Collection": ""
+    "Collection": "",
+    "IdFieldName": ""
 }
 ```
 
+#### IdFieldName Parameter
+
+The `IdFieldName` parameter allows you to specify which field from the source data should be mapped to MongoDB's `_id` field. When set, the value from the specified field will be used as the document's `_id` in MongoDB, while the original field will also remain in the document.
+
+**Example:**
+
+If your source data has an `id` field that you want to use as MongoDB's `_id`:
+
+```json
+{
+    "ConnectionString": "mongodb://localhost:27017",
+    "DatabaseName": "mydb",
+    "Collection": "mycollection",
+    "IdFieldName": "id"
+}
+```
+
+**Source data:**
+```json
+{
+    "id": "BSKT_1",
+    "CustomerId": 111
+}
+```
+
+**Result in MongoDB:**
+```json
+{
+    "_id": "BSKT_1",
+    "id": "BSKT_1",
+    "CustomerId": 111
+}
+```
+
+**Notes:**
+- The field name comparison is case-insensitive
+- If `IdFieldName` is not specified or is empty, MongoDB will generate a unique ObjectId for `_id` as usual
+- The original field specified in `IdFieldName` will remain in the document along with the `_id` field
+
 # MongoDB Legacy Extension (Wire Version 2)
 
 The MongoDB Legacy extension provides source and sink capabilities for reading from and writing to MongoDB instances that use wire version 2, which is not supported by the standard MongoDB extension. This extension is specifically designed for older MongoDB instances, including Azure Cosmos DB for MongoDB API instances using the `documents.azure.com` endpoint.
@@ -103,10 +143,15 @@ Source and sink settings require both `ConnectionString` and `DatabaseName` para
     "ConnectionString": "",
     "DatabaseName": "",
     "Collection": "",
-    "BatchSize": 1000
+    "BatchSize": 1000,
+    "IdFieldName": ""
 }
 ```
 
+#### IdFieldName Parameter
+
+The `IdFieldName` parameter allows you to specify which field from the source data should be mapped to MongoDB's `_id` field. When set, the value from the specified field will be used as the document's `_id` in MongoDB, while the original field will also remain in the document. See the [MongoDB Extension Sink IdFieldName Parameter](#idfieldname-parameter) section above for more details and examples.
+
 # MongoDB Vector Extension (Beta)
 
 The MongoDB Vector extension is a Sink only extension that builds on the MongoDB extension by providing additional capabilities for generating embeddings using Azure OpenAI APIs.