From 9063e5db857f065f3005df5b157d5e948814a545 Mon Sep 17 00:00:00 2001
From: nk <nkwork1234@gmail.com>
Date: Fri, 12 Jun 2026 00:09:37 +0900
Subject: [PATCH] Fix dbt source identifiers with dots

Signed-off-by: nk <nkwork1234@gmail.com>
---
 sqlmesh/dbt/source.py    |  5 ++++-
 tests/dbt/test_config.py | 26 ++++++++++++++++++++++++++
 2 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/sqlmesh/dbt/source.py b/sqlmesh/dbt/source.py
index 832ed0e156..b5c5893563 100644
--- a/sqlmesh/dbt/source.py
+++ b/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)
diff --git a/tests/dbt/test_config.py b/tests/dbt/test_config.py
index 9fb2a442a1..e67c68655b 100644
--- a/tests/dbt/test_config.py
+++ b/tests/dbt/test_config.py
@@ -521,6 +521,32 @@ def test_quoting():
     assert str(BaseRelation.create(**source.relation_info)) == 'foo."bar"'
 
 
+@pytest.mark.parametrize(
+    ("identifier", "expected"),
+    [
+        ("FILENAME.CSV", 'raw."FILENAME.CSV"'),
+        ("FILE NAME", 'raw."FILE NAME"'),
+    ],
+)
+def test_source_config_canonical_name_quotes_identifier_with_dot_or_space(
+    identifier: str, expected: str
+):
+    context = DbtContext()
+    context.project_name = "test"
+    context.target = DuckDbConfig(name="target", schema="raw")
+
+    source = SourceConfig(
+        name="my_table",
+        source_name="my_source",
+        schema="raw",
+        identifier=identifier,
+        quoting={"identifier": False},
+    )
+    context.sources = {source.config_name: source}
+
+    assert source.canonical_name(context) == expected
+
+
 def _test_warehouse_config(
     config_yaml: str, target_class: t.Type[TargetConfig], *params_path: str
 ) -> TargetConfig: