fix: match configured snapshot keys and cursors to quoted columns

benchmarks/duckshop/README.md

@@ -119,15 +119,15 @@ project and does not change it:
 - raw `edge__*` sources in `sources/edge_cases.yml`, loaded by `python/loaders/edge_cases.py` from
   fixed rows that `data_scale` does not change, so core data and core expectations stay the same.
 
-It has 38 models (7 staging, 22 intermediate, 9 marts), 62 unit tests and 4 scenarios, in four
+It has 39 models (7 staging, 22 intermediate, 10 marts), 63 unit tests and 4 scenarios, in four
 groups:
 
 | Group | Covers |
 |-------|--------|
 | Semi-structured | JSON null against SQL NULL and missing keys; document text with sorted, case-differing and escaped keys; JSON number fidelity and casts out of `VARIANT`; path-like keys (`''`, `*`, `$`-prefixed, dotted) with `GET` and `GET_IGNORE_CASE`; `OBJECT_CONSTRUCT_KEEP_NULL`, `OBJECT_INSERT`, `OBJECT_DELETE`, `OBJECT_PICK`; array builders, `ARRAY_EXCEPT` and number spellings in `ARRAY_POSITION`; `FLATTEN` outer; lenient and malformed `TRY_PARSE_JSON`; typed lambdas |
 | Dates and times | `TRY_TO_TIMESTAMP_NTZ` and its siblings over free text; epoch milliseconds as numbers and text; daylight-saving gaps and overlaps; fractional offsets; nanosecond truncation and epoch parts after 2242; pre-1970 text with sub-microsecond digits; date-part normalisation; month ends and leap days; week numbering across a year; sentinel timestamps |
 | Ordering, numbers and text | NULL ordering under `DESC`; ties in ordered aggregates, `MAX_BY`, `MODE` and `ANY_VALUE`; default window frames; `UNION ALL BY NAME`; `ROUND` half-way values, decimal division, `MOD` and 38-digit amounts; integers past 2^53 against `FLOAT`; regular expressions and backslash escapes; `LIKE ... ESCAPE` and `ILIKE ANY`; `SPLIT_PART` and `SUBSTR` before the start; `NOT IN` with a NULL; names in many scripts |
-| Scenarios and lifecycle | typed values across model boundaries; `VARIANT` expected output; a star under a lower-case contract and a quoted camelCase export; a cursor-window incremental model over far-apart fixture loads; a change log merged across two loads, checked against a full rebuild |
+| Scenarios and lifecycle | typed values across model boundaries; `VARIANT` expected output; a star under a lower-case contract and a quoted camelCase export; a snapshot keyed and compared on quoted camelCase columns; a cursor-window incremental model over far-apart fixture loads; a change log merged across two loads, checked against a full rebuild |
 
 Unit-test expectations in the edge-case area were captured on Snowflake under the `edge` target's
 pinned session, rather than derived by hand; scenario expectations, and the assertions in tests

benchmarks/duckshop/project/models/marts/edge_cases/snap_edge_partner_listing_export.sql

@@ -0,0 +1,31 @@
+MODEL (
+  description """
+    Version history of the partner feed export for listings with a known status, keyed and compared
+    on its quoted camelCase columns
+  """,
+  tags [edge_case],
+  materialized snapshot,
+  unique_key ['"listingId"'],
+  snapshot_strategy check,
+  check_columns ['"unitPrice"', '"isLive"'],
+  columns (
+    "listingId" (nullable false, audits [not_null]),
+  ),
+);
+
+WITH export AS (
+  SELECT *
+  FROM __ref("rpt_edge_partner_listing_export")
+),
+
+final AS (
+  SELECT
+    e."listingId",
+    e."unitPrice",
+    e."isLive"
+  FROM export AS e
+  WHERE e."isLive" IS NOT NULL
+)
+
+SELECT *
+FROM final

benchmarks/duckshop/project/tests/unit/marts/edge_cases/test_snap_edge_partner_listing_export.sql

@@ -0,0 +1,30 @@
+TEST (name "snap_edge_partner_listing_export__tracks_listings_with_a_known_status");
+
+WITH __ref__rpt_edge_partner_listing_export AS (
+  SELECT
+    'LS-EIDER-03' AS "listingId",
+    0.05::NUMBER(18, 2) AS "unitPrice",
+    CAST(NULL AS BOOLEAN) AS "isLive"
+  UNION ALL
+  SELECT
+    'LS-MALLARD-01' AS "listingId",
+    12.99::NUMBER(18, 2) AS "unitPrice",
+    TRUE AS "isLive"
+  UNION ALL
+  SELECT
+    'LS-TEAL-02' AS "listingId",
+    8.99::NUMBER(18, 2) AS "unitPrice",
+    FALSE AS "isLive"
+),
+
+__expected__snap_edge_partner_listing_export AS (
+  SELECT
+    'LS-MALLARD-01' AS "listingId",
+    12.99::NUMBER(18, 2) AS "unitPrice",
+    TRUE AS "isLive"
+  UNION ALL
+  SELECT
+    'LS-TEAL-02' AS "listingId",
+    8.99::NUMBER(18, 2) AS "unitPrice",
+    FALSE AS "isLive"
+)

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters/snowflake.md

@@ -10,6 +10,7 @@ Online: https://sqlbuild.com/docs/concepts/adapters/snowflake/
 
 - Connection config
 - Session initialization
+- Case-sensitive column names
 - Planning metadata
 - Build metadata
 - State tables
@@ -63,6 +64,29 @@ one warehouse for build-style commands and another for ad-hoc commands such as `
 `sqb diff`, and `--warehouse` overrides both for one invocation. See
 [command-group warehouses](../project-configuration.md#command-group-warehouses).
 
+## Case-sensitive column names
+
+Snowflake stores an unquoted column name in upper case, so `order_id` and `ORDER_ID` name the
+same column. A column created with a quoted name in mixed or lower case, such as `"orderId"`,
+keeps its exact case, and only the quoted name reads it.
+
+Model settings that name such a column (`unique_key`, `updated_at`, `observed_at`,
+`check_columns`, `cursor` and `merge_exclude_columns`) take the quoted name inside a string:
+
+```sql
+MODEL (
+  description 'Order history.',
+  materialized snapshot,
+  unique_key ['"orderId"'],
+  snapshot_strategy timestamp,
+  updated_at '"updatedAt"',
+);
+```
+
+A snapshot or microbatch build stops when an unquoted setting names a case-sensitive column, with
+an error such as
+`unique_key 'orderId' names the case-sensitive column "orderId"; write it with its quotes, as '"orderId"'`.
+
 ## Planning metadata
 
 Planning reads relation and column metadata with `SHOW TABLES`, `SHOW VIEWS`, and `SHOW COLUMNS`

src/sqlbuild/adapter/contract/classes/base_adapter.py

@@ -38,6 +38,7 @@
     AdapterUserError,
     UnsupportedTypedSqlRenderingError,
 )
+from sqlbuild.adapter.contract.main.column_name_key import column_name_key
 from sqlbuild.adapter.contract.main.run_relation_read_probe import run_relation_read_probe
 from sqlbuild.adapter.contract.main.same_view_definition import same_view_definition
 from sqlbuild.adapter.contract.main.unsupported_relation_usage import unsupported_relation_usage
@@ -969,7 +970,7 @@ def render_merge(
         exclude_columns: tuple[str, ...] = (),
     ) -> tuple[str, ...]:
         immutable_columns: frozenset[str] = frozenset(
-            column.lower() for column in (*unique_key, *exclude_columns)
+            column_name_key(column) for column in (*unique_key, *exclude_columns)
         )
         join_condition: str = " AND ".join(
             f"__target.{self.render_identifier(k)} = __source.{self.render_identifier(k)}"
@@ -978,7 +979,7 @@ def render_merge(
         update_assignments: str = ", ".join(
             f"{self.render_identifier(col)} = __source.{self.render_identifier(col)}"
             for col in source_columns
-            if col.lower() not in immutable_columns
+            if column_name_key(col) not in immutable_columns
         )
         insert_columns: str = ", ".join(self.render_identifier(col) for col in source_columns)
         insert_values: str = ", ".join(

src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py

@@ -49,6 +49,7 @@
     QUALIFIED_NAME_SEPARATOR,
 )
 from sqlbuild.adapter.contract.exceptions import AdapterUserError, ViewRenameRefusedError
+from sqlbuild.adapter.contract.main.column_name_key import column_name_key
 from sqlbuild.adapter.contract.main.is_quoted_identifier import is_quoted_identifier
 from sqlbuild.adapter.contract.main.run_relation_read_probe import run_relation_read_probe
 from sqlbuild.adapter.contract.main.same_view_definition import same_view_definition
@@ -2079,7 +2080,7 @@ def render_merge(
         exclude_columns: tuple[str, ...] = (),
     ) -> tuple[str, ...]:
         immutable_columns: frozenset[str] = frozenset(
-            column.lower() for column in (*unique_key, *exclude_columns)
+            column_name_key(column) for column in (*unique_key, *exclude_columns)
         )
         join_condition: str = " AND ".join(
             f"__target.{self.render_identifier(k)} = __source.{self.render_identifier(k)}"
@@ -2088,7 +2089,7 @@ def render_merge(
         update_assignments: str = ", ".join(
             f"{self.render_identifier(col)} = __source.{self.render_identifier(col)}"
             for col in source_columns
-            if col.lower() not in immutable_columns
+            if column_name_key(col) not in immutable_columns
         )
         insert_columns: str = ", ".join(self.render_identifier(col) for col in source_columns)
         insert_values: str = ", ".join(

src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py

@@ -47,6 +47,7 @@
     DIFF_RIGHT_SIDE,
 )
 from sqlbuild.adapter.contract.exceptions import AdapterUserError
+from sqlbuild.adapter.contract.main.column_name_key import column_name_key
 from sqlbuild.adapter.contract.main.complete_table_freshness_results import (
     complete_table_freshness_results,
 )
@@ -1220,7 +1221,7 @@ def render_merge(
         exclude_columns: tuple[str, ...] = (),
     ) -> tuple[str, ...]:
         immutable_columns: frozenset[str] = frozenset(
-            column.lower() for column in (*unique_key, *exclude_columns)
+            column_name_key(column) for column in (*unique_key, *exclude_columns)
         )
         join_condition: str = " AND ".join(
             f"__target.{self.render_identifier(k)} = __source.{self.render_identifier(k)}"
@@ -1229,7 +1230,7 @@ def render_merge(
         update_assignments: str = ", ".join(
             f"{self.render_identifier(col)} = __source.{self.render_identifier(col)}"
             for col in source_columns
-            if col.lower() not in immutable_columns
+            if column_name_key(col) not in immutable_columns
         )
         insert_columns: str = ", ".join(self.render_identifier(col) for col in source_columns)
         insert_values: str = ", ".join(

src/sqlbuild/executor/run/_helpers/materializations/microbatch.py

@@ -58,6 +58,7 @@
 from sqlbuild._native import cursor_split_into_batches as split_into_batches
 from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
 from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder
+from sqlbuild.adapter.contract.main.column_name_key import column_name_key
 from sqlbuild.adapter.contract.models import ColumnInfo
 from sqlbuild.adapter.relations.main.cached_relation_exists import cached_relation_exists
 from sqlbuild.adapter.relations.main.resolve_qualified_name_parts import (
@@ -97,6 +98,7 @@
     _widenable_columns,
 )
 from sqlbuild.executor.run._helpers.reuse.fingerprinting import try_write_fingerprint
+from sqlbuild.executor.run._helpers.validation.configured_columns import require_quoted_spelling
 from sqlbuild.executor.run._helpers.validation.cursor_bounds import (
     build_runtime_cursor_spec,
     empty_input_rebuild_error,
@@ -3531,8 +3533,16 @@ def _validate_cursor_output_columns(
     cursor_column: str | None = entry.cursor_column
     if cursor_column is None:
         return
-    delta_names: frozenset[str] = frozenset(column.name.lower() for column in delta_columns)
-    if cursor_column.lower() in delta_names:
+    delta_names: frozenset[str] = frozenset(
+        column_name_key(column.name) for column in delta_columns
+    )
+    if column_name_key(cursor_column) in delta_names:
+        require_quoted_spelling(
+            model_name=entry.name,
+            setting="cursor",
+            configured=(cursor_column,),
+            listed_columns=delta_columns,
+        )
         return
     raise ExecutorInputError(
         f"microbatch cursor column '{cursor_column}' is not produced by model output; "

src/sqlbuild/executor/run/_helpers/materializations/snapshot.py

@@ -18,6 +18,7 @@
 )
 from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
 from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder
+from sqlbuild.adapter.contract.main.column_name_key import column_name_key
 from sqlbuild.adapter.contract.models import (
     ColumnInfo,
     NormalizedType,
@@ -56,6 +57,7 @@
     resolve_full_refresh_relations,
 )
 from sqlbuild.executor.run._helpers.reuse.fingerprinting import try_write_fingerprint
+from sqlbuild.executor.run._helpers.validation.configured_columns import require_quoted_spelling
 from sqlbuild.executor.run._helpers.validation.contracts import validate_runtime_contract
 from sqlbuild.executor.run.constants import (
     SNAPSHOT_ALL_CHECK_COLUMNS,
@@ -634,7 +636,9 @@ def _validate_delta_columns(
     delta_columns: tuple[ColumnInfo, ...],
     check_columns: tuple[str, ...],
 ) -> None:
-    column_names: frozenset[str] = frozenset(column.name.lower() for column in delta_columns)
+    column_names: frozenset[str] = frozenset(
+        column_name_key(column.name) for column in delta_columns
+    )
     required_columns: tuple[str, ...]
     if entry.snapshot_strategy == SnapshotStrategy.TIMESTAMP:
         required_columns = (*entry.unique_key, _require_updated_at(entry))
@@ -654,17 +658,35 @@ def _validate_delta_columns(
     ):
         required_columns = (*required_columns, entry.observed_at_column)
     missing_columns: list[str] = [
-        column for column in required_columns if column.lower() not in column_names
+        column for column in required_columns if column_name_key(column) not in column_names
     ]
     if missing_columns:
         raise ExecutorInputError(
             f"snapshot model '{entry.name}' query output is missing required columns: "
             f"{', '.join(missing_columns)}"
         )
+    setting: str
+    configured: tuple[str | None, ...]
+    for setting, configured in (
+        ("unique_key", entry.unique_key),
+        ("updated_at", (entry.updated_at_column,)),
+        ("observed_at", (entry.observed_at_column,)),
+        ("check_columns", check_columns),
+    ):
+        require_quoted_spelling(
+            model_name=entry.name,
+            setting=setting,
+            configured=tuple(
+                name for name in configured if name is not None and name in required_columns
+            ),
+            listed_columns=delta_columns,
+        )
     valid_from_column: str = _valid_from_column(entry)
     valid_to_column: str = _valid_to_column(entry)
     generated_collisions: list[str] = [
-        column for column in (valid_from_column, valid_to_column) if column.lower() in column_names
+        column
+        for column in (valid_from_column, valid_to_column)
+        if column_name_key(column) in column_names
     ]
     if generated_collisions:
         raise ExecutorInputError(
@@ -941,16 +963,18 @@ def _expanded_check_columns(
     ):
         return entry.check_columns
 
-    excluded_names: set[str] = {column.lower() for column in entry.unique_key}
-    excluded_names.add(_valid_from_column(entry).lower())
-    excluded_names.add(_valid_to_column(entry).lower())
+    excluded_names: set[str] = {column_name_key(column) for column in entry.unique_key}
+    excluded_names.add(column_name_key(_valid_from_column(entry)))
+    excluded_names.add(column_name_key(_valid_to_column(entry)))
     if entry.observed_at_column is not None:
-        excluded_names.add(entry.observed_at_column.lower())
+        excluded_names.add(column_name_key(entry.observed_at_column))
     if entry.updated_at_column is not None:
-        excluded_names.add(entry.updated_at_column.lower())
+        excluded_names.add(column_name_key(entry.updated_at_column))
 
     expanded_columns: tuple[str, ...] = tuple(
-        column.name for column in delta_columns if column.name.lower() not in excluded_names
+        column.name
+        for column in delta_columns
+        if column_name_key(column.name) not in excluded_names
     )
     if not expanded_columns:
         raise ExecutorInputError(

src/sqlbuild/executor/run/_helpers/validation/configured_columns.py

@@ -0,0 +1,37 @@
+"""Runtime validation of the column names a model's configuration names."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from sqlbuild.adapter.contract.main.column_name_key import column_name_key
+from sqlbuild.adapter.contract.main.is_quoted_identifier import is_quoted_identifier
+from sqlbuild.adapter.contract.models import ColumnInfo
+from sqlbuild.errors.contracts.exceptions import ExecutorInputError
+
+
+def require_quoted_spelling(
+    *,
+    model_name: str,
+    setting: str,
+    configured: Iterable[str],
+    listed_columns: tuple[ColumnInfo, ...],
+) -> None:
+    """Reject an unquoted configured name that only a quoted, case-sensitive column matches."""
+
+    spellings: dict[str, list[str]] = {}
+    column: ColumnInfo
+    for column in listed_columns:
+        spellings.setdefault(column_name_key(column.name), []).append(column.name)
+    name: str
+    for name in configured:
+        if is_quoted_identifier(name):
+            continue
+        matches: list[str] = spellings.get(column_name_key(name), [])
+        if not matches or not all(is_quoted_identifier(match) for match in matches):
+            continue
+        raise ExecutorInputError(
+            f"model '{model_name}' {setting} '{name}' names the case-sensitive column "
+            f"{matches[0]}; write it with its quotes, as '{matches[0]}', so the generated SQL "
+            "keeps its case"
+        )

tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py

@@ -1513,3 +1513,14 @@ class QuotedContractE2ETestCase:
     description: str
     declared_name: str
     expected_rows: list[tuple[object, ...]]
+
+
+@dataclass(frozen=True)
+class QuotedConfiguredColumnsE2ETestCase:
+    """A snapshot key and a microbatch cursor configured with quoted camelCase names."""
+
+    description: str
+    unique_key: str
+    cursor: str
+    expected_history_rows: list[tuple[object, ...]]
+    expected_activity_rows: list[tuple[object, ...]]