fix cclf transpilation: cast types, concat_str, unique order_by, recursion limit

- transpile.py: raise recursion limit to 5000 for deeply nested pivot SQL
- cclf.py: cast to Float64 before arithmetic in _negate_if_cancelled,
  use nw.concat_str for bill_type_code, add order_by to unique(),
  use keep="any" for stg_beneficiary_demographics
- tests: reduce known transpile failures from 18 to 2, add known
  parse failures for pivot expressions

src/aco/express/cclf.py

@@ -62,10 +62,11 @@ def _icd_indicator() -> nw.Expr:
 
 def _negate_if_cancelled(col: str) -> nw.Expr:
     """Negate amount when clm_adjsmt_type_cd = '1' (cancelled)."""
+    amt = nw.col(col).cast(nw.Float64)
     return (
         nw.when(nw.col("clm_adjsmt_type_cd") == "1")
-        .then(nw.col(col) * -1)
-        .otherwise(nw.col(col))
+        .then(amt * -1)
+        .otherwise(amt)
         .alias(col)
     )
 
@@ -103,13 +104,13 @@ def stg_beneficiary_xref(
     CCLF9 fields used: CRNT_NUM (current MBI), PRVS_NUM (previous MBI),
     PRVS_ID_EFCTV_DT (effective date of previous ID — used to pick latest).
     """
-    return (
-        cclf__cclf9.sort("prvs_id_efctv_dt", descending=True)
-        .unique(subset=["prvs_num"], keep="first")
-        .select(
-            nw.col("crnt_num"),
-            nw.col("prvs_num"),
-        )
+    return cclf__cclf9.unique(
+        subset=["prvs_num"],
+        keep="last",
+        order_by=["prvs_id_efctv_dt"],
+    ).select(
+        nw.col("crnt_num"),
+        nw.col("prvs_num"),
     )
 
 
@@ -505,10 +506,13 @@ def int_institutional_medical_claim(
         nw.col("bene_ptnt_stus_cd").alias("discharge_disposition_code"),
         nw.lit(None).cast(nw.String).alias("place_of_service_code"),
         # bill_type_code = fac + clsfctn + freq
-        (
-            nw.col("clm_bill_fac_type_cd").cast(nw.String).fill_null("")
-            + nw.col("clm_bill_clsfctn_cd").cast(nw.String).fill_null("")
-            + nw.col("clm_bill_freq_cd").cast(nw.String).fill_null("")
+        nw.concat_str(
+            [
+                nw.col("clm_bill_fac_type_cd").cast(nw.String).fill_null(""),
+                nw.col("clm_bill_clsfctn_cd").cast(nw.String).fill_null(""),
+                nw.col("clm_bill_freq_cd").cast(nw.String).fill_null(""),
+            ],
+            separator="",
         ).alias("bill_type_code"),
         nw.lit("ms-drg").alias("drg_code_type"),
         # DRG: rightmost 3 chars
@@ -1175,7 +1179,7 @@ def stg_beneficiary_demographics(
         .with_columns(
             nw.coalesce("crnt_num", "bene_mbi_id").alias("current_bene_mbi_id"),
         )
-        .unique(subset=["current_bene_mbi_id"], keep="first")
+        .unique(subset=["current_bene_mbi_id"], keep="any")
     )
 
 

src/aco/lake/transpile.py

@@ -154,11 +154,19 @@ def transpile(
     dict[str, str]
         Mapping of expression name to transpiled SQL string.
     """
+    import sys
+
     own_connection = False
     if con is None:
         con = _schema_only_connection()
         own_connection = True
 
+    # Pivot/unpivot expressions can generate 100K+ char SQL whose
+    # deeply nested AST exceeds Python's default recursion limit
+    # during sqlglot parse, tree traversal, and code generation.
+    old_limit = sys.getrecursionlimit()
+    sys.setrecursionlimit(max(old_limit, 5000))
+
     cache: dict[str, Any] = {}
     result: dict[str, str] = {}
     view_registry: dict[str, str] = {}
@@ -209,6 +217,7 @@ def transpile(
             )
             result[name] = target_sql
     finally:
+        sys.setrecursionlimit(old_limit)
         if own_connection:
             con.close()
 

tests/aco/test_transpile_databricks.py

@@ -102,30 +102,23 @@ def _flat_cases(all_transpiled):
 # ── Test: every expression transpiles without error ───────────────
 
 
-# Expressions with known sqlglot limitations (deeply nested DuckDB SQL
-# from pivot/unpivot operations) or narwhals schema-only incompatibilities.
-# The cclf pipeline cascades: once _int_diagnosis_pivot fails, all
-# downstream expressions that depend on it (directly or transitively)
-# also fail.
+# Schema-only transpilation limitation: nw.concat requires identical
+# schemas, but the three medical_claim intermediate views produce
+# different column counts in schema-only mode (DuckDB doesn't fully
+# preserve dynamically-generated null columns from narwhals expressions).
+# Works correctly with real data.
 _KNOWN_TRANSPILE_FAILURES = {
-    "cclf._stg_beneficiary_xref",
+    "cclf.medical_claim",
+    # DuckDB and Polars produce different column suffixes after join:
+    # Polars adds _rev to right-side columns, DuckDB schema-only doesn't.
+    "cclf._int_institutional_medical_claim",
+}
+
+# Pivot expressions produce valid SQL but are too deeply nested for
+# sqlglot to re-parse in a second pass (RecursionError).
+_KNOWN_PARSE_FAILURES = {
     "cclf._int_diagnosis_pivot",
     "cclf._int_procedure_pivot",
-    "cclf._int_institutional_medical_claim",
-    "cclf._stg_physician_claim",
-    "cclf._int_physician_claim_adr",
-    "cclf._int_physician_medical_claim",
-    "cclf._stg_dme_claim",
-    "cclf._int_dme_claim_adr",
-    "cclf._int_dme_medical_claim",
-    "cclf.medical_claim",
-    "cclf._stg_pharmacy_claim",
-    "cclf._int_pharmacy_claim_adr",
-    "cclf.pharmacy_claim",
-    "cclf._stg_beneficiary_demographics",
-    "cclf.eligibility",
-    "provider_attribution._int_current_steps",
-    "provider_attribution._int_yearly_steps",
 }
 
 
@@ -177,6 +170,8 @@ class TestValidDatabricksSQL:
         for name, sql in sql_map.items():
             if sql.startswith("-- ERROR"):
                 continue
+            if name in _KNOWN_PARSE_FAILURES:
+                continue
             try:
                 tree = sqlglot.parse_one(sql, dialect="databricks")
                 assert isinstance(tree, (exp.Select, exp.Union))