Fix: external model schema was using generic model name the key. sqlg…

…lot is lenient enough to accept double quotes instead of backticks for dialects like bigquery, but this makes the schema.yaml more consistent (TobikoData#1658)

sqlmesh/core/model/definition.py

@@ -1797,7 +1797,9 @@ def _create_model(
                 "jinja_macros": jinja_macros,
                 "dialect": dialect,
                 "depends_on": depends_on,
-                "physical_schema_override": physical_schema_override.get(exp.to_table(name).db),
+                "physical_schema_override": physical_schema_override.get(
+                    exp.to_table(name, dialect=dialect).db
+                ),
                 **kwargs,
             },
         )

sqlmesh/core/schema_loader.py

@@ -5,6 +5,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
+from sqlglot import exp
 from sqlglot.dialects.dialect import DialectType
 
 from sqlmesh.core.engine_adapter import EngineAdapter
@@ -63,7 +64,7 @@ def _get_columns(table: str) -> t.Optional[t.Dict[str, t.Any]]:
 
         schemas = [
             {
-                "name": table,
+                "name": exp.to_table(table).sql(dialect=dialect),
                 "columns": {c: t.sql(dialect=dialect) for c, t in columns.items()},
             }
             for table, columns in sorted(

tests/core/test_schema_loader.py

@@ -10,7 +10,7 @@
 from sqlmesh.core.config import Config, DuckDBConnectionConfig, GatewayConfig
 from sqlmesh.core.context import Context
 from sqlmesh.core.dialect import parse
-from sqlmesh.core.model import SqlModel, load_sql_based_model
+from sqlmesh.core.model import SqlModel, create_external_model, load_sql_based_model
 from sqlmesh.core.schema_loader import create_schema_file
 from sqlmesh.core.snapshot import SnapshotChangeCategory
 from sqlmesh.utils.yaml import YAML
@@ -108,17 +108,32 @@ def test_no_internal_model_conversion(tmp_path: Path, mocker: MockerFixture):
     state_reader_mock = mocker.Mock()
     state_reader_mock.nodes_exist.return_value = {"model_b"}
 
-    model = SqlModel(name="a", query=parse_one("select * FROM model_b, tbl_c"))
+    model_a = SqlModel(name="a", query=parse_one("select * FROM model_b, tbl_c"))
+    model_b = SqlModel(name="b", query=parse_one("select * FROM `tbl-d`", read="bigquery"))
 
     schema_file = tmp_path / c.SCHEMA_YAML
-    create_schema_file(schema_file, {"a": model}, engine_adapter_mock, state_reader_mock, "")
+    create_schema_file(
+        schema_file,
+        {
+            "a": model_a,
+            "b": model_b,
+        },
+        engine_adapter_mock,
+        state_reader_mock,
+        "bigquery",
+    )
 
     with open(schema_file, "r") as fd:
         schema = YAML().load(fd)
 
-    assert len(schema) == 1
-    assert schema[0]["name"] == "tbl_c"
+    assert len(schema) == 2
+    assert schema[0]["name"] == "`tbl-d`"
     assert list(schema[0]["columns"]) == ["b", "a"]
+    assert schema[1]["name"] == "tbl_c"
+    assert list(schema[1]["columns"]) == ["b", "a"]
+
+    for row in schema:
+        create_external_model(**row, dialect="bigquery")
 
 
 def test_missing_table(tmp_path: Path):