Skip to content

Commit 2230c03

Browse files
bowencodebowencode
committed
Merging from main
2 parents e19701b + 7eb9c90 commit 2230c03

File tree

6 files changed

+126
-14
lines changed

6 files changed

+126
-14
lines changed

‎Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension.UnitTests/CosmosDataSinkExtensionTests.cs

+78
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using Cosmos.DataTransfer.Interfaces;
2+
using Microsoft.VisualStudio.TestTools.UnitTesting;
23

34
namespace Cosmos.DataTransfer.CosmosExtension.UnitTests
45
{
@@ -54,5 +55,82 @@ public void BuildDynamicObjectTree_WithNestedArrays_WorksCorrectly()
5455

5556
Assert.AreEqual("sub2-1", secondSubArray[0].id);
5657
}
58+
59+
[TestMethod]
60+
public void BuildDynamicObjectTree_WithAnyCaseIds_UsesSourceIdValue()
61+
{
62+
var numeric = Random.Shared.Next();
63+
var lower = Guid.NewGuid().ToString();
64+
var upper = Guid.NewGuid().ToString();
65+
var mixed = Guid.NewGuid().ToString();
66+
var reversed = Guid.NewGuid().ToString();
67+
var item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
68+
{
69+
{ "id", numeric },
70+
});
71+
72+
dynamic obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: false)!;
73+
Assert.AreEqual(numeric.ToString(), obj.id);
74+
75+
item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
76+
{
77+
{ "id", lower },
78+
});
79+
80+
obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: false)!;
81+
Assert.AreEqual(lower, obj.id);
82+
83+
item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
84+
{
85+
{ "ID", upper },
86+
});
87+
obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: false)!;
88+
Assert.AreEqual(upper, obj.id);
89+
90+
item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
91+
{
92+
{ "Id", mixed },
93+
});
94+
obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: false)!;
95+
Assert.AreEqual(mixed, obj.id);
96+
97+
item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
98+
{
99+
{ "iD", reversed },
100+
});
101+
obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: false)!;
102+
Assert.AreEqual(reversed, obj.id);
103+
}
104+
105+
[TestMethod]
106+
public void BuildDynamicObjectTree_WithPreservedMixedCaseIds_PassesThroughSourceValues()
107+
{
108+
var id = Random.Shared.Next();
109+
var upper = Guid.NewGuid().ToString();
110+
var mixed = Guid.NewGuid().ToString();
111+
var item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
112+
{
113+
{ "id", id },
114+
{ "ID", upper },
115+
{ "Id", mixed }
116+
});
117+
118+
dynamic obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: true)!;
119+
Assert.AreEqual(id.ToString(), obj.id);
120+
Assert.AreEqual(upper, obj.ID);
121+
Assert.AreEqual(mixed, obj.Id);
122+
123+
item = new CosmosDictionaryDataItem(new Dictionary<string, object?>()
124+
{
125+
{ "ID", upper },
126+
{ "Id", mixed }
127+
});
128+
obj = item.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: true)!;
129+
Assert.AreEqual(upper, obj.ID);
130+
Assert.AreEqual(mixed, obj.Id);
131+
string? cosmosId = obj.id;
132+
Assert.IsNotNull(cosmosId);
133+
Assert.IsFalse(string.IsNullOrWhiteSpace(cosmosId));
134+
}
57135
}
58136
}

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

+6-3
Original file line numberDiff line numberDiff line change
@@ -89,11 +89,14 @@ void ReportCount(int i)
8989
addedCount += i;
9090
if (addedCount % 500 == 0)
9191
{
92-
logger.LogInformation("{AddedCount} records added after {TotalSeconds}s", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}");
92+
logger.LogInformation("{AddedCount} records added after {TotalSeconds}s ({AddRate} records/s)", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}", $"{(int)(addedCount / (timer.ElapsedMilliseconds / 1000.0))}");
9393
}
9494
}
9595

96-
var convertedObjects = dataItems.Select(di => di.BuildDynamicObjectTree(true)).Where(o => o != null).OfType<ExpandoObject>();
96+
var convertedObjects = dataItems
97+
.Select(di => di.BuildDynamicObjectTree(requireStringId: true, ignoreNullValues: settings.IgnoreNullValues, preserveMixedCaseIds: settings.PreserveMixedCaseIds))
98+
.Where(o => o != null)
99+
.OfType<ExpandoObject>();
97100
var batches = convertedObjects.Buffer(settings.BatchSize);
98101
var retry = GetRetryPolicy(settings.MaxRetryCount, settings.InitialRetryDurationMs);
99102
await foreach (var batch in batches.WithCancellation(cancellationToken))
@@ -111,7 +114,7 @@ void ReportCount(int i)
111114
throw new Exception($"Only {addedCount} of {inputCount} records were added to Cosmos");
112115
}
113116

114-
logger.LogInformation("Added {AddedCount} total records in {TotalSeconds}s", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}");
117+
logger.LogInformation("Added {AddedCount} total records in {TotalSeconds}s ({AddRate} records/s)", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}", $"{(int)(addedCount / (timer.ElapsedMilliseconds / 1000.0))}");
115118
}
116119

117120
private static AsyncRetryPolicy GetRetryPolicy(int maxRetryCount, int initialRetryDuration)

‎Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosExtensionServices.cs

+8
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,14 @@ public static CosmosClient CreateClient(CosmosSettingsBase settings, string disp
2323
EnableContentResponseOnWrite = false,
2424
Serializer = new RawJsonCosmosSerializer(),
2525
};
26+
27+
if (settings is CosmosSinkSettings sinkSettings)
28+
{
29+
clientOptions.SerializerOptions = new CosmosSerializationOptions
30+
{
31+
IgnoreNullValues = sinkSettings.IgnoreNullValues
32+
};
33+
}
2634

2735
CosmosClient? cosmosClient;
2836
if (settings.UseRbacAuth)

‎Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosSinkSettings.cs

+2
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,9 @@ public class CosmosSinkSettings : CosmosSettingsBase, IDataExtensionSettings
1414
public bool UseAutoscaleForCreatedContainer { get; set; } = true;
1515
public bool IsServerlessAccount { get; set; } = false;
1616
public bool UseSharedThroughput { get; set; } = false;
17+
public bool PreserveMixedCaseIds { get; set; } = false;
1718
public DataWriteMode WriteMode { get; set; } = DataWriteMode.Insert;
19+
public bool IgnoreNullValues { get; set; } = false;
1820
public List<string>? PartitionKeyPaths { get; set; }
1921

2022
public override IEnumerable<ValidationResult> Validate(ValidationContext validationContext)

‎Extensions/Cosmos/README.md

+2-1
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ Or with RBAC:
4444
}
4545
```
4646

47-
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.
47+
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. The `PreserveMixedCaseIds` parameter (`false` by default) ignores differently cased `id` fields and writes them through without modification, while generating a separate lowercased `id` field as required by Cosmos.
4848

4949
### Sink
5050

@@ -62,6 +62,7 @@ Sink requires an additional `PartitionKeyPath` parameter which is used when crea
6262
"CreatedContainerMaxThroughput": 1000,
6363
"UseAutoscaleForCreatedContainer": true,
6464
"WriteMode": "InsertStream",
65+
"PreserveMixedCaseIds": false,
6566
"IsServerlessAccount": false,
6667
"UseSharedThroughput": false
6768
}

‎Interfaces/Cosmos.DataTransfer.Interfaces/DataItemExtensions.cs

+30-10
Original file line numberDiff line numberDiff line change
@@ -9,36 +9,56 @@ public static class DataItemExtensions
99
/// </summary>
1010
/// <param name="source"></param>
1111
/// <param name="requireStringId">If true, adds a new GUID "id" field to any top level items where one is not already present.</param>
12+
/// <param name="preserveMixedCaseIds">If true, disregards differently cased "id" fields for purposes of required "id" and passes them through.</param>
1213
/// <returns>A dynamic object containing the entire data structure.</returns>
1314
/// <remarks>The returned ExpandoObject can be used directly as an IDictionary.</remarks>
14-
public static ExpandoObject? BuildDynamicObjectTree(this IDataItem? source, bool requireStringId = false)
15+
public static ExpandoObject? BuildDynamicObjectTree(this IDataItem? source, bool requireStringId = false, bool ignoreNullValues = false, bool preserveMixedCaseIds = false)
1516
{
16-
if (source == null)
17+
if (source == null)
18+
{
1719
return null;
20+
}
1821

1922
var fields = source.GetFieldNames().ToList();
2023
var item = new ExpandoObject();
21-
24+
2225
/*
2326
* If the item contains a lowercase id field, we can take it as is.
24-
* If we have an uppercase Id or ID field, but no lowercase id, we will rename it to id.
27+
* If we have an uppercase Id or ID field, but no lowercase id, we will rename it to id, unless `preserveMixedCaseIds` is set to true.
28+
* If `preserveMixedCaseIds` is set to true, any differently cased "id" fields will be passed through as normal properties with no casing change and a separate "id" will be generated.
2529
* Then it can be used i.e. as CosmosDB primary key, when `requireStringId` is set to true.
2630
*/
2731
var containsLowercaseIdField = fields.Contains("id", StringComparer.CurrentCulture);
2832
var containsAnyIdField = fields.Contains("id", StringComparer.CurrentCultureIgnoreCase);
29-
30-
if (requireStringId && !containsAnyIdField)
33+
34+
if (requireStringId)
3135
{
32-
item.TryAdd("id", Guid.NewGuid().ToString());
36+
bool mismatchedIdCasing = preserveMixedCaseIds && !containsLowercaseIdField;
37+
if (!containsAnyIdField || mismatchedIdCasing)
38+
{
39+
item.TryAdd("id", Guid.NewGuid().ToString());
40+
}
3341
}
34-
42+
3543
foreach (string field in fields)
3644
{
3745
object? value = source.GetValue(field);
46+
if (ignoreNullValues && value == null)
47+
{
48+
continue;
49+
}
50+
3851
var fieldName = field;
39-
if (string.Equals(field, "id", StringComparison.CurrentCultureIgnoreCase) && requireStringId)
52+
if (requireStringId && string.Equals(field, "id", StringComparison.CurrentCultureIgnoreCase))
4053
{
41-
if (!containsLowercaseIdField)
54+
if (preserveMixedCaseIds)
55+
{
56+
if (string.Equals(field, "id", StringComparison.CurrentCulture))
57+
{
58+
value = value?.ToString();
59+
}
60+
}
61+
else if (!containsLowercaseIdField)
4262
{
4363
value = value?.ToString();
4464
fieldName = "id";

0 commit comments

Comments
 (0)