Fix Serialization/Deserialization issue with $ prefix columns (#3051) (#3137)

## Why make this change?
Serialization and deserialization of metadata currently fail when column
names are prefixed with the $ symbol.

## What is this change?
This pull request enhances the serialization and deserialization logic
for database metadata objects to properly handle column names that start
with a dollar sign ($). It introduces a mechanism to escape such column
names during serialization and unescape them during deserialization,
ensuring compatibility and correctness when processing metadata with
special column names.

## How was this tested?
- [x] Unit Tests

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

Author: Alekhya-Polavarapu

src/Core/Services/MetadataProviders/Converters/DatabaseObjectConverter.cs

@@ -17,6 +17,11 @@ namespace Azure.DataApiBuilder.Core.Services.MetadataProviders.Converters
     public class DatabaseObjectConverter : JsonConverter<DatabaseObject>
     {
         private const string TYPE_NAME = "TypeName";
+        private const string DOLLAR_CHAR = "$";
+
+        // ``DAB_ESCAPE$`` is used to escape column names that start with `$` during serialization.
+        // It is chosen to be unique enough to avoid collisions with actual column names.
+        private const string ESCAPED_DOLLARCHAR = "DAB_ESCAPE$";
 
         public override DatabaseObject Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
         {
@@ -29,6 +34,15 @@ public override DatabaseObject Read(ref Utf8JsonReader reader, Type typeToConver
 
                 DatabaseObject objA = (DatabaseObject)JsonSerializer.Deserialize(document, concreteType, options)!;
 
+                foreach (PropertyInfo prop in objA.GetType().GetProperties().Where(IsSourceDefinitionOrDerivedClassProperty))
+                {
+                    SourceDefinition? sourceDef = (SourceDefinition?)prop.GetValue(objA);
+                    if (sourceDef is not null)
+                    {
+                        UnescapeDollaredColumns(sourceDef);
+                    }
+                }
+
                 return objA;
             }
         }
@@ -58,12 +72,72 @@ public override void Write(Utf8JsonWriter writer, DatabaseObject value, JsonSeri
                 }
 
                 writer.WritePropertyName(prop.Name);
-                JsonSerializer.Serialize(writer, prop.GetValue(value), options);
+                object? propVal = prop.GetValue(value);
+
+                // Only escape columns for properties whose type(derived type) is SourceDefinition.
+                if (IsSourceDefinitionOrDerivedClassProperty(prop) && propVal is SourceDefinition sourceDef)
+                {
+                    EscapeDollaredColumns(sourceDef);
+                }
+
+                JsonSerializer.Serialize(writer, propVal, options);
             }
 
             writer.WriteEndObject();
         }
 
+        private static bool IsSourceDefinitionOrDerivedClassProperty(PropertyInfo prop)
+        {
+            // Return true for properties whose type is SourceDefinition or any class derived from SourceDefinition
+            return typeof(SourceDefinition).IsAssignableFrom(prop.PropertyType);
+        }
+
+        /// <summary>
+        /// Escapes column keys that start with '$' by prefixing them with 'DAB_ESCAPE$' for serialization.
+        /// </summary>
+        private static void EscapeDollaredColumns(SourceDefinition sourceDef)
+        {
+            if (sourceDef.Columns is null || sourceDef.Columns.Count == 0)
+            {
+                return;
+            }
+
+            List<string> keysToEscape = sourceDef.Columns.Keys
+                .Where(k => k.StartsWith(DOLLAR_CHAR, StringComparison.Ordinal))
+                .ToList();
+
+            foreach (string key in keysToEscape)
+            {
+                ColumnDefinition col = sourceDef.Columns[key];
+                sourceDef.Columns.Remove(key);
+                string newKey = ESCAPED_DOLLARCHAR + key[1..];
+                sourceDef.Columns[newKey] = col;
+            }
+        }
+
+        /// <summary>
+        /// Unescapes column keys that start with 'DAB_ESCAPE$' by removing the prefix and restoring the original '$' for deserialization.
+        /// </summary>
+        private static void UnescapeDollaredColumns(SourceDefinition sourceDef)
+        {
+            if (sourceDef.Columns is null || sourceDef.Columns.Count == 0)
+            {
+                return;
+            }
+
+            List<string> keysToUnescape = sourceDef.Columns.Keys
+                .Where(k => k.StartsWith(ESCAPED_DOLLARCHAR, StringComparison.Ordinal))
+                .ToList();
+
+            foreach (string key in keysToUnescape)
+            {
+                ColumnDefinition col = sourceDef.Columns[key];
+                sourceDef.Columns.Remove(key);
+                string newKey = DOLLAR_CHAR + key[ESCAPED_DOLLARCHAR.Length..];
+                sourceDef.Columns[newKey] = col;
+            }
+        }
+
         private static Type GetTypeFromName(string typeName)
         {
             Type? type = Type.GetType(typeName);

src/Service.Tests/UnitTests/SerializationDeserializationTests.cs

@@ -277,8 +277,114 @@ public void TestDictionaryDatabaseObjectSerializationDeserialization()
             VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseTable.TableDefinition, _databaseTable.TableDefinition, "FirstName");
         }
 
-        private void InitializeObjects()
+        /// <summary>
+        /// Validates serialization and deserialization of Dictionary containing DatabaseTable
+        /// The table will have dollar sign prefix ($) in the column name
+        /// this is how we serialize and deserialize metadataprovider.EntityToDatabaseObject dict.
+        /// </summary>
+        [TestMethod]
+        public void TestDictionaryDatabaseObjectSerializationDeserialization_WithDollarColumn()
+        {
+            InitializeObjects(generateDollaredColumn: true);
+
+            _options = new()
+            {
+                Converters = {
+                    new DatabaseObjectConverter(),
+                    new TypeConverter()
+                },
+                ReferenceHandler = ReferenceHandler.Preserve
+            };
+
+            Dictionary<string, DatabaseObject> dict = new() { { "person", _databaseTable } };
+
+            string serializedDict = JsonSerializer.Serialize(dict, _options);
+            // Assert that the serialized JSON contains the escaped dollar sign in column name
+            Assert.IsTrue(serializedDict.Contains("DAB_ESCAPE$FirstName"),
+                "Serialized JSON should contain the dollar-prefixed column name in SourceDefinition's Columns.");
+
+            Dictionary<string, DatabaseObject> deserializedDict = JsonSerializer.Deserialize<Dictionary<string, DatabaseObject>>(serializedDict, _options)!;
+            DatabaseTable deserializedDatabaseTable = (DatabaseTable)deserializedDict["person"];
+
+            Assert.AreEqual(deserializedDatabaseTable.SourceType, _databaseTable.SourceType);
+            Assert.AreEqual(deserializedDatabaseTable.FullName, _databaseTable.FullName);
+            deserializedDatabaseTable.Equals(_databaseTable);
+            VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseTable.SourceDefinition, _databaseTable.SourceDefinition, "$FirstName");
+            VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseTable.TableDefinition, _databaseTable.TableDefinition, "$FirstName");
+        }
+
+        /// <summary>
+        /// Validates serialization and deserialization of Dictionary containing DatabaseView
+        /// The table will have dollar sign prefix ($) in the column name
+        /// this is how we serialize and deserialize metadataprovider.EntityToDatabaseObject dict.
+        /// </summary>
+        [TestMethod]
+        public void TestDatabaseViewSerializationDeserialization_WithDollarColumn()
+        {
+            InitializeObjects(generateDollaredColumn: true);
+
+            TestTypeNameChanges(_databaseView, "DatabaseView");
+
+            Dictionary<string, DatabaseObject> dict = new();
+            dict.Add("person", _databaseView);
+
+            // Test to catch if there is change in number of properties/fields
+            // Note: On Addition of property make sure it is added in following object creation _databaseView and include in serialization
+            // and deserialization test.
+            int fields = typeof(DatabaseView).GetProperties(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance).Length;
+            Assert.AreEqual(fields, 6);
+
+            string serializedDatabaseView = JsonSerializer.Serialize(dict, _options);
+            // Assert that the serialized JSON contains the escaped dollar sign in column name
+            Assert.IsTrue(serializedDatabaseView.Contains("DAB_ESCAPE$FirstName"),
+                "Serialized JSON should contain the dollar-prefixed column name in SourceDefinition's Columns.");
+            Dictionary<string, DatabaseObject> deserializedDict = JsonSerializer.Deserialize<Dictionary<string, DatabaseObject>>(serializedDatabaseView, _options)!;
+
+            DatabaseView deserializedDatabaseView = (DatabaseView)deserializedDict["person"];
+
+            Assert.AreEqual(deserializedDatabaseView.SourceType, _databaseView.SourceType);
+            deserializedDatabaseView.Equals(_databaseView);
+            VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseView.SourceDefinition, _databaseView.SourceDefinition, "$FirstName");
+            VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseView.ViewDefinition, _databaseView.ViewDefinition, "$FirstName");
+        }
+
+        /// <summary>
+        /// Validates serialization and deserialization of Dictionary containing DatabaseStoredProcedure
+        /// The table will have dollar sign prefix ($) in the column name
+        /// this is how we serialize and deserialize metadataprovider.EntityToDatabaseObject dict.
+        /// </summary>
+        [TestMethod]
+        public void TestDatabaseStoredProcedureSerializationDeserialization_WithDollarColumn()
         {
+            InitializeObjects(generateDollaredColumn: true);
+
+            TestTypeNameChanges(_databaseStoredProcedure, "DatabaseStoredProcedure");
+
+            Dictionary<string, DatabaseObject> dict = new();
+            dict.Add("person", _databaseStoredProcedure);
+
+            // Test to catch if there is change in number of properties/fields
+            // Note: On Addition of property make sure it is added in following object creation _databaseStoredProcedure and include in serialization
+            // and deserialization test.
+            int fields = typeof(DatabaseStoredProcedure).GetProperties(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Instance).Length;
+            Assert.AreEqual(fields, 6);
+
+            string serializedDatabaseSP = JsonSerializer.Serialize(dict, _options);
+            // Assert that the serialized JSON contains the escaped dollar sign in column name
+            Assert.IsTrue(serializedDatabaseSP.Contains("DAB_ESCAPE$FirstName"),
+                "Serialized JSON should contain the dollar-prefixed column name in SourceDefinition's Columns.");
+            Dictionary<string, DatabaseObject> deserializedDict = JsonSerializer.Deserialize<Dictionary<string, DatabaseObject>>(serializedDatabaseSP, _options)!;
+            DatabaseStoredProcedure deserializedDatabaseSP = (DatabaseStoredProcedure)deserializedDict["person"];
+
+            Assert.AreEqual(deserializedDatabaseSP.SourceType, _databaseStoredProcedure.SourceType);
+            deserializedDatabaseSP.Equals(_databaseStoredProcedure);
+            VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseSP.SourceDefinition, _databaseStoredProcedure.SourceDefinition, "$FirstName", true);
+            VerifySourceDefinitionSerializationDeserialization(deserializedDatabaseSP.StoredProcedureDefinition, _databaseStoredProcedure.StoredProcedureDefinition, "$FirstName", true);
+        }
+
+        private void InitializeObjects(bool generateDollaredColumn = false)
+        {
+            string columnName = generateDollaredColumn ? "$FirstName" : "FirstName";
             _options = new()
             {
                 // ObjectConverter behavior different in .NET8 most likely due to
@@ -290,10 +396,11 @@ private void InitializeObjects()
                     new DatabaseObjectConverter(),
                     new TypeConverter()
                 }
+
             };
 
             _columnDefinition = GetColumnDefinition(typeof(string), DbType.String, true, false, false, new string("John"), false);
-            _sourceDefinition = GetSourceDefinition(false, false, new List<string>() { "FirstName" }, _columnDefinition);
+            _sourceDefinition = GetSourceDefinition(false, false, new List<string>() { columnName }, _columnDefinition);
 
             _databaseTable = new DatabaseTable()
             {
@@ -312,10 +419,10 @@ private void InitializeObjects()
                 {
                     IsInsertDMLTriggerEnabled = false,
                     IsUpdateDMLTriggerEnabled = false,
-                    PrimaryKey = new List<string>() { "FirstName" },
+                    PrimaryKey = new List<string>() { columnName },
                 },
             };
-            _databaseView.ViewDefinition.Columns.Add("FirstName", _columnDefinition);
+            _databaseView.ViewDefinition.Columns.Add(columnName, _columnDefinition);
 
             _parameterDefinition = new()
             {
@@ -332,10 +439,10 @@ private void InitializeObjects()
                 SourceType = EntitySourceType.StoredProcedure,
                 StoredProcedureDefinition = new()
                 {
-                    PrimaryKey = new List<string>() { "FirstName" },
+                    PrimaryKey = new List<string>() { columnName },
                 }
             };
-            _databaseStoredProcedure.StoredProcedureDefinition.Columns.Add("FirstName", _columnDefinition);
+            _databaseStoredProcedure.StoredProcedureDefinition.Columns.Add(columnName, _columnDefinition);
             _databaseStoredProcedure.StoredProcedureDefinition.Parameters.Add("Id", _parameterDefinition);
         }