diff --git a/CHANGELOG.md b/CHANGELOG.md index caa921f..ee7e957 100644 --- a/CHANGELOG.md +++ b/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 `` 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_` 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 diff --git a/src/tablassert/fullmap.py b/src/tablassert/fullmap.py index 5e572ba..dc23540 100644 --- a/src/tablassert/fullmap.py +++ b/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( diff --git a/src/tablassert/lib.py b/src/tablassert/lib.py index c75475a..5b916b0 100644 --- a/src/tablassert/lib.py +++ b/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_`` 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_`` 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_`` 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", }