Merge pull request #118 from CityStructure/cosmosdb-ignorenullvalues

introduce IgnoreNullValue setting for CosmosDB sink

Author: bowencode

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

@@ -93,7 +93,7 @@ void ReportCount(int i)
                 }
             }
 
-            var convertedObjects = dataItems.Select(di => di.BuildDynamicObjectTree(true)).Where(o => o != null).OfType<ExpandoObject>();
+            var convertedObjects = dataItems.Select(di => di.BuildDynamicObjectTree(requireStringId: true, ignoreNullValues: settings.IgnoreNullValues)).Where(o => o != null).OfType<ExpandoObject>();
             var batches = convertedObjects.Buffer(settings.BatchSize);
             var retry = GetRetryPolicy(settings.MaxRetryCount, settings.InitialRetryDurationMs);
             await foreach (var batch in batches.WithCancellation(cancellationToken))

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

@@ -15,6 +15,7 @@ public class CosmosSinkSettings : CosmosSettingsBase, IDataExtensionSettings
         public bool IsServerlessAccount { get; set; } = false;
         public bool UseSharedThroughput { 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

@@ -11,10 +11,12 @@ public static class DataItemExtensions
     /// <param name="requireStringId">If true, adds a new GUID "id" field to any top level items where one is not already present.</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)
+    public static ExpandoObject? BuildDynamicObjectTree(this IDataItem? source, bool requireStringId = false, bool ignoreNullValues = false)
     {
-        if (source == null)
+        if (source == null) 
+        {
             return null;
+        }
 
         var fields = source.GetFieldNames().ToList();
         var item = new ExpandoObject();
@@ -35,6 +37,11 @@ public static class DataItemExtensions
         foreach (string field in fields)
         {
             object? value = source.GetValue(field);
+            if (ignoreNullValues && value == null) 
+            {
+                continue;
+            }
+
             var fieldName = field;
             if (string.Equals(field, "id", StringComparison.CurrentCultureIgnoreCase) && requireStringId && !containsLowercaseIdField)
             {