perf: materialize sections once in resolve_batch and trim before resolution

CHANGELOG.md

@@ -8,6 +8,9 @@ All notable changes to this project are documented in this file.
 - **The final graph QC now asserts no field in the emitted NDJSON is null or empty, and that every node carries a name.** `build-kg --qc`'s stage-7 study pass (in the spirit of `studyKGtsvs.pl`) gained two assertions. `empty-or-null-values`: any field in the nodes or edges file whose value is JSON `null`, a string that strips to empty, or an empty container — checked recursively, so a null or blank nested inside an `attributes` list counts — fails the build. The check is deliberately stricter than the NDJSON writer's `strip_nulls` (`rust/src/json.rs`), which scrubs dict entries at every depth but passes array scalars (`["x", ""]`) and emptied nested objects (`[{}]`) through verbatim; the study stage now asserts the stronger contract — no null or empty value anywhere — so the first such shape to reach an emitted file fails the build loudly instead of shipping silently. Null-like *strings* (`NA`/`NaN`/`null`/`none`) are also dropped by the writer but are neither null nor empty, and are deliberately not flagged; the `original_*` whitespace exemption does not extend to emptiness. `unnamed-nodes`: a node record whose `name` key is missing, `null`, or strips to empty fails the build — the missing-key case is what pipeline output surfaces, since `strip_nulls` deletes empty and null-like names before the file is written. Both report offenders per field or per node id, capped at 10 examples like the other assertions.
 - **The final graph QC now asserts every node has an `id` and every edge has `subject`, `predicate`, and `object`.** Two more stage-7 study assertions join the `unnamed-nodes` check (which already requires a non-empty node `name`). `unidentified-nodes`: a node record whose `id` key is missing, `null`, or strips to empty fails the build; since the id is exactly what is absent, examples key on the node's `name` (or `<no name>` when it has none). `incomplete-edges`: an edge record missing any of the three core slots — missing key, `null`, or strips-to-empty — fails, counted per slot (e.g. `predicate (2)`). The writer's `strip_nulls` deletes a null slot outright rather than emitting it, so on pipeline output a hit means the slot was null upstream and the record shipped broken — exactly the condition these assertions exist to catch loudly. Non-string, non-null ids and slots pass, mirroring the name convention (no writer emits them).
 
+### Performance
+- **Entity resolution now materializes each section once instead of once per node column.** `resolve_batch` used to re-execute the entire upstream lazy plan — the source scan, every encoding/regex op, and both NLP normalization levels — once per resolved column: `distinct(...).collect()` per column, `join_matches`' collect per column, and `log_unmatched`'s anti-join collect when `--log` is on, so a section resolving subject + object + two qualifiers ran that pipeline roughly five times (up to nine with logging). The batch now collects the frame a single time and runs term extraction, unmatched logging, and the level-one/level-two join-backs against the in-memory frame; `join_matches` keeps its LazyFrame-in/LazyFrame-out contract over a new eager core, and output rows are unchanged. On a 50k-row synthetic section with four resolved columns the resolve phase runs about 1.75× faster, with the gain growing as the upstream plan gets heavier. Separately, `trim` — the op that drops the spent raw `column_<n>` columns — moved from the finalize ops to just before `resolve_batch`, so the materialized frame and every downstream join no longer carry columns nothing reads; its progress phase label falls back to `transform` instead of flashing `finalize` early.
+
 ## 12.1.0 - 2026-08-18
 
 ### Added

src/tablassert/fullmap.py

@@ -499,14 +499,31 @@ def join_matches(lf: pl.LazyFrame, col: str, matches: pl.DataFrame, tag: str = "
 
     Notes:
         Split out of ``resolve`` so ``resolve_batch`` can apply per-column
-        matches from one shared redb fetch.
+        matches from one shared redb fetch. ``resolve_batch`` calls the eager
+        twin directly so the whole batch materializes the frame only once.
+    """
+    # Collection point: join after redb query, then re-lazy.
+    return _join_matches_eager(lf.collect(), col, matches, tag, drop_unresolved).lazy()
+
+
+def _join_matches_eager(df: pl.DataFrame, col: str, matches: pl.DataFrame, tag: str = "_two", drop_unresolved: bool = True) -> pl.DataFrame:
+    """Eager core of ``join_matches``: join ranked matches into an in-memory frame.
+
+    Args:
+        df: Source DataFrame, already materialized.
+        col: Column being resolved.
+        matches: Ranked matches for this column from ``filter_and_rank``.
+        tag: Suffix used to derive the level-two column name.
+        drop_unresolved: When True rows whose ``col`` did not match are dropped;
+            when False the row is kept and the resolved columns stay null.
+
+    Returns:
+        DataFrame with resolved columns; rows whose ``col`` did not match are
+        dropped unless ``drop_unresolved`` is False.
     """
-    # Split out of resolve so resolve_batch can apply per-column matches from one shared redb fetch.
     l1: str = col
     l2: str = l1 + tag
 
-    # Collection point: join after redb query, then re-lazy.
-    df: pl.DataFrame = lf.collect()
     result: pl.DataFrame = df.join(matches.filter(pl.col("NLP_LEVEL").eq(1)), left_on=l1, right_on="term", how="left", suffix="_l1")
 
     l2_matches: pl.DataFrame = matches.filter(pl.col("NLP_LEVEL").eq(2))
@@ -522,7 +539,7 @@ def join_matches(lf: pl.LazyFrame, col: str, matches: pl.DataFrame, tag: str = "
         # nullable qualifier keeps the edge and leaves the column null for the null-stripper.
         result = result.filter(pl.col(col).is_not_null())
 
-    return result.lazy()
+    return result
 
 
 class ResolveSpec(NamedTuple):
@@ -616,27 +633,32 @@ def resolve_batch(
             code="resolve-bad-specs",
         )
 
-    terms_by_col: dict[str, pl.LazyFrame] = {spec.col: distinct(lf, spec.col, spec.col + tag) for spec in specs}
-    collected_terms: dict[str, pl.DataFrame] = {col: terms.collect() for col, terms in terms_by_col.items()}
+    # Single collection point: the upstream plan (scan, encodings, NLP normalization)
+    # executes exactly once here. Per-column term extraction, unmatched logging, and
+    # the join backs all run against this in-memory frame, instead of re-executing
+    # the whole lazy plan once per column as separate collects.
+    df: pl.DataFrame = lf.collect()
+
+    terms_by_col: dict[str, pl.DataFrame] = {spec.col: distinct(df.lazy(), spec.col, spec.col + tag).collect() for spec in specs}
 
-    union_terms: list[str] = pl.concat([t.select("term") for t in collected_terms.values()]).unique().get_column("term").to_list()
+    union_terms: list[str] = pl.concat([t.select("term") for t in terms_by_col.values()]).unique().get_column("term").to_list()
 
     rows: list[dict[str, object]] = lookup_rows(db, union_terms, threads=threads) if union_terms else []
     raw: pl.DataFrame = pl.DataFrame(rows)
 
-    result: pl.LazyFrame = lf
+    result: pl.DataFrame = df
     for spec in specs:
         if on_phase is not None:
             on_phase(f"resolve:{spec.col}")
-        terms_df: pl.DataFrame = collected_terms[spec.col]
+        terms_df: pl.DataFrame = terms_by_col[spec.col]
         matches: pl.DataFrame = filter_and_rank(
             raw, terms_df, spec.taxon, spec.prioritize, spec.avoid, column_context, spec.exclude_prefixes, spec.exclude_regex
         )
         if log:
-            log_unmatched(spec.col, terms_by_col[spec.col], matches, section_hash, config_file)
-        result = join_matches(result, spec.col, matches, tag, drop_unresolved=not spec.nullable)
+            log_unmatched(spec.col, terms_df.lazy(), matches, section_hash, config_file)
+        result = _join_matches_eager(result, spec.col, matches, tag, drop_unresolved=not spec.nullable)
 
-    return result
+    return result.lazy()
 
 
 def resolve(

src/tablassert/lib.py

@@ -1089,9 +1089,10 @@ def _node_ops(self: Self, db: Path) -> list[Any]:
             db: Path to the fullmap redb used for entity resolution.
 
         Returns:
-            Raw op list: one ``node_prep`` block per node column, the single
-            shared ``resolve_batch`` op, then per-column ``fullmap_audit`` ops
-            when QC is enabled.
+            Raw op list: one ``node_prep`` block per node column, ``trim`` to drop
+            the spent raw ``column_<n>`` columns, the single shared
+            ``resolve_batch`` op, then per-column ``fullmap_audit`` ops when QC is
+            enabled.
         """
         # Subject/object/qualifiers share one resolve_batch call instead of one per column.
         # Enum-ranged qualifiers are excluded from resolution: their range is a closed
@@ -1124,6 +1125,10 @@ def _node_ops(self: Self, db: Path) -> list[Any]:
             # Encode only: no pre-resolution snapshot and no NLP normalization, both of
             # which exist to feed entity resolution these columns never undergo.
             [self.encoding(x, x.qualifier) for x in literals],
+            # Raw ``column_<n>`` columns are dead weight once the encodings above have
+            # copied them into named slots; trim before resolution so the frame that
+            # resolve_batch materializes and joins stays narrow.
+            (trim, ()),
             # ``"_two"`` is spelled explicitly (it is ``resolve_batch``'s own default tag)
             # only so ``threads`` can follow positionally: ``compile_subgraph`` applies op
             # args positionally (``on_phase`` arrives separately as a keyword).
@@ -1144,7 +1149,9 @@ def _provenance_ops(self: Self) -> list[Any]:
 
         Returns:
             Raw op list: predicate and edge category, provenance metadata,
-            then the trim/format/write finalize ops.
+            then the format/write finalize ops. (``trim`` now runs earlier, in
+            ``_node_ops``, so resolution joins never carry the spent raw
+            ``column_<n>`` columns.)
         """
         override = self.provenance.override
         # The edge primary knowledge source is ALWAYS the explicit graph-level infores
@@ -1186,7 +1193,6 @@ def _provenance_ops(self: Self) -> list[Any]:
             # Prune first so class-rejected values are handed to the study rather than lost.
             (prune_to_class, ()),
             (inline_supporting_study, (study_id, sheet, identified)),
-            (trim, ()),
             (format_numeric, ()),
             (to_store, (self.store, self.config.name)),
         ]
@@ -1240,7 +1246,6 @@ def collect(self: Self, db: Path) -> list[tuple[Callable, tuple[Any]]] | Path:
     sig: "significance",
     drop_not_significant: "significance",
     drop_zero_effect_size: "significance",
-    trim: "finalize",
     format_numeric: "finalize",
     to_store: "write",
 }