Merging updates from main

Author: bowencode

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

@@ -89,12 +89,12 @@ void ReportCount(int i)
                 addedCount += i;
                 if (addedCount % 500 == 0)
                 {
-                    logger.LogInformation("{AddedCount} records added after {TotalSeconds}s", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}");
+                    logger.LogInformation("{AddedCount} records added after {TotalSeconds}s ({AddRate} records/s)", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}", $"{(int)(addedCount / (timer.ElapsedMilliseconds / 1000.0))}");
                 }
             }
 
             var convertedObjects = dataItems
-                .Select(di => di.BuildDynamicObjectTree(requireStringId: true, preserveMixedCaseIds: settings.PreserveMixedCaseIds))
+                .Select(di => di.BuildDynamicObjectTree(requireStringId: true, ignoreNullValues: settings.IgnoreNullValues, preserveMixedCaseIds: settings.PreserveMixedCaseIds))
                 .Where(o => o != null)
                 .OfType<ExpandoObject>();
             var batches = convertedObjects.Buffer(settings.BatchSize);
@@ -114,7 +114,7 @@ void ReportCount(int i)
                 throw new Exception($"Only {addedCount} of {inputCount} records were added to Cosmos");
             }
 
-            logger.LogInformation("Added {AddedCount} total records in {TotalSeconds}s", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}");
+            logger.LogInformation("Added {AddedCount} total records in {TotalSeconds}s ({AddRate} records/s)", addedCount, $"{timer.ElapsedMilliseconds / 1000.0:F2}", $"{(int)(addedCount / (timer.ElapsedMilliseconds / 1000.0))}");
         }
 
         private static AsyncRetryPolicy GetRetryPolicy(int maxRetryCount, int initialRetryDuration)

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

@@ -22,6 +22,14 @@ public static CosmosClient CreateClient(CosmosSettingsBase settings, string disp
                 AllowBulkExecution = true,
                 EnableContentResponseOnWrite = false,
             };
+            
+            if (settings is CosmosSinkSettings sinkSettings) 
+            {
+                clientOptions.SerializerOptions = new CosmosSerializationOptions 
+                {
+                    IgnoreNullValues = sinkSettings.IgnoreNullValues
+                };
+            }
 
             CosmosClient? cosmosClient;
             if (settings.UseRbacAuth)

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

@@ -16,6 +16,7 @@ public class CosmosSinkSettings : CosmosSettingsBase, IDataExtensionSettings
         public bool UseSharedThroughput { get; set; } = false;
         public bool PreserveMixedCaseIds { get; set; } = false;
         public DataWriteMode WriteMode { get; set; } = DataWriteMode.Insert;
+        public bool IgnoreNullValues { get; set; } = false;
         public List<string>? PartitionKeyPaths { get; set; }
 
         public override IEnumerable<ValidationResult> Validate(ValidationContext validationContext)

Interfaces/Cosmos.DataTransfer.Interfaces/DataItemExtensions.cs

@@ -12,10 +12,12 @@ public static class DataItemExtensions
     /// <param name="preserveMixedCaseIds">If true, disregards differently cased "id" fields for purposes of required "id" and passes them through.</param>
     /// <returns>A dynamic object containing the entire data structure.</returns>
     /// <remarks>The returned ExpandoObject can be used directly as an IDictionary.</remarks>
-    public static ExpandoObject? BuildDynamicObjectTree(this IDataItem? source, bool requireStringId = false, bool preserveMixedCaseIds = false)
+    public static ExpandoObject? BuildDynamicObjectTree(this IDataItem? source, bool requireStringId = false, bool ignoreNullValues = false, bool preserveMixedCaseIds = false)
     {
-        if (source == null)
+        if (source == null) 
+        {
             return null;
+        }
 
         var fields = source.GetFieldNames().ToList();
         var item = new ExpandoObject();
@@ -41,6 +43,11 @@ public static class DataItemExtensions
         foreach (string field in fields)
         {
             object? value = source.GetValue(field);
+            if (ignoreNullValues && value == null) 
+            {
+                continue;
+            }
+
             var fieldName = field;
             if (requireStringId && string.Equals(field, "id", StringComparison.CurrentCultureIgnoreCase))
             {