Fix serialization object mutation and implement double-encoding escaping mechanism

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;
             }
         }
@@ -48,20 +62,161 @@ public override void Write(Utf8JsonWriter writer, DatabaseObject value, JsonSeri
             // "TypeName": "Azure.DataApiBuilder.Config.DatabasePrimitives.DatabaseTable, Azure.DataApiBuilder.Config",
             writer.WriteString(TYPE_NAME, GetTypeNameFromType(value.GetType()));
 
-            // Add other properties of DatabaseObject
-            foreach (PropertyInfo prop in value.GetType().GetProperties())
+            // Track SourceDefinition objects and their original Columns dictionaries
+            // to restore them after serialization (avoiding permanent mutation)
+            Dictionary<SourceDefinition, Dictionary<string, ColumnDefinition>> originalColumnsDictionaries = new();
+
+            try
             {
-                // Skip the TypeName property, as it has been handled above
-                if (prop.Name == TYPE_NAME)
+                // Add other properties of DatabaseObject
+                foreach (PropertyInfo prop in value.GetType().GetProperties())
                 {
-                    continue;
+                    // Skip the TypeName property, as it has been handled above
+                    if (prop.Name == TYPE_NAME)
+                    {
+                        continue;
+                    }
+
+                    writer.WritePropertyName(prop.Name);
+                    object? propVal = prop.GetValue(value);
+
+                    // Only escape columns for properties whose type(derived type) is SourceDefinition.
+                    // Temporarily replace dictionary contents with escaped version, then restore after serialization.
+                    if (IsSourceDefinitionOrDerivedClassProperty(prop) && propVal is SourceDefinition sourceDef)
+                    {
+                        if (!originalColumnsDictionaries.ContainsKey(sourceDef))
+                        {
+                            // Store original dictionary contents
+                            originalColumnsDictionaries[sourceDef] = new Dictionary<string, ColumnDefinition>(sourceDef.Columns, sourceDef.Columns.Comparer);
+
+                            // Replace dictionary contents with escaped version
+                            Dictionary<string, ColumnDefinition> escapedColumns = CreateEscapedColumnsDictionary(sourceDef.Columns);
+                            sourceDef.Columns.Clear();
+                            foreach (var kvp in escapedColumns)
+                            {
+                                sourceDef.Columns[kvp.Key] = kvp.Value;
+                            }
+                        }
+                    }
+
+                    JsonSerializer.Serialize(writer, propVal, options);
                 }
 
-                writer.WritePropertyName(prop.Name);
-                JsonSerializer.Serialize(writer, prop.GetValue(value), options);
+                writer.WriteEndObject();
+            }
+            finally
+            {
+                // Always restore original Columns dictionary contents to avoid mutating the input object
+                foreach (var kvp in originalColumnsDictionaries)
+                {
+                    kvp.Key.Columns.Clear();
+                    foreach (var colKvp in kvp.Value)
+                    {
+                        kvp.Key.Columns[colKvp.Key] = colKvp.Value;
+                    }
+                }
+            }
+        }
+
+        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>
+        /// Creates a new dictionary with escaped column keys for serialization.
+        /// Uses a double-encoding approach to handle edge cases:
+        /// 1. First escapes columns starting with 'DAB_ESCAPE$' to 'DAB_ESCAPE$DAB_ESCAPE$...'
+        /// 2. Then escapes columns starting with '$' to 'DAB_ESCAPE$...'
+        /// This ensures that even if a column is named 'DAB_ESCAPE$xyz', it will be properly handled.
+        /// Returns a new dictionary without modifying the original.
+        /// </summary>
+        private static Dictionary<string, ColumnDefinition> CreateEscapedColumnsDictionary(Dictionary<string, ColumnDefinition> originalColumns)
+        {
+            if (originalColumns is null || originalColumns.Count == 0)
+            {
+                return new Dictionary<string, ColumnDefinition>(StringComparer.InvariantCultureIgnoreCase);
             }
 
-            writer.WriteEndObject();
+            // Create a new dictionary with the same comparer as the original
+            Dictionary<string, ColumnDefinition> escapedColumns = new(originalColumns.Comparer);
+
+            // Step 1: Escape columns that start with the escape sequence itself
+            // This prevents collision when a column name already contains 'DAB_ESCAPE$'
+            foreach (var kvp in originalColumns)
+            {
+                string key = kvp.Key;
+                ColumnDefinition value = kvp.Value;
+
+                if (key.StartsWith(ESCAPED_DOLLARCHAR, StringComparison.Ordinal))
+                {
+                    // Double-escape: DAB_ESCAPE$FirstName -> DAB_ESCAPE$DAB_ESCAPE$FirstName
+                    escapedColumns[ESCAPED_DOLLARCHAR + key] = value;
+                }
+                else if (key.StartsWith(DOLLAR_CHAR, StringComparison.Ordinal))
+                {
+                    // Escape dollar: $FirstName -> DAB_ESCAPE$FirstName
+                    escapedColumns[ESCAPED_DOLLARCHAR + key.Substring(1)] = value;
+                }
+                else
+                {
+                    // No escaping needed
+                    escapedColumns[key] = value;
+                }
+            }
+
+            return escapedColumns;
+        }
+
+        /// <summary>
+        /// Unescapes column keys for deserialization using reverse double-encoding:
+        /// 1. First unescapes columns starting with 'DAB_ESCAPE$DAB_ESCAPE$' to 'DAB_ESCAPE$...'
+        /// 2. Then unescapes columns starting with 'DAB_ESCAPE$' to '$...'
+        /// This ensures proper reconstruction of original column names even in edge cases.
+        /// Modifies the dictionary in-place during deserialization.
+        /// </summary>
+        private static void UnescapeDollaredColumns(SourceDefinition sourceDef)
+        {
+            if (sourceDef.Columns is null || sourceDef.Columns.Count == 0)
+            {
+                return;
+            }
+
+            // Create a new dictionary with unescaped keys
+            Dictionary<string, ColumnDefinition> unescapedColumns = new(sourceDef.Columns.Comparer);
+
+            foreach (var kvp in sourceDef.Columns)
+            {
+                string key = kvp.Key;
+                ColumnDefinition value = kvp.Value;
+
+                // Step 1: Unescape columns that were double-escaped (originally started with 'DAB_ESCAPE$')
+                string doubleEscapeSequence = ESCAPED_DOLLARCHAR + ESCAPED_DOLLARCHAR;
+                if (key.StartsWith(doubleEscapeSequence, StringComparison.Ordinal))
+                {
+                    // Remove the first 'DAB_ESCAPE$' prefix: DAB_ESCAPE$DAB_ESCAPE$FirstName -> DAB_ESCAPE$FirstName
+                    unescapedColumns[key.Substring(ESCAPED_DOLLARCHAR.Length)] = value;
+                }
+                // Step 2: Unescape columns that start with 'DAB_ESCAPE$' (originally started with '$')
+                else if (key.StartsWith(ESCAPED_DOLLARCHAR, StringComparison.Ordinal))
+                {
+                    // Add back the '$' prefix: DAB_ESCAPE$FirstName -> $FirstName
+                    unescapedColumns[DOLLAR_CHAR + key.Substring(ESCAPED_DOLLARCHAR.Length)] = value;
+                }
+                else
+                {
+                    // No unescaping needed
+                    unescapedColumns[key] = value;
+                }
+            }
+
+            // Replace the dictionary contents with the unescaped version
+            sourceDef.Columns.Clear();
+            foreach (var kvp in unescapedColumns)
+            {
+                sourceDef.Columns[kvp.Key] = kvp.Value;
+            }
         }
 
         private static Type GetTypeFromName(string typeName)