fix(dataflow): cache short-circuit in _summary_sd for warm filt scope (#814)

tests/unit/dataflow/sd_cache_test.py

@@ -14,6 +14,7 @@
     DropCol, FillNA, GroupBy, NoOp, SafeInt, Search)
 from buckaroo.customizations.pd_autoclean_conf import NoCleaningConf
 from buckaroo.dataflow.autocleaning import AutocleaningConfig, PandasAutocleaning
+from buckaroo.dataflow.dataflow import CustomizableDataflow, StylingAnalysis
 from buckaroo.dataflow.sd_cache import hash_chain, split_chain_by_scope
 from buckaroo.jlisp.lisp_utils import s, sA, sQ
 from buckaroo.pluggable_analysis_framework.col_analysis import ColAnalysis
@@ -117,3 +118,143 @@ def test_filter_flip_only_grows_filt_entry(dirty_df):
     assert len(df.summary_stats_cache) == cache_size_before + 1
     assert raw_before in df.summary_stats_cache
     assert clean_before in df.summary_stats_cache
+
+
+class _CountingDataflow(CustomizableDataflow):
+    """CustomizableDataflow subclass that records every ``_get_summary_sd``
+    call by the row-count of the df it was passed.
+
+    Use this to assert that ``_summary_sd`` (which calls
+    ``_get_summary_sd`` on ``processed_df``) hits the cache on a
+    warm-cache state_change instead of recomputing. The raw/clean
+    scopes are populated through a separate call path inside
+    ``_populate_sd_cache`` — those calls also land here but are easy
+    to distinguish by their row-count (raw/clean run on the full
+    ``sampled_df``, the filt scope runs on the filtered
+    ``processed_df``).
+    """
+    autocleaning_klass = PandasAutocleaning
+    autoclean_conf = tuple([_Conf, NoCleaningConf])
+    analysis_klasses = [StylingAnalysis, DefaultSummaryStats]
+
+    def __init__(self, *args, **kwargs):
+        self.summary_sd_calls = []
+        super().__init__(*args, **kwargs)
+
+    def _get_summary_sd(self, df):
+        try:
+            self.summary_sd_calls.append(len(df))
+        except Exception:
+            self.summary_sd_calls.append(-1)
+        return super()._get_summary_sd(df)
+
+
+def test_warm_filt_cache_skips_get_summary_sd_on_state_change(dirty_df):
+    """Issue #814 regression.
+
+    A state_change that re-applies a previously-computed filter must
+    NOT call ``_get_summary_sd`` through ``_summary_sd`` again — the
+    filt scope's cached entry from the first application must be
+    reused.
+
+    Cycle: filter=abc → clear → filter=abc. The third state-change
+    must not run ``_get_summary_sd`` on any df with the filt scope's
+    row count (the only "new compute" the cache is supposed to skip).
+
+    Currently fails because ``_summary_sd`` reads ``self.operations``
+    for its cache key, but ``self.operations`` is the PRIOR state's
+    chain at the moment ``_summary_sd`` fires (during
+    ``self.cleaned = result`` — before
+    ``self.operations = result[3]``). So the cache lookup never sees
+    the new chain's entry — actually, today there is no cache lookup
+    at all, so the call always happens. The fix re-introduces the
+    lookup but keys it off ``self.merged_operations`` (== the freshly
+    set ``self.cleaned[3]``) so the right entry is found.
+    """
+    dfc = _CountingDataflow(dirty_df, debug=False)
+
+    # Apply filter the first time — populates filt_key_abc.
+    dfc.quick_command_args = {'search': ['10']}
+    filt_rows_first_apply = len(dfc.processed_df)
+    raw_rows = len(dfc.sampled_df)
+    assert filt_rows_first_apply < raw_rows, (
+        "precondition: search should have reduced rows"
+    )
+
+    # Clear filter — back to empty-filter chain (cache hit from init).
+    dfc.quick_command_args = {}
+
+    calls_before_replay = list(dfc.summary_sd_calls)
+
+    # Replay the same filter. This MUST be a cache hit in _summary_sd —
+    # no _get_summary_sd call on the filtered (smaller-row) df.
+    dfc.quick_command_args = {'search': ['10']}
+
+    new_calls = dfc.summary_sd_calls[len(calls_before_replay):]
+    filt_scope_calls = [n for n in new_calls if n == filt_rows_first_apply]
+    assert filt_scope_calls == [], (
+        f"warm-cache filter replay must skip _get_summary_sd for the "
+        f"filt scope (row-count={filt_rows_first_apply}). Saw "
+        f"{len(filt_scope_calls)} call(s) — _summary_sd missed the "
+        f"cache and recomputed. New calls in this state_change: "
+        f"{new_calls}."
+    )
+
+
+def test_summary_sd_uses_new_state_chain_not_prior():
+    """Regression for the cascade-ordering bug that motivated the
+    original removal of the ``_summary_sd`` cache lookup
+    (commit 5bc7bbfb).
+
+    If ``_summary_sd`` keys off ``self.operations`` instead of the
+    fresh chain in ``self.cleaned[3]``, then on a state_change the
+    cached entry it returns corresponds to the PRIOR state's chain —
+    so ``summary_sd`` ends up labelled with the new state but holds
+    the prior state's data.
+
+    Construct the mislabel scenario:
+      1. Apply search 'foo' — populates filt_key_FOO with SD_foo
+         (computed on 3 'foo' rows).
+      2. Apply search 'bar' — populates filt_key_BAR with SD_bar
+         (computed on 1 'bar' row).
+
+    After step 2, the filt cache slot MUST hold SD_bar. If the bug
+    were present, the cache lookup at step 2 would key off the prior
+    state's chain (still in ``self.operations``), find filt_key_FOO,
+    and assign that to ``summary_sd`` — which ``_populate_sd_cache``
+    would then write under filt_key_BAR. Reading filt_key_BAR back
+    would yield 3-row stats, not 1-row.
+    """
+    df = pd.DataFrame({'a': [10, 20, 30, 40, 50],
+                       'b': ['foo', 'bar', 'foo', 'baz', 'foo']})
+    dfc = _CountingDataflow(df, debug=False)
+
+    dfc.quick_command_args = {'search': ['foo']}
+    foo_rows = len(dfc.processed_df)
+    assert foo_rows == 3, (
+        f"precondition: search 'foo' should match 3 rows, got {foo_rows}"
+    )
+
+    dfc.quick_command_args = {'search': ['bar']}
+    bar_rows = len(dfc.processed_df)
+    assert bar_rows == 1, (
+        f"precondition: search 'bar' should match 1 row, got {bar_rows}"
+    )
+
+    cached_filt = dfc.summary_stats_cache[dfc.filt_sd_key]
+    assert cached_filt is not None
+    # The processed_df has 1 row; any column-level length stat should
+    # reflect that.
+    saw_length_stat = False
+    for col, stats in cached_filt.items():
+        if 'length' in stats:
+            saw_length_stat = True
+            assert stats['length'] == bar_rows, (
+                f"cached filt SD for column {col!r} reports length="
+                f"{stats['length']}; expected {bar_rows} (current "
+                f"'bar'-filtered df). A wrong length means _summary_sd "
+                f"reused the prior state's cache entry."
+            )
+    assert saw_length_stat, (
+        "precondition: at least one column should have a `length` stat"
+    )