fix(ol_dbt_cli): sql_parser — fix 3 Jinja-collapse defects (trailing-comma CTE, alias loss, subquery FP) (#2454)

* fix(ol_dbt_cli): sql_parser — fix 3 Jinja-collapse defects (trailing-comma CTE, alias loss, subquery FP)

Closes the three remaining post-render collapse defects from the 2026-07-13
sql_parser audit (tk-...770aaa). After this, all 659 models parse (was 3
failing) and a full `ol-dbt validate` reports 0 ERRORs (was 1).

(1) TRAILING-COMMA CTE MACRO — a CTE-injecting macro followed by a comma
(`{{ deduplicate_raw_table(...) }},`) renders to `__macro__,` alone on its
line in a WITH list. The bare standalone-line rule only matched a placeholder
alone on its line, so the trailing comma left an invalid bare identifier and
sqlglot failed with "Could not find outermost SELECT" (3 stg__ models). Add a
rule that replaces it with a valid placeholder CTE — with a leading comma,
since these injector macros emit leading-comma CTEs and always follow a base
CTE that carries no trailing comma.

(2) NESTED-ALIAS LOSS — a multi-line `if(cond, a, {{ macro(...) }}) as alias`
(int__micromasters__program_certificates) was mis-collapsed: the broken-column
repair treated the `if`'s last argument as a split column and deleted the `)`,
swallowing the alias into the function and dropping the output column. Gate the
collapse on a parse check — it only runs when the SQL is *actually* broken
(unbalanced), which a well-formed multi-line function call is not.

(3) SUBQUERY FALSE POSITIVE — get_columns_read_from_ref Pass 2 descended into
nested subqueries via `clause.find_all(exp.Column)`, so
`where x in (select y from other_ref where other_col = ...)` attributed
`other_col` to the OUTER ref (the sole ERROR-level FP in a full validate run,
in edxorg_to_mitxonline_enrollments). Skip columns whose nearest enclosing
Select is not the passthrough source's own Select.

355 CLI tests pass (6 new), pre-commit (ruff+mypy) clean. Full-repo sweep: 0
parse errors across 659 models; full validate: 0 ERRORs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* perf(ol_dbt_cli): gate the broken-column parse-check behind a cheap regex precheck

Addresses Copilot review on #2454: _collapse_broken_column_expressions called
_parses_cleanly (a full sqlglot parse) for every model, doubling parse work
across validate/impact. Add a _BROKEN_COL_RE.search() fast path so the parse
gate only runs when the broken-column shape is actually present. Measured over
the repo: the expensive parse now runs 15× across 659 models instead of 659×,
with 0 parse errors unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: blarghmatey

src/ol_dbt_cli/ol_dbt_cli/lib/sql_parser.py

@@ -148,6 +148,21 @@ def _render_jinja(sql: str) -> tuple[str, list[str], list[str], dict[str, str],
     template = env.from_string(sql)
     rendered = template.render()
 
+    # A block macro that injects a *CTE definition* renders as "__macro__," alone on
+    # its line inside a WITH list (e.g. `{{ deduplicate_raw_table(...) }},` between two
+    # named CTEs). The bare standalone-line rule below only matches a placeholder alone
+    # on its line, so the trailing comma would leave `__macro__,` as an invalid bare
+    # identifier in the CTE list and sqlglot fails with "Could not find outermost
+    # SELECT". Replace it with a syntactically valid placeholder CTE first so the
+    # surrounding CTE list still parses (the placeholder CTE is never referenced).
+    # The leading comma reconnects to the preceding CTE: these injector macros
+    # (e.g. deduplicate_raw_table) emit leading-comma CTEs and always follow a
+    # `source`/base CTE, which itself carries no trailing comma.
+    rendered = re.sub(
+        r"(?m)^[^\S\r\n]*(?:__macro__|__undefined__)[^\S\r\n]*,[^\S\r\n]*$",
+        ", __jinja_cte__ as (select 1),",
+        rendered,
+    )
     # Block-level macro injections (macros that inject CTE SQL fragments) render as
     # "__macro__" which occupies a statement-level position where a string literal
     # is invalid SQL.  Replace standalone __macro__ / __undefined__ lines with SQL comments.
@@ -263,7 +278,25 @@ def _collapse_broken_column_expressions(sql: str) -> str:
     Example (after)::
 
         , __jinja__ as course_level
+
+    The repair only runs when the SQL is *actually* broken. A well-formed
+    multi-line function call whose last argument is a macro placeholder —
+    ``if(cond, a, __macro__) as alias`` spanning several lines — parses cleanly,
+    and the ``, __macro__`` there is a function argument, not a split column. The
+    ``) as alias`` tokens close the ``if(`` and carry the real alias; collapsing
+    would delete that ``)`` and swallow the alias into the function, dropping the
+    output column. So we skip the collapse entirely when the input already parses.
     """
+    # Fast path: the parse gate below is a full sqlglot parse, so only pay for it
+    # when the broken-column shape (`, __placeholder__ … ) as alias`) is actually
+    # present. The overwhelming majority of models never match, so they skip both
+    # the parse and the collapse on a single cheap regex search.
+    if not _BROKEN_COL_RE.search(sql):
+        return sql
+    # The shape is present but may be a well-formed multi-line function call that
+    # merely looks like it — only repair when the SQL genuinely does not parse.
+    if _parses_cleanly(sql):
+        return sql
 
     def _replace(m: re.Match[str]) -> str:
         return f"{m.group(1)}__jinja__ as {m.group(2)}\n"
@@ -276,6 +309,15 @@ def _replace(m: re.Match[str]) -> str:
     return sql
 
 
+def _parses_cleanly(sql: str) -> bool:
+    """Return True if *sql* parses under Trino without raising a syntax error."""
+    try:
+        sqlglot.parse(sql, dialect="trino", error_level=sqlglot.ErrorLevel.RAISE)
+    except Exception:  # noqa: BLE001 — any parse error means "not clean"
+        return False
+    return True
+
+
 def strip_jinja(sql: str) -> JinjaStripResult:
     """Process Jinja in *sql* and return a :class:`JinjaStripResult`.
 
@@ -582,6 +624,21 @@ def _first_non_cte_select(root: exp.Expression) -> exp.Select | None:
     return None
 
 
+def _nearest_enclosing_select(node: exp.Expression) -> exp.Select | None:
+    """Return the closest ``Select`` ancestor of *node*, or None if it has none.
+
+    Used to tell whether a column lives directly in a given SELECT or inside one
+    of its nested subqueries — a column in ``where x in (select y from ...)`` has
+    the inner query as its nearest enclosing Select, not the outer one.
+    """
+    parent = node.parent
+    while parent is not None:
+        if isinstance(parent, exp.Select):
+            return parent
+        parent = parent.parent
+    return None
+
+
 def _get_star_source(select: exp.Select) -> str:
     """Return the alias-or-name of the relation the outermost SELECT * draws from."""
     from_clause = select.args.get("from_")
@@ -1001,14 +1058,21 @@ def get_columns_read_from_ref(
             clauses.append(where)
         for clause in clauses:
             for col_node in clause.find_all(exp.Column):
-                if col_node.args.get("table") is None:
-                    col_name = col_node.name.lower()
-                    if (
-                        col_name
-                        and col_name != "*"
-                        and col_name not in jinja_placeholders
-                        and col_name not in passthrough_computed
-                    ):
-                        cols_read.add(col_name)
+                if col_node.args.get("table") is not None:
+                    continue
+                # Prune columns that live inside a nested subquery of this clause
+                # (e.g. `where x in (select y from other_ref where z = ...)`): their
+                # nearest enclosing SELECT is the inner query, so they belong to a
+                # different ref and must not be attributed to this passthrough source.
+                if _nearest_enclosing_select(col_node) is not select_node:
+                    continue
+                col_name = col_node.name.lower()
+                if (
+                    col_name
+                    and col_name != "*"
+                    and col_name not in jinja_placeholders
+                    and col_name not in passthrough_computed
+                ):
+                    cols_read.add(col_name)
 
     return cols_read if cols_read else None

src/ol_dbt_cli/tests/test_impact.py

@@ -258,6 +258,43 @@ def test_returns_none_when_upstream_not_referenced(self, tmp_path: Path) -> None
         result = get_columns_read_from_ref(parsed, "stg_users")
         assert result is None
 
+    def test_subquery_column_not_attributed_to_outer_ref(self, tmp_path: Path) -> None:
+        """A column inside a WHERE subquery over a different ref is not attributed to the outer ref.
+
+        `where x in (select y from other_ref where other_col = ...)`: `other_col`
+        belongs to the inner subquery's ref, not the outer passthrough source, so it
+        must not be reported as a column read from the outer upstream. Regression for
+        the sole ERROR-level false positive in a full `ol-dbt validate` run.
+        """
+        from ol_dbt_cli.lib.sql_parser import get_columns_read_from_ref, parse_model_sql
+
+        sql = (
+            "with base as (select * from ref_reversion_version)\n"
+            "select base.version_id\n"
+            "from base\n"
+            "where base.contenttype_id in (\n"
+            "    select ct.id from ref_django_contenttype ct\n"
+            "    where ct.contenttype_full_name = 'x'\n"
+            ")"
+        )
+        sql_file = tmp_path / "downstream.sql"
+        sql_file.write_text(sql)
+        parsed = parse_model_sql("downstream", sql)
+        parsed.refs = ["reversion_version", "django_contenttype"]
+        parsed.ref_placeholder_map = {
+            "ref_reversion_version": "reversion_version",
+            "ref_django_contenttype": "django_contenttype",
+        }
+        parsed.source_path = sql_file
+
+        result = get_columns_read_from_ref(parsed, "reversion_version")
+        # version_id and contenttype_id are genuine reads from the outer ref…
+        assert result is not None
+        assert "version_id" in result
+        assert "contenttype_id" in result
+        # …but contenttype_full_name belongs to the inner subquery's ref, not this one.
+        assert "contenttype_full_name" not in result
+
     def test_detects_aliased_column_read(self, tmp_path: Path) -> None:
         """Column read from upstream but aliased to different name in output is detected."""
         from ol_dbt_cli.lib.sql_parser import get_columns_read_from_ref, parse_model_sql

src/ol_dbt_cli/tests/test_sql_parser.py

@@ -304,6 +304,55 @@ def test_broken_column_regex_stops_at_plain_multiline_alias(self) -> None:
         assert "user_hashed_id" in result.output_columns
         assert "unique_courses" in result.output_columns
 
+    def test_block_macro_with_trailing_comma_becomes_cte(self) -> None:
+        """A CTE-injecting macro followed by a comma (`{{ macro(...) }},`) must parse.
+
+        The macro renders to `__macro__,` alone on its line inside a WITH list; the
+        bare-line rule only matches a placeholder alone on its line, so without special
+        handling the trailing comma leaves an invalid bare identifier and sqlglot fails
+        with "Could not find outermost SELECT". It is replaced with a placeholder CTE
+        (leading comma reconnects to the preceding base CTE, which carries no comma).
+        """
+        sql = (
+            "with\n"
+            "    source as (\n"
+            "        select 1 as id, 2 as user_id\n"
+            "    )\n"
+            "    {{ deduplicate_raw_table(order_by='id', partition_columns='user_id') }},\n"
+            "    cleaned as (\n"
+            "        select id as courserunenrollment_id, user_id\n"
+            "        from most_recent_source\n"
+            "    )\n"
+            "select * from cleaned"
+        )
+        result = parse_model_sql("my_model", sql)
+        assert result.parse_error is None
+        assert "courserunenrollment_id" in result.output_columns
+        assert "user_id" in result.output_columns
+
+    def test_multiline_function_over_macro_keeps_alias(self) -> None:
+        """`if(cond, a, {{ macro(...) }}) as alias` spanning lines must keep the alias.
+
+        Here `, __macro__` is the function's last argument, not a split column, and
+        `) as alias` closes the `if(` while carrying the real alias. The broken-column
+        collapse must NOT fire (it would delete the `)` and swallow the alias into the
+        function), so the output column survives.
+        """
+        sql = (
+            "with report as (select 1 as program_id, 'x' as program_title)\n"
+            "select\n"
+            "    report.program_id\n"
+            "    , if(\n"
+            "        report.program_id is not null\n"
+            "        , report.program_title\n"
+            "        , {{ generate_micromasters_program_readable_id('report.program_id', 'report.program_title') }}\n"
+            "    ) as program_readable_id\n"
+            "from report"
+        )
+        result = parse_model_sql("my_model", sql)
+        assert result.parse_error is None
+        assert "program_readable_id" in result.output_columns
+
     def test_var_in_where_clause_parseable(self) -> None:
         """A model with '{{ var(...) }}' in WHERE must parse without error."""
         sql = (