table_diff: KeyError when `-o` key columns are not lower case on Bigquery

[Anant-gif]

sqlmesh table_diff fails with a KeyError when they columns passed via -o exist in the table with non lower case names, on engines whose sqlglot dialect lowercases identifiers (BigQuery, DuckDB)

The values passed to -o are run through normalize_identifiers with the connection dialect which lowercase them (quoted or not). On BigQuery, so -o KEY becomes key. The column names reported by the engine via adapter.columns(...) keep their stored casing , so the schema dict is keyed by KEY. Subsequent exact-string dictionary lookups in Python miss the match and raise a KeyError before any query is executed.

Reproduction

• Run table diff using sqlmesh table_diff CLI on any tables whose key columns are not lowercase in BigQuery (any table created with dialect bigquery,normaliztion_strategy=case_sensitive).

• The error occurs for both single and multiple keys. It can also be reproduced using duckdb:

import pandas as pd
from sqlglot import exp
from sqlmesh.core.config.connection import DuckDBConnectionConfig
from sqlmesh.core.table_diff import TableDiff

adapter = DuckDBConnectionConfig().create_engine_adapter()
cols = {"KEY1": exp.DataType.build("int"), "KEY2": exp.DataType.build("int"), "VALUE": exp.DataType.build("varchar")}
adapter.create_table("src", cols)
adapter.create_table("target", cols)
df = pd.DataFrame([(1, 2, "a")], columns=cols.keys())
adapter.insert_append("src", df)
adapter.insert_append("target", df)

TableDiff(adapter=adapter, source="src", target="target", on=["KEY1", "KEY2"]).row_diff()

Output:

Traceback (most recent call last):
  File "/home/anant/Documents/sqlmesh/test.py", line 14, in <module>
    TableDiff(adapter=adapter, source="src", target="target", on=["KEY1", "KEY2"]).row_diff()
  File "/home/anant/Documents/sqlmesh/sqlmesh/core/table_diff.py", line 397, in row_diff
    self.source_key_expression.as_(SQLMESH_JOIN_KEY_COL),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/anant/Documents/sqlmesh/sqlmesh/core/table_diff.py", line 311, in source_key_expression
    return self._key_expression(s_index, self.source_schema)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/anant/Documents/sqlmesh/sqlmesh/core/table_diff.py", line 326, in _key_expression
    key_columns_to_types = {key.name: schema[key.name] for key in cols}
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/anant/Documents/sqlmesh/sqlmesh/core/table_diff.py", line 326, in <dictcomp>
    key_columns_to_types = {key.name: schema[key.name] for key in cols}
                                      ~~~~~~^^^^^^^^^^
KeyError: 'key1'

Expected behavior and Proposal

There could be two ways of solving this problem:

1. On engines like BigQuery and DuckDB (which are case-insensitive), the casing passed in the keys should not matter. The table_diff command should work for both lowercase and uppercase names by performing a case-insensitive comparison. But this could lead to unexpected behavior in engines like Postgres.
2. Instead of directly comparing normalized column name with the adapter returned columns, attempt an exact match first. If that fails, fall back to case insensitive matching.

Environment

• SQLMesh : Reproduced on the latest version as well as 0.228.2
• Engine: BigQuery (also reproduces on DuckDB)
• Python: 3.12

Addition context

• --skip-columns uses the same normalize-then-exact-match pattern against the engine-reported schema.