feat: filter reindex rows with a polars regex op

src/tablassert/lib.py

@@ -1070,6 +1070,31 @@ def reindex(df: pl.LazyFrame, col: str, op: Callable, comp: str | int | float, c
     return df.filter(op(expr, comp))
 
 
+def reindex_regex(df: pl.LazyFrame, col: str, pattern: str, negate: bool = False) -> pl.LazyFrame:
+    """Reindex a LazyFrame by filtering rows on a polars regex match against a column.
+
+    Args:
+        df: Source LazyFrame.
+        col: Column to test.
+        pattern: Polars-compatible regex, matched unanchored and case-sensitively (write `^`/`$`
+            to pin a whole cell).
+        negate: When True keep the rows that do NOT match (the `ne` filter); when False keep the
+            matches (the `eq` filter).
+
+    Returns:
+        LazyFrame containing only rows whose predicate is true.
+
+    Notes:
+        The column is cast to String first, so a typed source (parquet, or a numeric column in a
+        text table) filters on its text form instead of raising polars' ``InvalidOperationError:
+        expected String type``. A null cell matches nothing and ``not_()`` leaves it null, so null
+        rows drop under both directions -- the same null rule literal `eq`/`ne` already have, since
+        a null comparison is never true.
+    """
+    match: pl.Expr = pl.col(col).cast(pl.String).str.contains(pattern)
+    return df.filter(match.not_() if negate else match)
+
+
 def drop_not_significant(lf: pl.LazyFrame, col: str = "statistical_significance_qualifier") -> pl.LazyFrame:
     """Drop release-mode edges whose significance qualifier is ``not_significant``.
 
@@ -1339,7 +1364,12 @@ def _source_ops(self: Self) -> list[Any]:
             (crop, (self.source.row_slice,)) if self.source.row_slice else None,
             (pick, (self.source.rows,)) if self.source.rows else None,
             [
-                (reindex, (idxname(x.column), getattr(operator, x.comparison), x.comparator))
+                # `regex` is only legal with `eq`/`ne` (models.Reindex rejects it elsewhere), so the
+                # regex arm never needs the numeric cast or an `operator` lookup. Legacy entries keep
+                # emitting exactly the tuples they always did, so their run-cache op digests do not move.
+                (reindex_regex, (idxname(x.column), x.comparator, x.comparison == "ne"))
+                if x.regex
+                else (reindex, (idxname(x.column), getattr(operator, x.comparison), x.comparator))
                 if x.comparison not in ["ne", "eq"]
                 else (reindex, (idxname(x.column), getattr(operator, x.comparison), x.comparator, False))
                 for x in self.source.reindex
@@ -1588,6 +1618,7 @@ def collect(self: Self, db: Path) -> list[tuple[Callable, tuple[Any]]] | Path:
     crop: "filter",
     pick: "filter",
     reindex: "filter",
+    reindex_regex: "filter",
     head: "filter",
     coerce_columns: "clean",
     clean_numeric: "clean",

src/tablassert/runcache.py

@@ -258,6 +258,7 @@ def prefix_digest(ops: list[tuple[Callable, tuple[Any, ...]]]) -> str:
     lib.crop: 1,
     lib.pick: 1,
     lib.reindex: 1,
+    lib.reindex_regex: 1,
     # head samples min(HEAD_ROWS, height) rows, so it needs a height pass before slicing.
     lib.head: 2,
     # Encode: each is one lazy expression over a single column.

tests/test_cover_lib.py

@@ -20,6 +20,7 @@
 from tablassert.enums import Functions, Tokens
 from tablassert.ingests import from_yaml
 from tablassert.lib import Tcode
+from tablassert.runcache import prefix_digest
 
 
 def test_math_op_copysign_applies_values_token() -> None:
@@ -181,6 +182,118 @@ def test_reindex_filters_without_cast() -> None:
     assert result == ["a", "a"]
 
 
+def test_reindex_regex_keeps_matching_rows() -> None:
+    """``reindex_regex`` keeps only rows whose cell matches (the ``eq`` filter arm).
+
+    Why: the regex comparator exists so one filter drops a whole family of junk cells; an ``eq``
+    that matched nothing would emit an empty graph, and one that matched everything would keep the
+    ``N/A`` rows the config was written to remove.
+    """
+    lf: pl.LazyFrame = pl.LazyFrame({"c": ["N/A", "n/a", "ok1", "ok2"]})
+    assert lib.reindex_regex(lf, "c", r"(?i)^n/?a$").collect()["c"].to_list() == ["N/A", "n/a"]
+
+
+def test_reindex_regex_negate_keeps_non_matching_rows() -> None:
+    """``negate=True`` is the ``ne`` arm: matching rows drop, every other row survives."""
+    lf: pl.LazyFrame = pl.LazyFrame({"c": ["N/A", "n/a", "ok1", "ok2"]})
+    assert lib.reindex_regex(lf, "c", r"(?i)^n/?a$", negate=True).collect()["c"].to_list() == ["ok1", "ok2"]
+
+
+def test_reindex_regex_is_unanchored_without_markers() -> None:
+    """Polars ``str.contains`` is a search, so anchoring stays the config author's job.
+
+    Why: this is the documented contract shared with ``NodeEncoding.exclude_regex``. Assuming an
+    implicit full-cell match here would drop rows a reviewer cannot see being dropped.
+    """
+    lf: pl.LazyFrame = pl.LazyFrame({"c": ["N/A", "NA", "N/A2"]})
+    assert lib.reindex_regex(lf, "c", "^N/A$").collect()["c"].to_list() == ["N/A"]
+    assert lib.reindex_regex(lf, "c", "N/A").collect()["c"].to_list() == ["N/A", "N/A2"]
+
+
+def test_reindex_regex_compares_typed_columns_as_text() -> None:
+    """A numeric column filters on its cast string form instead of raising.
+
+    Why: polars ``str.contains`` on a Float64 column is an ``InvalidOperationError``, and a
+    self-describing ``parquet`` source can hand the pipeline typed columns, so the op casts to
+    String first and matches text like ``22.0``.
+    """
+    lf: pl.LazyFrame = pl.LazyFrame({"c": [1.5, 2.0, 22.0]})
+    assert lib.reindex_regex(lf, "c", r"^2\d?\.0$").collect()["c"].to_list() == [2.0, 22.0]
+
+
+def test_reindex_regex_drops_null_rows_in_both_directions() -> None:
+    """A null cell matches nothing, so ``eq`` drops it and ``ne`` (``not_``) drops it too.
+
+    Why: literal ``eq``/``ne`` already drop nulls (a null comparison is never true), so opting a
+    filter into regex must not quietly change which rows survive for a reason nobody documented.
+    """
+    lf: pl.LazyFrame = pl.LazyFrame({"c": ["ok", None, "junk"]})
+    assert lib.reindex_regex(lf, "c", "junk").collect()["c"].to_list() == ["junk"]
+    assert lib.reindex_regex(lf, "c", "junk", negate=True).collect()["c"].to_list() == ["ok"]
+
+
+def _reindex_ops(tmp_path: Path, fixtures_path: Path, reindex: list[dict[str, Any]]) -> list[tuple[Any, tuple[Any, ...]]]:
+    """Return just the row-filter ops a config emits, in pipeline order.
+
+    Args:
+        tmp_path: Scratch directory for the section store path.
+        fixtures_path: Repo fixture directory holding ``minimal_section.yaml``.
+        reindex: The ``source.reindex`` entries to attach to that section.
+
+    Returns:
+        The ``(callable, args)`` pairs for ``reindex``/``reindex_regex`` only.
+    """
+    data: Any = from_yaml(fixtures_path / "minimal_section.yaml")
+    tcode_model: Tcode = Tcode.model_validate(  # pyright: ignore
+        {
+            **data,
+            "source": {**data["source"], "reindex": reindex},
+            "config": fixtures_path / "minimal_section.yaml",
+            "store": tmp_path / "out.parquet",
+        }
+    )
+    # ``_source_ops`` returns nested groups (the reindex block is one list), and ``clean`` is the
+    # flattener the pipeline itself uses before dispatching ops.
+    return [op for op in tcode_model.clean(tcode_model._source_ops()) if op[0] in (lib.reindex, lib.reindex_regex)]
+
+
+def test_reindex_dispatch_keeps_legacy_ops_and_digests(tmp_path: Path, fixtures_path: Path) -> None:
+    """Backwards compatibility: a `reindex` entry without `regex` emits exactly the pre-feature op.
+
+    Why: run-cache keys are digests of these op tuples, so a legacy config whose tuple shape moved
+    would rebuild every section for no reason, and a config whose filter *semantics* moved would
+    change which rows survive a shipped build. Both are pinned against the literal tuples the code
+    emitted before `Reindex.regex` existed, including the ``cast=False`` slot ``eq``/``ne`` already
+    passed and the absent fourth slot the numeric operators never passed.
+    """
+    ops: list[tuple[Any, tuple[Any, ...]]] = _reindex_ops(
+        tmp_path, fixtures_path, [{"column": "A", "comparison": "ne", "comparator": "N/A"}, {"column": "C", "comparison": "lt", "comparator": 0.05}]
+    )
+    legacy: list[tuple[Any, tuple[Any, ...]]] = [
+        (lib.reindex, ("column_1", operator.ne, "N/A", False)),
+        (lib.reindex, ("column_3", operator.lt, 0.05)),
+    ]
+    assert ops == legacy
+    assert prefix_digest(ops) == prefix_digest(legacy), "a regex-less config must keep its pre-feature run-cache digest"
+
+
+def test_reindex_dispatch_emits_regex_op_per_comparison(tmp_path: Path, fixtures_path: Path) -> None:
+    """A `regex: true` entry emits `reindex_regex` with the negate flag derived from the comparison.
+
+    Why: the negate slot is what separates `ne` (drop the junk rows) from `eq` (keep only the junk
+    rows); a dispatch that mapped them alike would invert a section's row filter.
+    """
+    ops: list[tuple[Any, tuple[Any, ...]]] = _reindex_ops(
+        tmp_path,
+        fixtures_path,
+        [
+            {"column": "B", "comparison": "ne", "comparator": "^(?:n/?a|-{1,2})$", "regex": True},
+            {"column": "A", "comparison": "eq", "comparator": "^ok\\d$", "regex": True},
+        ],
+    )
+    assert ops == [(lib.reindex_regex, ("column_2", "^(?:n/?a|-{1,2})$", True)), (lib.reindex_regex, ("column_1", "^ok\\d$", False))]
+
+
 def test_tcode_collect_quick_exit_returns_existing_store(fixtures_path: Path, tmp_path: Path) -> None:
     """Line 749: ``collect`` returns the store path immediately when it already exists."""
     data: Any = from_yaml(fixtures_path / "minimal_section.yaml")

tests/test_production_patterns.py

@@ -346,6 +346,54 @@ def test_reindex_filters_rows_all_comparisons(
     assert {e["subject"] for e in edges} == expected_subjects
 
 
+@pytest.mark.parametrize(
+    ("comparison", "pattern", "driver_values", "expected_subjects"),
+    [
+        # One `ne` pattern replaces the per-spelling literal filters a shipped config would otherwise
+        # need for the same missing-value family.
+        pytest.param("ne", r"(?i)^n/?a$", ["N/A", "n/a", "ok1", "ok2"], {"HGNC:11998", "HGNC:3236"}, id="ne-junk-family"),
+        pytest.param("eq", r"^ok\d$", ["ok1", "ok2", "N/A", "NA"], {"HGNC:1100", "HGNC:6871"}, id="eq-shape"),
+        # Anchored: "NA" and "N/A2" survive a `^N/A$` exclusion, pinning that anchoring is opt-in.
+        pytest.param("ne", r"^N/A$", ["N/A", "NA", "N/A2", "ok"], {"HGNC:6871", "HGNC:11998", "HGNC:3236"}, id="ne-anchored"),
+    ],
+)
+def test_reindex_regex_filters_rows(
+    comparison: str,
+    pattern: str,
+    driver_values: list[str],
+    expected_subjects: set[str],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    rig_factory: Any,
+) -> None:
+    """``source.reindex`` with ``regex: true`` filters rows by a polars regex during a real build.
+
+    This is the feature's reason to exist: a source whose missing values are spelled several ways
+    (``N/A``, ``n/a``) drops all of them with one filter instead of one literal ``ne`` per
+    spelling. It also proves the dispatch: a regex entry must emit the ``reindex_regex`` op, while
+    the ``regex``-less entries above keep emitting ``reindex``.
+    """
+    fullmap: Path = _build_rich_redb(tmp_path / "fullmap")
+    data: Path = tmp_path / "data.tsv"
+    rows: list[str] = [f"{subject}\t{driver}" for subject, driver in zip(["brca1", "mapk1", "tp53", "egfr"], driver_values, strict=True)]
+    data.write_text("\n".join(rows) + "\n")
+
+    config: dict[str, Any] = _table_config(
+        data,
+        {
+            "subject": {"method": "column", "encoding": "A"},
+            "predicate": "associated_with",
+            "object": {"method": "value", "encoding": "MONDO:0004979"},
+        },
+    )
+    config["template"]["source"]["reindex"] = [{"column": "B", "comparison": comparison, "comparator": pattern, "regex": True}]
+
+    _nodes, edge_text = _run_build(tmp_path, monkeypatch, rig_factory, config, fullmap, "REINDEX_REGEX_KG")
+
+    edges: list[dict[str, Any]] = _parse_edges(edge_text)
+    assert {e["subject"] for e in edges} == expected_subjects
+
+
 def test_regex_backref_replacement(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, rig_factory: Any) -> None:
     """``${1}`` capture-group backrefs and qualifier regex mapping survive the pipeline.