fix(dbt): preserve source identifier quoting when it contains dots or spaces

Author: jnallaliveaction

sqlmesh/dbt/source.py

@@ -82,10 +82,13 @@ def canonical_name(self, context: DbtContext) -> str:
                     f"'source' macro failed for '{self.config_name}' with exception '{e}'."
                 )
 
+            needs_identifier_quoting = bool(
+                self.table_name and ("." in self.table_name or " " in self.table_name)
+            )
             relation = relation.quote(
                 database=False,
                 schema=False,
-                identifier=False,
+                identifier=needs_identifier_quoting,
             )
             if relation.database == context.target.database:
                 relation = relation.include(database=False)

tests/dbt/test_config.py

@@ -521,6 +521,71 @@ def test_quoting():
     assert str(BaseRelation.create(**source.relation_info)) == 'foo."bar"'
 
 
+def test_source_canonical_name_with_dots_and_spaces(mocker):
+    from dbt.adapters.base import BaseRelation
+
+    mock_context = mocker.Mock()
+    mock_context.target.database = "target_db"
+
+    def mock_source_macro(source_name, table_name):
+        if table_name == "my_table_dot":
+            identifier = "FILENAME.CSV"
+        elif table_name == "my_table_space":
+            identifier = "my table space"
+        else:
+            identifier = "my_table_std"
+        return BaseRelation.create(
+            database="RAW_DEV",
+            schema="raw_schema",
+            identifier=identifier,
+        )
+
+    mock_context.get_callable_macro.return_value = mock_source_macro
+
+    # 1. Identifier with a dot
+    source_dot = SourceConfig(
+        name="my_table_dot",
+        source_name="my_source",
+        identifier="FILENAME.CSV",
+    )
+    assert source_dot.canonical_name(mock_context) == 'RAW_DEV.raw_schema."FILENAME.CSV"'
+
+    # 2. Identifier with a space
+    source_space = SourceConfig(
+        name="my_table_space",
+        source_name="my_source",
+        identifier="my table space",
+    )
+    assert source_space.canonical_name(mock_context) == 'RAW_DEV.raw_schema."my table space"'
+
+    # 3. Standard identifier (without dots or spaces) should not be quoted
+    source_std = SourceConfig(
+        name="my_table_std",
+        source_name="my_source",
+        identifier="my_table_std",
+    )
+    assert source_std.canonical_name(mock_context) == "RAW_DEV.raw_schema.my_table_std"
+
+    # 4. Standard identifier, but with database matching target database (to test database omission)
+    mock_context_target_db = mocker.Mock()
+    mock_context_target_db.target.database = "RAW_DEV"
+    mock_context_target_db.get_callable_macro.return_value = mock_source_macro
+
+    source_dot_target = SourceConfig(
+        name="my_table_dot",
+        source_name="my_source",
+        identifier="FILENAME.CSV",
+    )
+    source_std_target = SourceConfig(
+        name="my_table_std",
+        source_name="my_source",
+        identifier="my_table_std",
+    )
+
+    assert source_dot_target.canonical_name(mock_context_target_db) == 'raw_schema."FILENAME.CSV"'
+    assert source_std_target.canonical_name(mock_context_target_db) == "raw_schema.my_table_std"
+
+
 def _test_warehouse_config(
     config_yaml: str, target_class: t.Type[TargetConfig], *params_path: str
 ) -> TargetConfig: