fix(schema_loader): quote case-sensitive column names in external models

create_external_models wrote column names from the adapter verbatim, so
on case-folding dialects like postgres a quoted column such as "ID" was
serialized as the unquoted key ID. On load, normalize_identifiers folded
it to id and the generated external_models.yaml no longer described the
table, breaking plan and lint with ambiguousorinvalidcolumn errors.

Only quote identifiers whose unquoted form would be rewritten by the
dialect's normalization; already-quoted names and identifiers that
normalize to themselves (including pseudo columns like _sync_row_hash)
are written unchanged.

Signed-off-by: devtechedge <devtechedge@gmail.com>

Author: devtechedge

sqlmesh/core/schema_loader.py

@@ -6,6 +6,7 @@
 
 from sqlglot import exp
 from sqlglot.dialects.dialect import DialectType
+from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 
 from sqlmesh.core.console import get_console
 from sqlmesh.core.engine_adapter import EngineAdapter
@@ -90,6 +91,16 @@ def create_external_models_file(
             yaml.dump(entries_to_keep + schemas, file)
 
 
+def _serialize_column_name(name: str, dialect: DialectType) -> str:
+    if name.startswith('"') and name.endswith('"'):
+        return name
+    return (
+        name
+        if normalize_identifiers(exp.to_column(name), dialect=dialect).name == name
+        else f'"{name}"'
+    )
+
+
 def get_columns(
     adapter: EngineAdapter, dialect: DialectType, table: str, strict: bool
 ) -> t.Optional[t.Dict[str, t.Any]]:
@@ -98,7 +109,10 @@ def get_columns(
     """
     try:
         columns = adapter.columns(table, include_pseudo_columns=True)
-        return {c: dtype.sql(dialect=dialect) for c, dtype in columns.items()}
+        return {
+            _serialize_column_name(c, dialect): dtype.sql(dialect=dialect)
+            for c, dtype in columns.items()
+        }
     except Exception as e:
         msg = f"Unable to get schema for '{table}': '{e}'."
         if strict:

tests/core/test_schema_loader.py

@@ -389,6 +389,54 @@ def test_no_internal_model_conversion(tmp_path: Path, mocker: MockerFixture):
         create_external_model(**row, dialect="bigquery")
 
 
+def test_create_external_models_quotes_case_sensitive_columns(
+    tmp_path: Path, mocker: MockerFixture
+):
+    engine_adapter_mock = mocker.Mock()
+    engine_adapter_mock.columns.return_value = {
+        "ID": exp.DataType.build("bigint"),
+        "ORGANID": exp.DataType.build("text"),
+        "CDATE": exp.DataType.build("date"),
+        "billno": exp.DataType.build("text"),
+        "_sync_row_hash": exp.DataType.build("text"),
+    }
+
+    state_reader_mock = mocker.Mock()
+    state_reader_mock.nodes_exist.return_value = set()
+
+    model_a = SqlModel(name="a", query=parse_one("select * FROM raw_fruits"))
+
+    filename = tmp_path / c.EXTERNAL_MODELS_YAML
+    create_external_models_file(
+        filename,
+        {"a": model_a},  # type: ignore
+        engine_adapter_mock,
+        state_reader_mock,
+        "postgres",
+    )
+
+    schema = yaml.load(filename)
+    assert len(schema) == 1
+    # only identifiers that would be case-folded by the dialect are quoted
+    assert list(schema[0]["columns"]) == [
+        '"ID"',
+        '"ORGANID"',
+        '"CDATE"',
+        "billno",
+        "_sync_row_hash",
+    ]
+
+    # the quoted keys must round-trip through the model loader without being folded to lowercase
+    external_model = create_external_model(**schema[0], dialect="postgres")
+    assert list(external_model.columns_to_types) == [
+        "ID",
+        "ORGANID",
+        "CDATE",
+        "billno",
+        "_sync_row_hash",
+    ]
+
+
 def test_missing_table(tmp_path: Path):
     config = Config(gateways=GatewayConfig(connection=DuckDBConnectionConfig()))
     context = Context(paths=[str(tmp_path.absolute())], config=config)