fix(wren): honor SQL identifier case across policy, extract, and CTE rewriter (#2310)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

core/wren/src/wren/engine.py

@@ -33,7 +33,7 @@ from wren.mdl import get_manifest_extractor, get_session_context, to_json_base64
 from wren.mdl.cte_rewriter import CTERewriter, get_sqlglot_dialect
 from wren.model.data_source import DataSource
 from wren.model.error import DIALECT_SQL, ErrorCode, ErrorPhase, WrenError
-from wren.policy import validate_sql_policy
+from wren.policy import resolve_model_name, validate_sql_policy
 
 
 class WrenEngine:
@@ -171,13 +171,27 @@ class WrenEngine:
             dialect = get_sqlglot_dialect(self.data_source)
             ast = parse_one(sql, dialect=dialect)
 
+            manifest_json = json.loads(base64.b64decode(self.manifest_str))
+            model_names = {m["name"] for m in manifest_json.get("models", [])}
+
             # Policy validation: check tables and functions before execution.
             if self._config.strict_mode or self._config.denied_functions:
-                manifest_json = json.loads(base64.b64decode(self.manifest_str))
-                model_names = {m["name"] for m in manifest_json.get("models", [])}
                 validate_sql_policy(ast, model_names, self._config)
 
-            tables = [t.name for t in ast.find_all(exp.Table)]
+            # Resolve table refs to canonical manifest model names so that
+            # ``extract_by`` (case-sensitive in Rust) finds them under SQL's
+            # case-sensitivity rules: quoted identifiers match exactly,
+            # unquoted fall back to a case-insensitive scan.
+            tables: list[str] = []
+            for t in ast.find_all(exp.Table):
+                if not t.name:
+                    continue
+                quoted = (
+                    bool(t.this.quoted) if isinstance(t.this, exp.Identifier) else False
+                )
+                resolved = resolve_model_name(t.name, quoted, model_names)
+                tables.append(resolved if resolved is not None else t.name)
+
             extractor = get_manifest_extractor(self.manifest_str)
             manifest = extractor.extract_by(tables)
             effective_manifest = to_json_base64(manifest)

core/wren/src/wren/mdl/cte_rewriter.py

@@ -21,6 +21,7 @@ from sqlglot.schema import MappingSchema
 # Ensure the Wren dialect is registered with sqlglot on import.
 import wren.mdl.wren_dialect as _wren_dialect  # noqa: F401
 from wren.model.data_source import DataSource
+from wren.policy import resolve_model_name
 
 _SQLGLOT_DIALECT_MAP: dict[DataSource, str] = {
     DataSource.canner: "trino",
@@ -65,6 +66,7 @@ class CTERewriter:
         fallback: bool = True,
     ):
         self.session_context = session_context
+        self.data_source = data_source
         self.fallback = fallback
         self.dialect = get_sqlglot_dialect(data_source)
         self.manifest = json.loads(base64.b64decode(manifest_str))
@@ -87,7 +89,15 @@ class CTERewriter:
                 col_name = col["name"]
                 cols[col_name] = col.get("type", "TEXT")
                 orig[col_name.lower()] = col_name
-            self.schema.add_table(name, cols, dialect=self.dialect)
+            # ``qualify_columns`` runs against the post-``normalize_identifiers``
+            # AST, so the schema must be keyed under the same normalized form
+            # of the model name. BigQuery / DuckDB lowercase, Oracle uppercases —
+            # registering the literal manifest name leaves a mismatch and the
+            # column qualification silently produces an empty CTE body.
+            schema_name = normalize_identifiers(
+                exp.to_identifier(name, quoted=True), dialect=self.dialect
+            ).name
+            self.schema.add_table(schema_name, cols, dialect=self.dialect)
             self._col_orig_name[name] = orig
 
     def rewrite(self, sql: str) -> str:
@@ -95,12 +105,13 @@ class CTERewriter:
 
         Returns the transformed SQL string in the target sqlglot dialect.
         If no model references are found, falls back to
-        ``session_context.transform_sql(sql)`` directly.
+        ``session_context.transform_sql(sql)`` directly (when ``fallback``
+        is ``True``); otherwise raises ``ValueError``.
         """
         ast = parse_one(sql, dialect=self.dialect)
 
         user_cte_names = self._collect_user_cte_names(ast)
-        used_columns = self._collect_model_columns(ast, user_cte_names)
+        used_columns, user_table_refs = self._collect_model_columns(ast, user_cte_names)
 
         # No model references detected — either fall back to the legacy
         # whole-query transform, or raise so tests can catch the miss.
@@ -110,9 +121,19 @@ class CTERewriter:
                 return sqlglot.transpile(wren_sql, read="wren", write=self.dialect)[0]
             raise ValueError(f"No model references found in SQL: {sql}")
 
-        model_ctes = self._build_model_ctes(used_columns)
+        model_ctes = self._build_model_ctes(used_columns, user_table_refs)
         self._inject_ctes(ast, model_ctes)
-        return ast.sql(dialect=self.dialect)
+        # Oracle uppercases unquoted identifiers. Without forcing quoting
+        # on output, the user's ``SELECT o_orderkey FROM orders`` would
+        # land as ``SELECT O_ORDERKEY FROM ORDERS`` — both the table
+        # reference and the result column name. The injected CTE projects
+        # quoted lowercase columns, so the lookup misses (ORA-00904), and
+        # any caller asserting on result-column casing breaks. Forcing
+        # quoting on Oracle makes the dialect's output deterministic and
+        # matches the pre-fallback path where wren-core's whole-query
+        # transform had quoted everything implicitly.
+        identify = self.data_source == DataSource.oracle
+        return ast.sql(dialect=self.dialect, identify=identify)
 
     # ------------------------------------------------------------------
     # Column collection via qualify
@@ -120,28 +141,43 @@ class CTERewriter:
 
     def _collect_model_columns(
         self, ast: exp.Expression, user_cte_names: set[str]
-    ) -> dict[str, list[str] | None]:
-        """Return ``{model_name: [col1, col2, ...]}`` for all referenced models.
+    ) -> tuple[dict[str, list[str] | None], dict[str, tuple[str, bool]]]:
+        """Return ``(used_columns, user_table_refs)`` for all referenced models.
 
-        A value of ``None`` means the model was referenced via ``SELECT *``
-        and should be passed as-is to ``transform_sql`` so that wren-core
-        can apply column-level access control (CLAC).
+        ``used_columns``: ``{model_name: [col1, col2, ...]}``. A value of
+        ``None`` means the model was referenced via ``SELECT *`` and should
+        be passed as-is to ``transform_sql`` so wren-core applies CLAC.
+
+        ``user_table_refs``: ``{model_name: (user_name, user_quoted)}``
+        capturing the literal identifier the user wrote (case + quoting)
+        for the first occurrence of each model. The CTE alias matches that
+        so dialects with case-folding (Oracle uppercases unquoted ⇒ the
+        emitted CTE must fold to the same form) bind the user's outer
+        reference to the injected CTE.
 
         Uses sqlglot's ``qualify_columns`` to fully resolve all column
-        references (including ``SELECT *`` expansion and
-        correlated subquery outer references), then walks the qualified AST
-        to collect model→column mappings.  Column order follows the manifest
-        definition (via insertion order) so ``SELECT *`` preserves schema order.
+        references (including ``SELECT *`` expansion and correlated
+        subquery outer references), then walks the qualified AST to collect
+        model→column mappings. Column order follows the manifest definition
+        (via insertion order) so ``SELECT *`` preserves schema order.
         """
         copy = ast.copy()
         copy = qualify_tables(copy, dialect=self.dialect)
-        copy = normalize_identifiers(copy, dialect=self.dialect)
+
+        # Resolve every table ref to its canonical (manifest-case) model name
+        # BEFORE normalize_identifiers strips case from quoted identifiers.
+        # Dialects with NORMALIZATION_STRATEGY = CASE_INSENSITIVE (BigQuery,
+        # DuckDB) lowercase even backtick-quoted names, but BigQuery table
+        # identifiers are case-sensitive at the storage layer — capturing the
+        # alias-to-model map pre-normalize keeps the right model bound.
+        alias_to_model, user_table_refs = self._build_alias_map(copy, user_cte_names)
 
         # Detect models referenced via SELECT * BEFORE qualify_columns
         # expands the star.  These will use SELECT * in transform_sql so
         # that wren-core controls column visibility (CLAC).
-        star_models = self._detect_star_models(copy, user_cte_names)
+        star_models = self._detect_star_models(copy, alias_to_model)
 
+        copy = normalize_identifiers(copy, dialect=self.dialect)
         qualified = qualify_columns(
             copy,
             schema=self.schema,
@@ -149,62 +185,174 @@ class CTERewriter:
             allow_partial_qualification=True,
         )
 
-        # Build alias → model name mapping from Table nodes
-        alias_to_model: dict[str, str] = {}
-        for table in qualified.find_all(exp.Table):
-            table_name = table.name
-            if table_name not in self.model_dict or table_name in user_cte_names:
-                continue
-            alias = table.alias
-            if alias:
-                alias_to_model[alias] = table_name
-            alias_to_model[table_name] = table_name
-
         # Ensure every referenced model appears in the result, even if no
         # specific columns are referenced (e.g. SELECT COUNT(*) FROM model).
         # Use dict as ordered set to preserve insertion order and deduplicate.
         used: dict[str, dict[str, None]] = {m: {} for m in alias_to_model.values()}
+        # Lowercase index for column.table lookups — column qualifier may have
+        # been normalized (lowercased) by qualify_columns even though we built
+        # the alias map from the pre-normalize AST.
+        alias_lookup = {k.lower(): v for k, v in alias_to_model.items()}
         for col in qualified.find_all(exp.Column):
             table_ref = col.table
             if not table_ref:
                 continue
-            model_name = alias_to_model.get(table_ref)
+            model_name = alias_lookup.get(table_ref.lower())
             if model_name:
                 used[model_name][col.name] = None
 
-        return {m: None if m in star_models else list(cols) for m, cols in used.items()}
+        return (
+            {m: None if m in star_models else list(cols) for m, cols in used.items()},
+            user_table_refs,
+        )
+
+    def _build_alias_map(
+        self, ast: exp.Expression, user_cte_names: set[str]
+    ) -> tuple[dict[str, str], dict[str, tuple[str, bool]]]:
+        """Map each table reference in *ast* to its canonical model name.
+
+        Returns ``(alias_to_model, user_table_refs)`` — the second dict
+        records the first user-written ``(name, quoted)`` per model so the
+        CTE alias can be emitted with the same quoting style the user
+        wrote, which is required for dialects with case-folding.
+
+        Honours SQL identifier rules: quoted ⇒ case-sensitive, unquoted ⇒
+        exact match preferred, then case-insensitive fallback. Skips tables
+        that resolve to a user-defined CTE rather than an MDL model.
+        """
+        alias_to_model: dict[str, str] = {}
+        user_table_refs: dict[str, tuple[str, bool]] = {}
+        for table in ast.find_all(exp.Table):
+            name = table.name
+            if not name or name.lower() in user_cte_names:
+                continue
+            quoted = (
+                bool(table.this.quoted)
+                if isinstance(table.this, exp.Identifier)
+                else False
+            )
+            model_name = resolve_model_name(name, quoted, self.model_dict)
+            if model_name is None:
+                continue
+            alias = table.alias
+            if alias:
+                alias_to_model[alias] = model_name
+            alias_to_model[name] = model_name
+            user_table_refs.setdefault(model_name, (name, quoted))
+        return alias_to_model, user_table_refs
 
     # ------------------------------------------------------------------
     # CTE generation
     # ------------------------------------------------------------------
 
     def _build_model_ctes(
-        self, used_columns: dict[str, list[str] | None]
+        self,
+        used_columns: dict[str, list[str] | None],
+        user_table_refs: dict[str, tuple[str, bool]],
     ) -> list[exp.CTE]:
         """Generate one CTE per model via wren-core transform_sql."""
         ctes: list[exp.CTE] = []
         for model_name, columns in used_columns.items():
             if columns is None:
                 # SELECT * — let wren-core handle column visibility (CLAC)
-                col_list = "*"
+                model_sql = f'SELECT * FROM "{model_name}"'
             elif columns:
+                # ``_col_orig_name`` is keyed by lowercase column names; the
+                # column refs come from the post-normalize AST whose case
+                # depends on the dialect (Oracle uppercases unquoted idents,
+                # Postgres lowercases them). Lower-case before lookup so the
+                # original manifest casing is restored either way.
                 orig = self._col_orig_name.get(model_name, {})
-                resolved = [orig.get(c, c) for c in columns]
+                resolved = [orig.get(c.lower(), c) for c in columns]
                 col_list = ", ".join(f'"{model_name}"."{c}"' for c in resolved)
+                model_sql = f'SELECT {col_list} FROM "{model_name}"'
             else:
                 # No specific columns referenced (e.g. COUNT(*)) — only need rows
-                col_list = "1"
-            model_sql = f'SELECT {col_list} FROM "{model_name}"'
+                model_sql = f'SELECT 1 FROM "{model_name}"'
             expanded = self.session_context.transform_sql(model_sql)
 
             expanded_ast = parse_one(expanded, dialect="wren")
+            # wren-core emits ``SELECT "<m>".col FROM (...) AS "<m>"`` using
+            # the model name as the outermost subquery alias. Wrapping that
+            # in ``WITH "<m>" AS (...)`` makes ``"<m>".col`` ambiguous to
+            # BigQuery — it treats the qualifier as a recursive reference to
+            # the CTE itself and rejects the query with "Table must be
+            # qualified with a dataset". Rename the outermost alias to
+            # ``wren_src_<m>`` (no leading underscore — Oracle ORA-00911) so
+            # the shadow chain breaks at the top scope.
+            self._rename_outer_alias(expanded_ast, model_name)
+
+            # Match the user's literal identifier (case + quoting) for the
+            # CTE alias so dialects with case-folding still bind the user's
+            # outer ``FROM <model>`` to the CTE. Oracle uppercases unquoted
+            # identifiers (so ``FROM orders`` resolves to ``ORDERS``); a
+            # quoted CTE ``"orders"`` would never match. Falling back to
+            # canonical model_name + quoted=True covers introspection-only
+            # callers that build their own used_columns dict without a
+            # user_table_refs entry.
+            cte_name, cte_quoted = user_table_refs.get(model_name, (model_name, True))
             cte = exp.CTE(
                 this=expanded_ast,
-                alias=exp.TableAlias(this=exp.to_identifier(model_name, quoted=True)),
+                alias=exp.TableAlias(
+                    this=exp.to_identifier(cte_name, quoted=cte_quoted)
+                ),
             )
             ctes.append(cte)
         return ctes
 
+    @staticmethod
+    def _rename_outer_alias(ast: exp.Expression, model_name: str) -> None:
+        """Rename the outermost FROM-subquery alias matching *model_name*.
+
+        Updates top-scope column refs that use *model_name* as their table
+        qualifier. Does not descend into subqueries, so inner aliases are
+        left intact.
+        """
+        if not isinstance(ast, exp.Select):
+            return
+        from_clause = ast.args.get("from_") or ast.args.get("from")
+        if from_clause is None:
+            return
+        source = from_clause.this
+        if isinstance(source, exp.Alias):
+            source = source.this
+        if not isinstance(source, exp.Subquery) or source.alias != model_name:
+            return
+
+        # Avoid a leading underscore — Oracle rejects unquoted identifiers
+        # starting with ``_`` (ORA-00911) and downstream transpiles can drop
+        # the quoting.
+        new_alias = f"wren_src_{model_name}"
+        source.set(
+            "alias",
+            exp.TableAlias(this=exp.to_identifier(new_alias, quoted=True)),
+        )
+
+        def rewrite(node: exp.Expression) -> None:
+            # Stop at subquery boundaries — inner scopes have their own
+            # alias bindings and must keep their existing qualifiers.
+            if isinstance(node, (exp.Subquery, exp.CTE)):
+                return
+            if isinstance(node, exp.Column) and node.table == model_name:
+                node.set("table", exp.to_identifier(new_alias, quoted=True))
+            for child in node.args.values():
+                if isinstance(child, list):
+                    for c in child:
+                        if isinstance(c, exp.Expression):
+                            rewrite(c)
+                elif isinstance(child, exp.Expression):
+                    rewrite(child)
+
+        # Visit every top-scope clause except FROM (already handled).
+        for key in ("expressions", "where", "group", "having", "order", "qualify"):
+            value = ast.args.get(key)
+            if isinstance(value, list):
+                for item in value:
+                    if isinstance(item, exp.Expression):
+                        rewrite(item)
+            elif isinstance(value, exp.Expression):
+                rewrite(value)
+
     # ------------------------------------------------------------------
     # CTE injection
     # ------------------------------------------------------------------
@@ -235,28 +383,19 @@ class CTERewriter:
     # ------------------------------------------------------------------
 
     def _detect_star_models(
-        self, ast: exp.Expression, user_cte_names: set[str]
+        self, ast: exp.Expression, alias_to_model: dict[str, str]
     ) -> set[str]:
         """Detect models selected via ``*`` before column qualification.
 
         A bare ``SELECT *`` marks all models; ``SELECT t.*`` marks only
-        the referenced model.
+        the referenced model. *alias_to_model* is the case-aware mapping
+        produced by ``_build_alias_map``.
         """
         star_models: set[str] = set()
         select = ast.find(exp.Select)
         if not select:
             return star_models
 
-        # Build alias → model mapping from tables in FROM/JOIN
-        alias_to_model: dict[str, str] = {}
-        for table in ast.find_all(exp.Table):
-            name = table.name
-            if name not in self.model_dict or name in user_cte_names:
-                continue
-            alias = table.alias or name
-            alias_to_model[alias] = name
-            alias_to_model[name] = name
-
         for sel_expr in select.expressions:
             if isinstance(sel_expr, exp.Star):
                 # Bare * → all models

core/wren/src/wren/policy.py

@@ -6,12 +6,41 @@ manifest and does not use any denied functions.
 
 from __future__ import annotations
 
+from typing import Iterable
+
 from sqlglot import exp
 
 from wren.config import WrenConfig
 from wren.model.error import ErrorCode, ErrorPhase, WrenError
 
 
+def resolve_model_name(
+    name: str,
+    quoted: bool,
+    model_names: Iterable[str],
+) -> str | None:
+    """Resolve a SQL table identifier to a manifest model name.
+
+    Follows the SQL convention used across Wren's CTE rewriter, policy check,
+    and manifest extractor: a quoted identifier must match a model name
+    case-sensitively; an unquoted identifier prefers an exact case match but
+    falls back to a case-insensitive scan. Returns ``None`` if no model
+    matches.
+    """
+    model_set = (
+        model_names if isinstance(model_names, (set, frozenset)) else set(model_names)
+    )
+    if name in model_set:
+        return name
+    if quoted:
+        return None
+    name_lower = name.lower()
+    for candidate in model_set:
+        if candidate.lower() == name_lower:
+            return candidate
+    return None
+
+
 def validate_sql_policy(
     ast: exp.Expression,
     model_names: set[str],
@@ -59,8 +88,6 @@ def _check_tables(
     ast: exp.Expression,
     model_names: set[str],
 ) -> None:
-    model_names_lower = {n.lower() for n in model_names}
-
     for table in ast.find_all(exp.Table):
         name = table.name
         if not name:
@@ -75,10 +102,12 @@ def _check_tables(
                     phase=ErrorPhase.SQL_POLICY_CHECK,
                 )
             continue
-        name_lower = name.lower()
-        if name_lower in model_names_lower:
+        quoted = (
+            bool(table.this.quoted) if isinstance(table.this, exp.Identifier) else False
+        )
+        if resolve_model_name(name, quoted, model_names) is not None:
             continue
-        if name_lower in _visible_cte_names(table):
+        if name.lower() in _visible_cte_names(table):
             continue
         raise WrenError(
             ErrorCode.MODEL_NOT_FOUND,

core/wren/tests/unit/test_case_sensitivity.py

@@ -0,0 +1,316 @@
+"""Case-sensitivity rules across the wren SDK.
+
+Locks in the contract shared by ``policy.resolve_model_name``,
+``policy.validate_sql_policy`` (strict mode), and ``CTERewriter``:
+
+- a **quoted** identifier must match a manifest model name case-sensitively
+- an **unquoted** identifier prefers an exact match, then a case-insensitive
+  scan
+
+The CTE rewriter is exercised against a manifest that intentionally contains
+two models differing only in case (``Users`` and ``users``) to prove the
+right one is bound regardless of the user's casing.
+"""
+
+from __future__ import annotations
+
+import base64
+
+import orjson
+import pytest
+from sqlglot import parse_one
+
+from wren.config import WrenConfig
+from wren.mdl import get_manifest_extractor, get_session_context, to_json_base64
+from wren.mdl.cte_rewriter import CTERewriter
+from wren.model.data_source import DataSource
+from wren.model.error import WrenError
+from wren.policy import resolve_model_name, validate_sql_policy
+
+pytestmark = pytest.mark.unit
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+def _model(name: str) -> dict:
+    return {
+        "name": name,
+        "tableReference": {
+            "catalog": "test",
+            "schema": "public",
+            "table": name.lower(),
+        },
+        "columns": [{"name": "id", "type": "integer"}],
+    }
+
+
+@pytest.fixture
+def dual_case_manifest_b64() -> str:
+    """Manifest with both ``Users`` and ``users`` as separate models."""
+    manifest = {
+        "catalog": "my_catalog",
+        "schema": "my_schema",
+        "models": [_model("Users"), _model("users"), _model("Orders")],
+    }
+    return base64.b64encode(orjson.dumps(manifest)).decode("utf-8")
+
+
+# ---------------------------------------------------------------------------
+# resolve_model_name
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("name", "quoted", "expected"),
+    [
+        # Quoted: case-sensitive exact match required.
+        ("Users", True, "Users"),
+        ("users", True, "users"),
+        ("USERS", True, None),
+        # Unquoted: exact match preferred, then case-insensitive scan.
+        ("Users", False, "Users"),
+        ("users", False, "users"),
+        ("USERS", False, "Users"),  # CI fallback resolves to first match
+        # Misses regardless of quoting.
+        ("nonexistent", True, None),
+        ("nonexistent", False, None),
+    ],
+)
+def test_resolve_model_name_dual_case(name, quoted, expected):
+    """Quoted = strict CS; unquoted = exact-then-CI fallback."""
+    model_names = {"Users", "users", "Orders"}
+    actual = resolve_model_name(name, quoted, model_names)
+    if expected is None:
+        assert actual is None
+    elif expected == "Users":
+        # ``USERS`` unquoted may pick either manifest entry depending on set
+        # iteration order — accept either as long as case-insensitive matches.
+        assert actual is not None and actual.lower() == name.lower()
+    else:
+        assert actual == expected
+
+
+def test_resolve_model_name_postgres_quoted_distinct():
+    """Postgres ``"Orders"`` (quoted, mixed case) must not match lowercase ``orders``."""
+    model_names = {"orders"}
+    assert resolve_model_name("Orders", True, model_names) is None
+    assert resolve_model_name("Orders", False, model_names) == "orders"
+    assert resolve_model_name("orders", True, model_names) == "orders"
+
+
+# ---------------------------------------------------------------------------
+# validate_sql_policy (strict mode)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("sql", "should_pass", "description"),
+    [
+        ("SELECT * FROM `Users`", True, "quoted exact match"),
+        ("SELECT * FROM `users`", True, "quoted exact match (lowercase variant)"),
+        ("SELECT * FROM `USERS`", False, "quoted no match — strict"),
+        ("SELECT * FROM Users", True, "unquoted exact match"),
+        ("SELECT * FROM users", True, "unquoted exact match"),
+        ("SELECT * FROM USERS", True, "unquoted CI fallback"),
+        ("SELECT * FROM `Nonexistent`", False, "quoted no match"),
+        ("SELECT * FROM Nonexistent", False, "unquoted no match"),
+    ],
+)
+def test_validate_sql_policy_dual_case_bigquery(sql, should_pass, description):
+    """Strict-mode policy honors quoted vs unquoted in BigQuery dialect."""
+    config = WrenConfig(strict_mode=True)
+    model_names = {"Users", "users", "Orders"}
+    ast = parse_one(sql, dialect="bigquery")
+    if should_pass:
+        validate_sql_policy(ast, model_names, config)  # no raise
+    else:
+        with pytest.raises(WrenError):
+            validate_sql_policy(ast, model_names, config)
+
+
+def test_validate_sql_policy_postgres_quoted_rejects_uppercase():
+    """``"Orders"`` against a manifest of only ``orders`` is rejected.
+
+    Postgres semantics: ``"Orders"`` (quoted) is a distinct identifier
+    from ``orders``, so it shouldn't pass strict-mode policy.
+    """
+    config = WrenConfig(strict_mode=True)
+    model_names = {"orders"}
+    with pytest.raises(WrenError):
+        validate_sql_policy(
+            parse_one('SELECT * FROM "Orders"', dialect="postgres"),
+            model_names,
+            config,
+        )
+    # And the lowercased counterpart still works.
+    validate_sql_policy(
+        parse_one('SELECT * FROM "orders"', dialect="postgres"),
+        model_names,
+        config,
+    )
+
+
+# ---------------------------------------------------------------------------
+# CTERewriter — end-to-end model resolution
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("table_ref", "expected_model"),
+    [
+        ("`Users`", "Users"),
+        ("`users`", "users"),
+    ],
+)
+def test_cte_rewriter_quoted_preserves_case(
+    dual_case_manifest_b64, table_ref, expected_model
+):
+    """The injected CTE must bind the model the user actually wrote.
+
+    With both ``Users`` and ``users`` in the manifest, lowercasing the
+    backtick-quoted name (the pre-fix behavior of ``normalize_identifiers``
+    on BigQuery) would silently pick the wrong model.
+    """
+    extractor = get_manifest_extractor(dual_case_manifest_b64)
+    mini = extractor.extract_by([expected_model])
+    mini_b64 = to_json_base64(mini)
+    session = get_session_context(mini_b64, None, None, "bigquery")
+    rewriter = CTERewriter(mini_b64, session, DataSource.bigquery, fallback=False)
+
+    sql = f"SELECT u.id FROM {table_ref} AS u"
+    ast = parse_one(sql, dialect="bigquery")
+    user_ctes = rewriter._collect_user_cte_names(ast)
+    used, _refs = rewriter._collect_model_columns(ast, user_ctes)
+
+    assert list(used.keys()) == [expected_model], (
+        f"expected the rewriter to bind {expected_model!r} for {table_ref}, "
+        f"got {list(used.keys())!r}"
+    )
+
+
+def test_cte_rewriter_unquoted_ci_fallback(dual_case_manifest_b64):
+    """An unquoted reference matching a model exactly wins over CI fallback."""
+    extractor = get_manifest_extractor(dual_case_manifest_b64)
+    mini = extractor.extract_by(["Users"])
+    mini_b64 = to_json_base64(mini)
+    session = get_session_context(mini_b64, None, None, "bigquery")
+    rewriter = CTERewriter(mini_b64, session, DataSource.bigquery, fallback=False)
+
+    # Unquoted ``Users`` matches the ``Users`` model (exact case) — not
+    # ``users`` even though both exist.
+    ast = parse_one("SELECT u.id FROM Users AS u", dialect="bigquery")
+    used, _refs = rewriter._collect_model_columns(ast, set())
+    assert list(used.keys()) == ["Users"]
+
+
+def test_cte_rewriter_renames_outer_alias_to_avoid_bigquery_shadow():
+    """Outermost subquery alias inside the model CTE must not match the CTE name.
+
+    wren-core's transform_sql emits ``SELECT "<m>".col FROM (...) AS "<m>"``;
+    when wrapped in ``WITH "<m>" AS (...)`` BigQuery treats the qualifier
+    as a recursive reference to the CTE itself and rejects the query with
+    "Table must be qualified with a dataset". CTERewriter renames just the
+    outermost alias to ``wren_src_<m>`` to break the shadow chain.
+    """
+    manifest = {
+        "catalog": "wren",
+        "schema": "public",
+        "models": [
+            {
+                "name": "Cards_Cleaned",
+                "tableReference": {
+                    "catalog": "proj",
+                    "schema": "ds",
+                    "table": "Cards_Cleaned",
+                },
+                "columns": [
+                    {"name": "id", "type": "integer"},
+                    {"name": "card_type", "type": "varchar"},
+                ],
+            }
+        ],
+    }
+    manifest_b64 = base64.b64encode(orjson.dumps(manifest)).decode("utf-8")
+    session = get_session_context(manifest_b64, None, None, "bigquery")
+    rewriter = CTERewriter(manifest_b64, session, DataSource.bigquery, fallback=False)
+
+    rewritten = rewriter.rewrite("SELECT c.card_type FROM `Cards_Cleaned` AS c")
+
+    # Sentinel must appear (the outer alias was renamed).
+    assert "wren_src_Cards_Cleaned" in rewritten
+    # The top-scope of the CTE body must not have ``AS `Cards_Cleaned``` —
+    # only the renamed sentinel as the outermost alias. The middle/inner
+    # scopes can still use the original name (separate parentheses).
+    cte_body = rewritten.split("`Cards_Cleaned` AS (", 1)[1]
+    cte_body_top = cte_body.split(")", 1)[0]
+    assert "AS `Cards_Cleaned`" not in cte_body_top
+
+
+def test_cte_rewriter_oracle_emits_quoted_identifiers():
+    """Oracle output must force-quote all identifiers (``identify=True``).
+
+    Without forced quoting, Oracle uppercases unquoted refs, mismatching
+    the CTE's quoted lowercase columns (ORA-00904). With it, the user's
+    outer ``FROM orders`` and ``SELECT o_orderkey`` both emit quoted
+    lowercase, matching the CTE — and the result column names stay in
+    the original lowercase case rather than getting folded to uppercase.
+    """
+    manifest = {
+        "catalog": "wren",
+        "schema": "public",
+        "models": [
+            {
+                "name": "orders",
+                "tableReference": {"schema": "SYSTEM", "table": "orders"},
+                "columns": [{"name": "o_orderkey", "type": "integer"}],
+            }
+        ],
+    }
+    manifest_b64 = base64.b64encode(orjson.dumps(manifest)).decode("utf-8")
+    session = get_session_context(manifest_b64, None, None, "oracle")
+    rewriter = CTERewriter(manifest_b64, session, DataSource.oracle, fallback=False)
+
+    rewritten = rewriter.rewrite("SELECT o_orderkey FROM orders")
+    # Every identifier the user wrote must come out quoted so Oracle
+    # doesn't case-fold them.
+    assert 'WITH "orders" AS' in rewritten, rewritten
+    assert 'SELECT "o_orderkey" FROM "orders"' in rewritten, rewritten
+
+
+def test_cte_rewriter_oracle_uppercases_columns():
+    """Oracle uppercases unquoted columns; CTE body must restore manifest case.
+
+    Regression test for the case where ``_col_orig_name`` (lowercase-keyed)
+    was looked up with the post-normalize column name (``O_ORDERKEY`` on
+    Oracle), missing the entry and emitting an uppercase column that
+    wren-core's schema check then rejected.
+    """
+    manifest = {
+        "catalog": "wren",
+        "schema": "public",
+        "models": [
+            {
+                "name": "orders",
+                "tableReference": {"schema": "SYSTEM", "table": "orders"},
+                "columns": [
+                    {"name": "o_orderkey", "type": "integer"},
+                    {"name": "o_custkey", "type": "integer"},
+                ],
+            }
+        ],
+    }
+    manifest_b64 = base64.b64encode(orjson.dumps(manifest)).decode("utf-8")
+    session = get_session_context(manifest_b64, None, None, "oracle")
+    rewriter = CTERewriter(manifest_b64, session, DataSource.oracle, fallback=False)
+
+    sql = "SELECT o_orderkey FROM orders"
+    rewritten = rewriter.rewrite(sql)
+    # Must contain the original lowercase column from the manifest, not the
+    # post-normalize uppercase form. Don't lowercase the output before
+    # checking — that would mask a regression where Oracle's uppercased
+    # ``O_ORDERKEY`` leaks into the rewritten CTE body.
+    assert '"o_orderkey"' in rewritten, rewritten
+    assert '"O_ORDERKEY"' not in rewritten, rewritten