Pipeline Collect Barriers — Full Analysis¶

Current State¶

The pipeline has .collect() calls across several engine files. The hot-path collects (every calculation run) are concentrated in CRM processing and the pipeline's split-once + collect_all pattern. The pipeline cannot currently run as a single LazyFrame operation due to three categories of blockers.

Category 1: Polars Optimizer Limitations (Hardest to Fix)¶

These are platform-level constraints in Polars itself:

Why they exist: Polars' lazy engine lacks robust CSE (Common Subexpression Elimination) for deep plan trees. When the same LazyFrame is referenced by multiple downstream consumers (fan-out pattern), the optimizer either re-executes the shared upstream per consumer or segfaults on very deep plans.

What would fix them upstream: Polars would need:

1. Reliable CSE that handles deep plan trees without re-optimization per branch
2. Streaming engine CSE support
3. Optimizer stability with very deep plans (no segfaults)

Category 2: Graph Traversal Algorithm (Design Choice)¶

Why they exist: The hierarchy resolver needs to walk parent→child chains of arbitrary depth to find ultimate roots (counterparty and facility). This is inherently an iterative/recursive algorithm that Polars expressions can't express natively.

Data size: Small (unique org/facility edges — typically <1,000 rows). Performance impact is negligible.

Alternatives:

• Iterative self-join — Polars join in a Python loop up to max_depth times. Stays "lazier" but still requires a fixed iteration count and builds a very wide plan.
• Accept as-is — The data is small and early in the pipeline. This is the lowest-impact collect in the entire chain.

Category 3: Validation & Edge Cases (Non-blocking)¶

These are all either off the hot path or trivially small.

Architecture Diagram — Where Laziness Breaks¶

LAZY ──────────────────────────────────────────────────────────
  Loader (scan_parquet)
    │
  HierarchyResolver
    │  ├── COLLECT: graph edges (~1K rows) ← Category 2
    │  └── all joins/enrichments stay lazy
    │
  Classifier (fully lazy, schema checks only)
    │
  CRM: provisions → CCF → init_ead
    │
EAGER ─── COLLECT #1: post-init_ead flatten ── Category 1 ────
    │
LAZY ──────────────────────────────────────────────────────────
  CRM: build 3 lookup tables (group_by)
    │
EAGER ─── COLLECT #2: 3 small lookups ──── Category 1 ────────
    │
LAZY ──────────────────────────────────────────────────────────
  CRM: collateral allocation, guarantees, finalize_ead
    │   (no final CRM collect — plan tree is shallow post-collateral)
    │
  Pipeline: _run_single_pass()
    │
EAGER ─── COLLECT #3: pre-branch flatten ── Category 1 ───────
    │   (pipeline.py:631 — materialise CRM output before split)
    │
LAZY ──────────────────────────────────────────────────────────
  SA calculate_unified() (Basel 3.1 only — SA-equiv RW on all rows)
  Split once by approach
  ├── SA calculator (lazy)
  ├── IRB calculator (lazy)
  └── Slotting calculator (lazy)
    │
EAGER ─── COLLECT #4: collect_all(3 branches) ─────────────────
    │   (pipeline.py:665 — CPU engine, not streaming)
    │
  Aggregator (re-lazify for summaries, then final output)

Path to Fewer Collects¶

Out-of-Core Support (Implemented)¶

All hot-path collects now go through engine/materialise.py, which selects strategy based on config.collect_engine:

Streaming mode caps peak memory to approximately one column-batch at a time by spilling intermediate results to temp parquet files. This enables the pipeline to process datasets larger than available RAM.

Config options: - collect_engine: "streaming" (default) — disk-spill for out-of-core support - collect_engine: "cpu" — in-memory collect for backward compatibility - spill_dir: Path | None — directory for temp files (default: system temp)

Fallback: If sink_parquet fails for a particular expression (unsupported in streaming engine), the barrier falls back to in-memory .collect().lazy().

Cleanup: Temp files are cleaned up via cleanup_spill_files() in pipeline.run_with_data()'s finally block, plus an atexit safety net.

Key Takeaway¶

Most hot-path collects exist because of Polars optimizer limitations (deep plan re-execution, no streaming CSE). The remaining 2 are algorithmic (graph traversal on small data). The materialization barriers are now strategy-aware via engine/materialise.py, supporting both in-memory and disk-spill modes. The realistic path forward is:

1. Short term (done): Disk-spill materialization via materialise_barrier / materialise_branches enables out-of-core processing for any dataset size.
2. Medium term: Monitor Polars' LazyFrame.cache() and CSE roadmap. When available, the 3 lookup collects and the pre-branch collect can likely be replaced.
3. Long term: A single collect_all() at the output boundary becomes feasible only when Polars can handle deep plan trees with fan-out without re-execution or segfaults.