Fix(mssql): Properly quote table and views when dropping in mssql (#4773)

Author: themisvaltinos

sqlmesh/core/engine_adapter/mssql.py

@@ -172,15 +172,18 @@ def drop_schema(
         if cascade:
             objects = self._get_data_objects(schema_name)
             for obj in objects:
+                # Build properly quoted table for MSSQL using square brackets when needed
+                object_table = exp.table_(obj.name, obj.schema_name)
+
                 # _get_data_objects is catalog-specific, so these can't accidentally drop view/tables in another catalog
                 if obj.type == DataObjectType.VIEW:
                     self.drop_view(
-                        ".".join([obj.schema_name, obj.name]),
+                        object_table,
                         ignore_if_not_exists=ignore_if_not_exists,
                     )
                 else:
                     self.drop_table(
-                        ".".join([obj.schema_name, obj.name]),
+                        object_table,
                         exists=ignore_if_not_exists,
                     )
         super().drop_schema(schema_name, ignore_if_not_exists=ignore_if_not_exists, cascade=False)

tests/core/engine_adapter/test_mssql.py

@@ -655,6 +655,36 @@ def test_drop_schema(make_mocked_engine_adapter: t.Callable):
     ]
 
 
+def test_drop_schema_with_special_identifiers(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(MSSQLEngineAdapter)
+
+    adapter._get_data_objects = mock.Mock()
+    adapter._get_data_objects.return_value = [
+        DataObject(
+            catalog="test_catalog",
+            schema="test schema",  # Schema with space
+            name="test view",  # Object with space
+            type=DataObjectType.from_str("VIEW"),
+        ),
+        DataObject(
+            catalog="test_catalog",
+            schema="test schema",
+            name="test table",  # Table with space
+            type=DataObjectType.from_str("TABLE"),
+        ),
+    ]
+
+    schema_name = exp.to_table("[test schema]", dialect="tsql")
+    adapter.drop_schema(schema_name, cascade=True)
+
+    # Validate that names with spaces/special chars are properly quoted with square brackets
+    assert to_sql_calls(adapter) == [
+        """DROP VIEW IF EXISTS [test schema].[test view];""",
+        """DROP TABLE IF EXISTS [test schema].[test table];""",
+        """DROP SCHEMA IF EXISTS [test schema];""",
+    ]
+
+
 def test_df_dates(make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(MSSQLEngineAdapter)