Merge pull request #122 from AzureCosmosDB/feature/type-metadata-pass…

…through

Fixes #97 for outputting JSON metadata properties on child objects

Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDictionaryDataItem.cs

@@ -1,4 +1,5 @@
 using Cosmos.DataTransfer.Interfaces;
+using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 
 namespace Cosmos.DataTransfer.CosmosExtension
@@ -31,11 +32,13 @@ public IEnumerable<string> GetFieldNames()
         {
             if (value is JObject element)
             {
-                return new CosmosDictionaryDataItem(element.ToObject<IDictionary<string, object?>>().ToDictionary(k => k.Key, v => v.Value));
+                return new CosmosDictionaryDataItem(element.ToObject<IDictionary<string, object?>>(JsonSerializer.Create(RawJsonCosmosSerializer.GetDefaultSettings()))
+                    .ToDictionary(k => k.Key, v => v.Value));
             }
             if (value is JArray array)
             {
-                return array.ToObject<List<object?>>().Select(GetChildObject).ToList();
+                return array.ToObject<List<object?>>(JsonSerializer.Create(RawJsonCosmosSerializer.GetDefaultSettings()))
+                    .Select(GetChildObject).ToList();
             }
 
             return value;

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

@@ -15,22 +15,23 @@ public static CosmosClient CreateClient(CosmosSettingsBase settings, string disp
         {
             string userAgentString = CreateUserAgentString(displayName, sourceDisplayName);
 
+            var cosmosSerializer = new RawJsonCosmosSerializer();
+            if (settings is CosmosSinkSettings sinkSettings)
+            {
+                cosmosSerializer.SerializerSettings.NullValueHandling = sinkSettings.IgnoreNullValues
+                    ? Newtonsoft.Json.NullValueHandling.Ignore
+                    : Newtonsoft.Json.NullValueHandling.Include;
+            }
+
             var clientOptions = new CosmosClientOptions
             {
                 ConnectionMode = settings.ConnectionMode,
                 ApplicationName = userAgentString,
                 AllowBulkExecution = true,
                 EnableContentResponseOnWrite = false,
+                Serializer = cosmosSerializer,
             };
 
-            if (settings is CosmosSinkSettings sinkSettings) 
-            {
-                clientOptions.SerializerOptions = new CosmosSerializationOptions 
-                {
-                    IgnoreNullValues = sinkSettings.IgnoreNullValues
-                };
-            }
-
             CosmosClient? cosmosClient;
             if (settings.UseRbacAuth)
             {

Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/RawJsonCosmosSerializer.cs

@@ -0,0 +1,62 @@
+using Microsoft.Azure.Cosmos;
+using Newtonsoft.Json;
+using System.Text;
+
+namespace Cosmos.DataTransfer.CosmosExtension;
+
+/// <summary>
+/// Serializer for Cosmos allowing access to internal JsonSerializer settings.
+/// </summary>
+/// <remarks>
+/// Defaults to disabling metadata handling to allow passthrough of recognized properties like "$type".
+/// </remarks>
+public class RawJsonCosmosSerializer : CosmosSerializer
+{
+    private static readonly Encoding DefaultEncoding = new UTF8Encoding(false, true);
+
+    public static JsonSerializerSettings GetDefaultSettings() =>
+        new()
+        {
+            DateParseHandling = DateParseHandling.None,
+            MetadataPropertyHandling = MetadataPropertyHandling.Ignore,
+            ContractResolver = null,
+            MaxDepth = 64,
+        };
+
+    public JsonSerializerSettings SerializerSettings { get; } = GetDefaultSettings();
+
+    public override T FromStream<T>(Stream stream)
+    {
+        using (stream)
+        {
+            if (typeof(Stream).IsAssignableFrom(typeof(T)))
+            {
+                return (T)(object)stream;
+            }
+
+            using var streamReader = new StreamReader(stream);
+            using var jsonReader = new JsonTextReader(streamReader);
+            var serializer = JsonSerializer.Create(SerializerSettings);
+            return serializer.Deserialize<T>(jsonReader);
+        }
+    }
+
+    public override Stream ToStream<T>(T input)
+    {
+        var memoryStream = new MemoryStream();
+        using (var streamWriter = new StreamWriter(memoryStream, DefaultEncoding, bufferSize: 1024, leaveOpen: true))
+        {
+            using (var jsonWriter = new JsonTextWriter(streamWriter))
+            {
+                jsonWriter.Formatting = Formatting.None;
+                var serializer = JsonSerializer.Create(SerializerSettings);
+                serializer.Serialize(jsonWriter, input);
+                jsonWriter.Flush();
+                streamWriter.Flush();
+            }
+        }
+
+        memoryStream.Position = 0;
+        return memoryStream;
+    }
+}

Extensions/Json/Cosmos.DataTransfer.JsonExtension.UnitTests/JsonSourceTests.cs

@@ -154,6 +154,56 @@ public async Task ReadAsync_WithFlatObjects_ReadsValuesFromUrl()
             }
         }
 
+        [TestMethod]
+        public async Task ReadAsync_WithTypeHintFields_IncludesAllInOutput()
+        {
+            var json = @"[
+{
+    ""id"": 1,
+    ""name"": ""One"",
+    ""$type"": ""Number"",
+    ""data"": {
+        ""$type"": ""Object"",
+        ""name"": ""A""
+    }
+},
+{
+    ""id"": 2,
+    ""name"": ""Two"",
+    ""$type"": ""Digit"",
+    ""data"": {
+        ""$type"": ""String"",
+        ""name"": ""B""
+    }
+}
+]";
+            var filePath = Path.Combine(Path.GetTempPath(), "TypeHintFields.json");
 
+            await File.WriteAllTextAsync(filePath, json);
+
+            var extension = new JsonFileSource();
+            var config = TestHelpers.CreateConfig(new Dictionary<string, string>
+            {
+                { "FilePath", filePath }
+            });
+
+            int counter = 0;
+            await foreach (var dataItem in extension.ReadAsync(config, NullLogger.Instance))
+            {
+                counter++;
+                var fields = dataItem.GetFieldNames().ToArray();
+                CollectionAssert.AreEquivalent(new[] { "id", "name", "$type", "data" }, fields);
+                Assert.IsNotNull(dataItem.GetValue("id"));
+                Assert.IsNotNull(dataItem.GetValue("name"));
+                Assert.IsNotNull(dataItem.GetValue("$type"));
+                var child = dataItem.GetValue("data") as JsonDictionaryDataItem;
+                Assert.IsNotNull(child);
+                CollectionAssert.AreEquivalent(new[] { "$type", "name" }, child.GetFieldNames().ToArray());
+                Assert.IsNotNull(child.GetValue("$type"));
+                Assert.IsNotNull(child.GetValue("name"));
+            }
+
+            Assert.AreEqual(2, counter);
+        }
     }
 }