fix all cclf transpile errors: pivot rewrite, column fix, schema alignment

- Rewrite diagnosis/procedure pivots from 25-iteration progressive
  joins to conditional aggregation (group_by + when/then/max),
  eliminating deeply nested SQL that caused sqlglot RecursionError
- Fix clm_line_srvc_unit_qty column reference (no _rev suffix needed,
  column is unique to right side of join)
- Move _nullify_sentinel into claim_start_date/admission_date aliases
  instead of producing extra clm_from_dt column that broke nw.concat
  schema alignment across institutional/physician/DME claim types
- All transpile known failures resolved: 0 remaining

src/aco/express/cclf.py

@@ -302,31 +302,24 @@ def int_diagnosis_pivot(
         nw.col("clm_val_sqnc_num").cast(nw.String).str.strip_chars() != ""
     )
     key_cols = ["cur_clm_uniq_id", "bene_mbi_id", "dgns_prcdr_icd_ind"]
-    rows = []
+    seq = nw.col("clm_val_sqnc_num").cast(nw.String).str.strip_chars()
+    agg_exprs = []
     for i in range(1, 26):
-        seq = str(i)
-        chunk = df.filter(
-            nw.col("clm_val_sqnc_num").cast(nw.String).str.strip_chars() == seq
-        ).select(
-            *key_cols,
-            nw.col("clm_dgns_cd").alias(f"diagnosis_code_{i}"),
-            nw.col("clm_poa_ind").alias(f"diagnosis_poa_{i}"),
+        agg_exprs.append(
+            nw.when(seq == str(i))
+            .then(nw.col("clm_dgns_cd"))
+            .otherwise(nw.lit(None).cast(nw.String))
+            .max()
+            .alias(f"diagnosis_code_{i}")
         )
-        rows.append(chunk)
-
-    if not rows:  # pragma: no cover — range(1,26) always yields items
-        return df.head(0).select(*key_cols)
-
-    result = rows[0]
-    for chunk in rows[1:]:
-        result = result.join(chunk, on=key_cols, how="full", suffix="_r")
-        for col_name in key_cols:
-            r_col = f"{col_name}_r"
-            if r_col in result.columns:
-                result = result.with_columns(
-                    nw.coalesce(col_name, r_col).alias(col_name)
-                ).drop(r_col)
-    return result
+        agg_exprs.append(
+            nw.when(seq == str(i))
+            .then(nw.col("clm_poa_ind"))
+            .otherwise(nw.lit(None).cast(nw.String))
+            .max()
+            .alias(f"diagnosis_poa_{i}")
+        )
+    return df.group_by(key_cols).agg(*agg_exprs)
 
 
 # ── Stage 5: Procedure pivot (CCLF3 long → wide) ────────────────
@@ -365,31 +358,24 @@ def int_procedure_pivot(
         nw.col("clm_val_sqnc_num").cast(nw.String).str.strip_chars() != ""
     )
     key_cols = ["cur_clm_uniq_id", "bene_mbi_id", "dgns_prcdr_icd_ind"]
-    rows = []
+    seq = nw.col("clm_val_sqnc_num").cast(nw.String).str.strip_chars()
+    agg_exprs = []
     for i in range(1, 26):
-        seq = str(i)
-        chunk = df.filter(
-            nw.col("clm_val_sqnc_num").cast(nw.String).str.strip_chars() == seq
-        ).select(
-            *key_cols,
-            nw.col("clm_prcdr_cd").alias(f"procedure_code_{i}"),
-            nw.col("clm_prcdr_prfrm_dt").alias(f"procedure_date_{i}"),
+        agg_exprs.append(
+            nw.when(seq == str(i))
+            .then(nw.col("clm_prcdr_cd"))
+            .otherwise(nw.lit(None).cast(nw.String))
+            .max()
+            .alias(f"procedure_code_{i}")
         )
-        rows.append(chunk)
-
-    if not rows:  # pragma: no cover — range(1,26) always yields items
-        return df.head(0).select(*key_cols)
-
-    result = rows[0]
-    for chunk in rows[1:]:
-        result = result.join(chunk, on=key_cols, how="full", suffix="_r")
-        for col_name in key_cols:
-            r_col = f"{col_name}_r"
-            if r_col in result.columns:
-                result = result.with_columns(
-                    nw.coalesce(col_name, r_col).alias(col_name)
-                ).drop(r_col)
-    return result
+        agg_exprs.append(
+            nw.when(seq == str(i))
+            .then(nw.col("clm_prcdr_prfrm_dt"))
+            .otherwise(nw.lit(None).cast(nw.Date))
+            .max()
+            .alias(f"procedure_date_{i}")
+        )
+    return df.group_by(key_cols).agg(*agg_exprs)
 
 
 # ── Stage 6: Institutional medical_claim ─────────────────────────
@@ -494,12 +480,11 @@ def int_institutional_medical_claim(
         nw.col("current_bene_mbi_id").alias("member_id"),
         nw.lit("medicare").alias("payer"),
         nw.lit("medicare").alias("plan"),
-        _nullify_sentinel("clm_from_dt"),
-        nw.col("clm_from_dt").alias("claim_start_date"),
+        _nullify_sentinel("clm_from_dt").alias("claim_start_date"),
         nw.col("clm_thru_dt").alias("claim_end_date"),
         nw.col("clm_line_from_dt").alias("claim_line_start_date"),
         nw.col("clm_line_thru_dt").alias("claim_line_end_date"),
-        nw.col("clm_from_dt").alias("admission_date"),
+        _nullify_sentinel("clm_from_dt").alias("admission_date"),
         nw.col("clm_thru_dt").alias("discharge_date"),
         nw.col("clm_admsn_src_cd").alias("admit_source_code"),
         nw.col("clm_admsn_type_cd").alias("admit_type_code"),
@@ -518,9 +503,7 @@ def int_institutional_medical_claim(
         # DRG: rightmost 3 chars
         nw.col("dgns_drg_cd").cast(nw.String).str.slice(-3).alias("drg_code"),
         nw.col("clm_line_prod_rev_ctr_cd").alias("revenue_center_code"),
-        nw.col("clm_line_srvc_unit_qty_rev")
-        .cast(nw.Int64)
-        .alias("service_unit_quantity"),
+        nw.col("clm_line_srvc_unit_qty").cast(nw.Int64).alias("service_unit_quantity"),
         nw.col("clm_line_hcpcs_cd").alias("hcpcs_code"),
         nw.col("hcpcs_1_mdfr_cd").alias("hcpcs_modifier_1"),
         nw.col("hcpcs_2_mdfr_cd").alias("hcpcs_modifier_2"),

tests/aco/test_express_cclf.py

@@ -1369,7 +1369,7 @@ class TestIntInstitutionalMedicalClaim:
                 "clm_line_from_dt": ["2023-01-01"],
                 "clm_line_thru_dt": ["2023-01-10"],
                 "clm_line_prod_rev_ctr_cd": ["0120"],
-                "clm_line_srvc_unit_qty_rev": ["5"],
+                "clm_line_srvc_unit_qty": ["5"],
                 "clm_line_hcpcs_cd": ["99213"],
                 "hcpcs_1_mdfr_cd": ["25"],
                 "hcpcs_2_mdfr_cd": [None],
@@ -1538,8 +1538,9 @@ class TestIntInstitutionalMedicalClaim:
             cclf___int_diagnosis_pivot=dx_pivot,
             cclf___int_procedure_pivot=px_pivot,
         )
-        # The column produced by _nullify_sentinel should be null
-        assert result["clm_from_dt"].to_list() == [None]
+        # Sentinel clm_from_dt should be nullified in claim_start_date
+        assert result["claim_start_date"].to_list() == [None]
+        assert result["admission_date"].to_list() == [None]
 
 
 # ── medical_claim (final union) ─────────────────────────────

tests/aco/test_transpile_databricks.py

@@ -107,19 +107,9 @@ def _flat_cases(all_transpiled):
 # 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.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",
-}
+_KNOWN_TRANSPILE_FAILURES: set[str] = set()
 
-# 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",
-}
+_KNOWN_PARSE_FAILURES: set[str] = set()
 
 
 class TestTranspileSuccess: