From 926326ce6218c130be71cde5c62899de486f3db0 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 3 Oct 2026 22:51:32 -0400 Subject: [PATCH 01/14] test(dataflow): characterize widget trait order and merged_sd before the _handle_widget_change split (rows-first s1) A default-tier XorqBuckarooWidget and BuckarooWidget publish df_data_dict, then df_display_args, then the rest of the widget_args_tuple observers on a search change, and merged_sd carries the full stat set. These pass on main and pin the behaviour the split must keep. Co-Authored-By: Claude Sonnet 5.5 --- tests/unit/basic_widget_test.py | 35 +++++++++++++ tests/unit/test_xorq_buckaroo_widget.py | 69 +++++++++++++++++++++++++ 2 files changed, 104 insertions(+) diff --git a/tests/unit/basic_widget_test.py b/tests/unit/basic_widget_test.py index a028751ed..9e63765e1 100644 --- a/tests/unit/basic_widget_test.py +++ b/tests/unit/basic_widget_test.py @@ -288,6 +288,41 @@ def counting_handle(*args, **kwargs): ) +def test_default_tier_widget_change_sequence_and_merged_sd(): + """Characterization for the ``_handle_widget_change`` split (rows-first + s1): a default-tier pandas widget keeps the order in which a search change + reaches the dataflow's and widget's traits, and its merged_sd.""" + bw = BuckarooWidget(pd.DataFrame({ + 'price': [12.5, 18.9, 7.4, 22.1, 14.0], 'qty': [1, 2, 1, 3, 2], + 'category': ['a', 'b', 'a', 'c', 'b']})) + msd = bw.dataflow.merged_sd + assert {col: (sd['length'], sd['_type'], sd['dtype']) for col, sd in msd.items()} == { + 'a': (5, 'float', 'float64'), 'b': (5, 'integer', 'int64'), + 'c': (5, 'string', 'object')} + assert msd['a']['mean'] == pytest.approx(14.98) + + seen = [] + bw.dataflow.observe(lambda c: seen.append(('dataflow', c['name'])), names=[ + 'summary_sd', 'merged_sd', 'widget_args_tuple', 'df_data_dict', + 'df_display_args', 'df_meta', 'operations', 'raw_sd_key', 'clean_sd_key', + 'filt_sd_key', 'summary_stats_cache', 'processed_result', 'cleaned']) + bw.observe(lambda c: seen.append(('widget', c['name'])), + names=['df_data_dict', 'df_display_args', 'df_meta', 'operations']) + bw.dataflow.quick_command_args = {'search': ['a']} + assert seen == [ + ('widget', 'df_data_dict'), ('dataflow', 'df_data_dict'), + ('widget', 'df_display_args'), ('dataflow', 'df_display_args'), + ('dataflow', 'widget_args_tuple'), ('dataflow', 'merged_sd'), + ('dataflow', 'summary_sd'), ('dataflow', 'processed_result'), + ('widget', 'df_meta'), ('dataflow', 'df_meta'), ('dataflow', 'cleaned'), + ('dataflow', 'summary_stats_cache'), ('dataflow', 'widget_args_tuple'), + ('dataflow', 'merged_sd'), ('dataflow', 'filt_sd_key'), + ('widget', 'operations'), ('dataflow', 'operations')] + assert {col: (sd['length'], sd['filtered_length']) + for col, sd in bw.dataflow.merged_sd.items()} == { + 'a': (5, 2), 'b': (5, 2), 'c': (5, 2)} + + def test_multi_index_cols() -> None: df = get_multiindex_cols_df() bw = BuckarooWidget(df) diff --git a/tests/unit/test_xorq_buckaroo_widget.py b/tests/unit/test_xorq_buckaroo_widget.py index 196175748..9e8f4486f 100644 --- a/tests/unit/test_xorq_buckaroo_widget.py +++ b/tests/unit/test_xorq_buckaroo_widget.py @@ -60,6 +60,75 @@ def counting(self, q): ) +_CHANGE_ORDER_TRAITS = [ + "summary_sd", "merged_sd", "widget_args_tuple", "df_data_dict", + "df_display_args", "df_meta", "operations", "raw_sd_key", "clean_sd_key", + "filt_sd_key", "summary_stats_cache", "processed_result", "cleaned"] + +# Order in which the dataflow's observers see a search change on a default +# (full-tier) XorqBuckarooWidget, recorded before ``_handle_widget_change`` is +# split into separately callable halves. The split must leave it unchanged: +# df_data_dict, then df_display_args, then widget_args_tuple's other observers. +_SEARCH_CHANGE_ORDER = [ + ("widget", "df_data_dict"), ("dataflow", "df_data_dict"), + ("widget", "df_display_args"), ("dataflow", "df_display_args"), + ("dataflow", "widget_args_tuple"), ("dataflow", "merged_sd"), + ("dataflow", "summary_sd"), ("dataflow", "processed_result"), + ("widget", "df_meta"), ("dataflow", "df_meta"), ("dataflow", "cleaned"), + ("dataflow", "summary_stats_cache"), ("dataflow", "widget_args_tuple"), + ("dataflow", "merged_sd"), ("dataflow", "filt_sd_key"), + ("widget", "operations"), ("dataflow", "operations")] + +_FULL_TIER_STAT_KEYS = { + "a": ["_type", "distinct_count", "distinct_per", "dtype", "histogram", + "histogram_bins", "is_bool", "is_datetime", "is_float", "is_integer", + "is_numeric", "is_string", "length", "max", "mean", "median", "min", + "nan_per", "non_null_count", "null_count", "orig_col_name", + "rewritten_col_name", "std"]} + + +class TestWidgetChangeSequence: + """Characterization of what a default-tier widget publishes and in what + order, so the ``_handle_widget_change`` split (rows-first s1) can be shown + to leave a full-stats widget unchanged.""" + + def _widget(self): + return XorqBuckarooWidget(xo.memtable( + {"price": [12.5, 18.9, 7.4, 22.1, 14.0], "qty": [1, 2, 1, 3, 2], + "category": ["a", "b", "a", "c", "b"]})) + + def test_search_change_keeps_trait_order(self): + w = self._widget() + seen = [] + w.dataflow.observe( + lambda c: seen.append(("dataflow", c["name"])), names=_CHANGE_ORDER_TRAITS) + w.observe( + lambda c: seen.append(("widget", c["name"])), + names=["df_data_dict", "df_display_args", "df_meta", "operations"]) + w.dataflow.quick_command_args = {"search": ["a"]} + assert seen == _SEARCH_CHANGE_ORDER + + def test_merged_sd_and_display_state_unchanged(self): + w = self._widget() + msd = w.dataflow.merged_sd + assert sorted(msd["a"]) == _FULL_TIER_STAT_KEYS["a"] + assert {col: (sd["length"], sd["_type"], sd["dtype"]) for col, sd in msd.items()} == { + "a": (5, "float", "float64"), "b": (5, "integer", "int64"), + "c": (5, "string", "string")} + assert msd["a"]["mean"] == pytest.approx(14.98) + assert list(w.df_data_dict) == ["main", "all_stats", "empty"] + main = w.df_display_args["main"] + assert [pr["primary_key_val"] for pr in main["df_viewer_config"]["pinned_rows"]] == [ + "dtype", "histogram"] + assert [cc["col_name"] for cc in main["df_viewer_config"]["column_config"]] == [ + "a", "b", "c"] + + w.dataflow.quick_command_args = {"search": ["a"]} + msd = w.dataflow.merged_sd + assert {col: (sd["length"], sd["filtered_length"]) for col, sd in msd.items()} == { + "a": (5, 2), "b": (5, 2), "c": (5, 2)} + + class TestExprCountMemoization: """Regression for #795: ``_expr_count`` issued a fresh ``expr.count().execute()`` against the backend on every call, even From 689aae8a98cc2abc9d759b6cc35f94b66d1d2ed9 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 3 Oct 2026 22:56:48 -0400 Subject: [PATCH 02/14] test(xorq): failing tests for the xorq schema stats tier and its load-handler fields (rows-first s1) A schema-tier XorqServerDataflow should match full stats on pinned_rows, data_key, summary_stats_key and (except the stats-derived minWidth) column_config, issue no data query besides the cached count, and keep init_sd hints and sorted windows working. A pending state must write nothing under a full-tier cache key, and a later full assignment must reach merged_sd for the raw, clean and filt scopes. assemble_merged_sd must equal merged_sd, and _handle_widget_change must be built from separately callable all_stats and display-args builders. /load_expr and /reload_expr accept stats_tier and stats_delivery, replay them on reload, and keep them out of the warm short-circuit's has_config tuple. Co-Authored-By: Claude Sonnet 5.5 --- .../dataflow/customizable_dataflow_test.py | 19 + .../dataflow/scoped_summary_stats_test.py | 67 +++ tests/unit/server/test_load_expr.py | 432 ++++++++++++++++++ 3 files changed, 518 insertions(+) diff --git a/tests/unit/dataflow/customizable_dataflow_test.py b/tests/unit/dataflow/customizable_dataflow_test.py index 09b36eb70..909302c7a 100644 --- a/tests/unit/dataflow/customizable_dataflow_test.py +++ b/tests/unit/dataflow/customizable_dataflow_test.py @@ -75,6 +75,25 @@ def test_widget_instatiation(): assert dfc.df_data_dict['main'] == BASIC_DF_JSON_DATA assert dfc.df_display_args['main']['df_viewer_config'] == DFVIEWER_CONFIG_DEFAULT +def test_all_stats_and_display_args_are_built_separately(monkeypatch): + """``_handle_widget_change`` is the composition of separately callable + builders (rows-first s1): the stats payload can be rebuilt without + rebuilding column_config, and column_config without serializing stats.""" + dfc = ACDFC(BASIC_DF) + _unused, processed_df, merged_sd = dfc.widget_args_tuple + serialized = [] + original = dfc._sd_to_jsondf + monkeypatch.setattr(dfc, '_sd_to_jsondf', lambda sd: serialized.append(sd) or original(sd)) + + display_args = dfc._build_df_display_args(processed_df, merged_sd) + assert serialized == [], "building display args must not serialize all_stats" + assert display_args == dfc.df_display_args + + data_dict = dfc._build_df_data_dict(processed_df, merged_sd) + assert serialized == [merged_sd] + assert data_dict == dfc.df_data_dict + + def test_widget_operations_instatiation(): dfc = ACDFC(BASIC_DF) # dfc starts with operations of [{'meta':'no-op'], but the first diff --git a/tests/unit/dataflow/scoped_summary_stats_test.py b/tests/unit/dataflow/scoped_summary_stats_test.py index b5648ab56..a8b0a5339 100644 --- a/tests/unit/dataflow/scoped_summary_stats_test.py +++ b/tests/unit/dataflow/scoped_summary_stats_test.py @@ -15,6 +15,8 @@ observable in ``merged_sd``. """ +import copy + import pandas as pd from typing import Any, TypedDict @@ -23,6 +25,7 @@ from buckaroo.customizations.pd_stats_v2 import PD_ANALYSIS_V2, PD_AUTOCLEAN_DEFAULT_V2, cleaning_gen_ops from buckaroo.dataflow.autocleaning import (AutocleaningConfig, PandasAutocleaning) from buckaroo.dataflow.dataflow import CustomizableDataflow, StylingAnalysis +from buckaroo.dataflow.sd_cache import split_chain_by_scope from buckaroo.pluggable_analysis_framework.stat_func import stat @@ -249,3 +252,67 @@ def test_analysis_klasses_change_invalidates_scoped_sd(): f"got the same key {key1} for both — likely the cache key still " f"omits analysis_klasses identity (Codex P2)" ) + + +def _scope_inputs(dfc): + """The pieces ``_merged_sd`` assembles, read back from the dataflow.""" + cache = dfc.summary_stats_cache + return dict( + init_sd=dfc.init_sd, cleaned_sd=dfc.cleaned_sd, raw_sd=cache[dfc.raw_sd_key], + processed_sd=dfc.processed_sd, processed_df=dfc.processed_df, + chains=split_chain_by_scope(dfc.operations), + clean_sd=cache[dfc.clean_sd_key], filt_sd=cache[dfc.filt_sd_key]) + + +def _assembly_dataflow(): + df = pd.DataFrame({'amount': ['10', '20', '30', '40', '50', '60', '70'], + 'word': ['foo', 'bar', 'foo', 'baz', 'foo', 'bar', 'foo']}) + # init_sd is keyed by the original column name; 'amount' is rewritten to 'a'. + dfc = ScopedDataflow(df, init_sd={ + 'amount': {'displayer_args': {'displayer': 'string', 'max_length': 99}, 'init_only': 1}}) + dfc.cleaning_method = 'default' + dfc.quick_command_args = {'search': ['foo']} + return dfc + + +def test_assemble_merged_sd_equals_merged_sd_with_init_sd_and_cleaning(): + """``assemble_merged_sd`` is the body of the ``merged_sd`` observer as a + pure function, so a server can assemble wire stats from an accumulating + raw sd without going through the traitlets cascade.""" + from buckaroo.dataflow.dataflow import assemble_merged_sd + dfc = _assembly_dataflow() + sd = dfc.merged_sd['a'] + assert sd['init_only'] == 1, "init_sd override must reach merged_sd under the rewritten name" + assert 'cleaned_mean' in sd and 'filtered_mean' in sd, "cleaning and filter layers must be active" + + assembled = assemble_merged_sd(**_scope_inputs(dfc)) + + assert assembled == dfc.merged_sd + + +def test_assemble_merged_sd_without_a_processed_df_merges_the_unrewritten_sds(): + from buckaroo.dataflow.dataflow import assemble_merged_sd + assembled = assemble_merged_sd( + init_sd={'x': {'init': 1}}, cleaned_sd={'x': {'cleaned': 2}}, raw_sd={'x': {'raw': 3}}, + processed_sd={'x': {'processed': 4}}, processed_df=None, + chains={'raw': [], 'clean': [], 'filt': []}) + assert assembled == {'x': {'init': 1, 'cleaned': 2, 'raw': 3, 'processed': 4}} + + +def test_assemble_merged_sd_does_not_mutate_its_inputs(): + from buckaroo.dataflow.dataflow import assemble_merged_sd + dfc = _assembly_dataflow() + inputs = _scope_inputs(dfc) + sd_names = ('init_sd', 'cleaned_sd', 'raw_sd', 'processed_sd', 'clean_sd', 'filt_sd') + + def key_sets(): + return {name: {col: sorted(stats) for col, stats in inputs[name].items()} for name in sd_names} + + before, chains_before = key_sets(), copy.deepcopy(inputs['chains']) + + assemble_merged_sd(**inputs) + + # Layering cleaned_* / filtered_* into a result column that shared its dict + # with an input would add keys to that input. + assert key_sets() == before + assert inputs['chains'] == chains_before diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 856752b35..c23ea3206 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -25,6 +25,9 @@ from xorq.common.utils.provenance_utils import read_parquet_provenance # noqa: E402 from xorq.expr.relations import CachedNode # noqa: E402 +from buckaroo.dataflow.sd_cache import split_chain_by_scope # noqa: E402 +from buckaroo.jlisp.lisp_utils import s as lisp_sym # noqa: E402 +from buckaroo.pluggable_analysis_framework.col_analysis import ColAnalysis # noqa: E402 from buckaroo.server import telemetry, xorq_loading # noqa: E402 from buckaroo.server.app import make_app as _make_app # noqa: E402 @@ -998,6 +1001,435 @@ def tracking_default_backend(): shutil.rmtree(builds_root, ignore_errors=True) +def _stats_tier_expr(): + """A 5-row table whose float column has a 1e9 maximum, so its estimated + column width depends on the min and max that only full stats supply.""" + return xo.memtable({ + "price": [12.5, 18.9, 7.4, 22.1, 1e9], + "qty": [1, 2, 1, 3, 2], + "category": ["a", "b", "a", "c", "b"]}) + + +def _build_dataflow(expr=None, **kwargs): + return xorq_loading.XorqServerDataflow( + _stats_tier_expr() if expr is None else expr, skip_main_serial=True, **kwargs) + + +def _three_scope_dataflow(expr=None, **kwargs): + """A dataflow with a user op (clean scope) and a search (filt scope) active, + so raw, clean and filt each hold their own summary-stats cache entry.""" + dataflow = _build_dataflow(expr, **kwargs) + dataflow.operations = [[lisp_sym("fillna"), {"symbol": "df"}, "qty", 0]] + dataflow.quick_command_args = {"search": ["a"]} + return dataflow + + +def _spy_data_queries(monkeypatch): + """Record the outermost op of every materialisation (``execute`` and + ``to_pyarrow``), and every stat query XorqStatPipeline sends.""" + from xorq.vendor.ibis.expr.types.core import Expr + + from buckaroo.pluggable_analysis_framework.xorq_stat_pipeline import XorqStatPipeline + ops, stat_queries = [], [] + original_execute, original_to_pyarrow = Expr.execute, Expr.to_pyarrow + original_stat_execute = XorqStatPipeline._execute + + def spy_execute(self, *args, **kwargs): + ops.append(type(self.op()).__name__) + return original_execute(self, *args, **kwargs) + + def spy_to_pyarrow(self, *args, **kwargs): + ops.append(type(self.op()).__name__) + return original_to_pyarrow(self, *args, **kwargs) + + def spy_stat_execute(self, query): + stat_queries.append(type(query.op()).__name__) + return original_stat_execute(self, query) + + monkeypatch.setattr(Expr, "execute", spy_execute) + monkeypatch.setattr(Expr, "to_pyarrow", spy_to_pyarrow) + monkeypatch.setattr(XorqStatPipeline, "_execute", spy_stat_execute) + return ops, stat_queries + + +def _without_min_width(column_config): + return [{**cc, "ag_grid_specs": {k: v for k, v in cc["ag_grid_specs"].items() if k != "minWidth"}} + for cc in column_config] + + +def _as_json(sd): + """Comparable form of an sd: json equates NaN with NaN where == would not.""" + return json.dumps(sd, sort_keys=True, default=str) + + +class _NoopPostProcessing(ColAnalysis): + provides_defaults = {} + post_processing_method = "noop_post" + + @classmethod + def post_process_df(cls, expr): + return [expr, {}] + + +class TestStatsTierSchema: + """``XorqServerDataflow(..., stats_tier="schema")`` (rows-first s1): the + dataflow publishes identity and typing for every column, with no data query + beyond the cached row count.""" + + def test_matches_full_stats_display_state(self): + expr = _stats_tier_expr() + full = _build_dataflow(expr) + schema = _build_dataflow(expr, stats_tier="schema") + assert schema.df_display_args.keys() == full.df_display_args.keys() + for name, full_arg in full.df_display_args.items(): + schema_arg = schema.df_display_args[name] + assert schema_arg["data_key"] == full_arg["data_key"] + assert schema_arg["summary_stats_key"] == full_arg["summary_stats_key"] + full_cfg, schema_cfg = full_arg["df_viewer_config"], schema_arg["df_viewer_config"] + assert schema_cfg["pinned_rows"] == full_cfg["pinned_rows"] + assert (_without_min_width(schema_cfg["column_config"]) + == _without_min_width(full_cfg["column_config"])) + + def test_min_width_is_the_stats_derived_difference(self): + expr = _stats_tier_expr() + widths = {} + for tier in ("full", "schema"): + cfg = _build_dataflow(expr, stats_tier=tier).df_display_args["main"]["df_viewer_config"] + widths[tier] = {cc["header_name"]: cc["ag_grid_specs"]["minWidth"] + for cc in cfg["column_config"]} + # price's 1e9 maximum widens it under full stats; without min and max + # the estimate falls back to a one-digit value. + assert widths["schema"]["price"] < widths["full"]["price"] + + def test_schema_sd_is_the_schema_keys_of_the_full_sd(self): + expr = _stats_tier_expr() + full_sd = _build_dataflow(expr).merged_sd + schema_sd = _build_dataflow(expr, stats_tier="schema").merged_sd + assert schema_sd.keys() == full_sd.keys() + for col, stats in schema_sd.items(): + assert {"orig_col_name", "rewritten_col_name", "dtype", "_type", "is_numeric", + "is_integer", "is_float", "is_bool", "is_datetime", "is_string", + "length"} <= stats.keys() + assert "mean" not in stats and "histogram" not in stats + assert stats == {k: full_sd[col][k] for k in stats} + + def test_issues_no_data_query_besides_the_cached_count(self, monkeypatch): + ops, stat_queries = _spy_data_queries(monkeypatch) + dataflow = _build_dataflow(stats_tier="schema") + dataflow.quick_command_args = {"search": ["a"]} + dataflow.add_analysis(_NoopPostProcessing) + assert stat_queries == [] + assert ops and set(ops) == {"CountStar"} + + def test_the_spy_sees_stat_queries_at_the_full_tier(self, monkeypatch): + ops, stat_queries = _spy_data_queries(monkeypatch) + _build_dataflow() + assert stat_queries, "the spy would not notice a stat query" + assert set(ops) - {"CountStar"} + + def test_init_sd_hints_and_overrides_still_apply(self): + dataflow = _build_dataflow( + stats_tier="schema", + init_sd={"qty": {"displayer_args": {"displayer": "string", "max_length": 200}}}, + column_config_overrides={ + "category": {"displayer_args": {"displayer": "string", "max_length": 5000}}}) + cfg = dataflow.df_display_args["main"]["df_viewer_config"]["column_config"] + by_header = {cc["header_name"]: cc for cc in cfg} + assert by_header["qty"]["displayer_args"]["max_length"] == 200 + assert by_header["category"]["displayer_args"]["max_length"] == 5000 + + def test_sorted_infinite_request_works(self): + dataflow = _build_dataflow(stats_tier="schema") + qty = next(k for k, v in dataflow.merged_sd.items() if v["orig_col_name"] == "qty") + resp, parquet = xorq_loading.handle_infinite_request_xorq( + dataflow, {"start": 0, "end": 5, "sourceName": "default", + "sort": qty, "sort_direction": "desc"}) + assert "error_info" not in resp + assert resp["length"] == 5 + assert pq.read_table(io.BytesIO(parquet)).column(qty).to_pylist() == [3, 2, 2, 1, 1] + + def test_pending_state_writes_no_full_tier_cache_key(self): + dataflow = _three_scope_dataflow(stats_tier="schema") + chains = split_chain_by_scope(dataflow.operations) + full_keys = {dataflow._scope_cache_key(chain, tier="full") for chain in chains.values()} + assert len(full_keys) == 3, "raw, clean and filt must each have their own key" + assert dataflow.summary_stats_cache + assert not full_keys & dataflow.summary_stats_cache.keys() + + def test_later_full_assignment_reaches_merged_sd_for_all_scopes(self): + expr = _stats_tier_expr() + full = _three_scope_dataflow(expr) + dataflow = _three_scope_dataflow(expr, stats_tier="schema") + assert "mean" not in dataflow.merged_sd["a"] + assert {"mean", "cleaned_mean", "filtered_mean"} <= full.merged_sd["a"].keys() + + dataflow.stats_tier = "full" + dataflow.summary_sd = full.summary_sd + + assert _as_json(dataflow.merged_sd) == _as_json(full.merged_sd) + + def test_a_summary_sd_from_another_tier_is_not_cached_under_the_new_tier(self): + """An sd computed at the schema tier must not become the full tier's + entry because the tier flipped before the next cascade: a present key + is a hit, so it would never be repaired.""" + dataflow = _three_scope_dataflow(stats_tier="schema") + stale = {col: {**stats, "stale": True} for col, stats in dataflow.summary_sd.items()} + + dataflow.stats_tier = "full" + dataflow.summary_sd = stale + + filt_sd = dataflow.summary_stats_cache[dataflow.filt_sd_key] + assert "mean" in filt_sd["a"] and "stale" not in filt_sd["a"] + assert "filtered_mean" in dataflow.merged_sd["a"] + + def test_full_scope_sds_cached_first_are_used_without_a_stat_query(self, monkeypatch): + expr = _stats_tier_expr() + full = _three_scope_dataflow(expr) + dataflow = _three_scope_dataflow(expr, stats_tier="schema") + cache = dict(dataflow.summary_stats_cache) + for scope, chain in split_chain_by_scope(dataflow.operations).items(): + cache[dataflow._scope_cache_key(chain, tier="full")] = full.summary_stats_cache[ + getattr(full, f"{scope}_sd_key")] + dataflow.summary_stats_cache = cache + _ops, stat_queries = _spy_data_queries(monkeypatch) + + dataflow.stats_tier = "full" + dataflow.summary_sd = full.summary_sd + + assert stat_queries == [] + assert _as_json(dataflow.merged_sd) == _as_json(full.merged_sd) + + +class TestLoadExprStatsPolicy(tornado.testing.AsyncHTTPTestCase): + """``stats_tier`` and ``stats_delivery`` on POST /load_expr and + /reload_expr (rows-first s1). The pair is stored beside dataflow_kwargs, + replayed by /reload_expr, and kept out of the has_config tuple.""" + + def get_app(self): + return make_app() + + def _session(self, sid): + return self._app.settings["sessions"].get(sid) + + @tornado.testing.gen_test + async def test_defaults_are_full_and_inline(self): + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + resp = await _post(self.get_http_port(), "/load_expr", + {"session": "sp-default", "build_dir": build_path}) + self.assertEqual(resp.code, 200) + session = self._session("sp-default") + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "inline")) + self.assertEqual(session.xorq_dataflow.stats_tier, "full") + self.assertIn("mean", session.xorq_dataflow.merged_sd["a"]) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_schema_tier_builds_a_schema_dataflow(self): + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + resp = await _post(self.get_http_port(), "/load_expr", + {"session": "sp-schema", "build_dir": build_path, "stats_tier": "schema"}) + self.assertEqual(resp.code, 200) + self.assertEqual(json.loads(resp.body)["rows"], 10) + session = self._session("sp-schema") + self.assertEqual((session.stats_tier, session.stats_delivery), ("schema", "inline")) + self.assertEqual(session.xorq_dataflow.stats_tier, "schema") + self.assertNotIn("mean", session.xorq_dataflow.merged_sd["a"]) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_deferred_delivery_publishes_a_schema_dataflow_and_serves_rows(self): + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + resp = await _post(self.get_http_port(), "/load_expr", + {"session": "sp-deferred", "build_dir": build_path, "stats_delivery": "deferred"}) + self.assertEqual(resp.code, 200) + session = self._session("sp-deferred") + # The session still targets full stats; the dataflow is built at + # the schema tier until a later phase delivers them. + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "deferred")) + self.assertEqual(session.xorq_dataflow.stats_tier, "schema") + + ws = await tornado.websocket.websocket_connect( + f"ws://localhost:{self.get_http_port()}/ws/sp-deferred") + initial = json.loads(await ws.read_message()) + self.assertEqual(initial["type"], "initial_state") + self.assertEqual(initial["df_meta"]["total_rows"], 10) + ws.write_message(json.dumps({"type": "infinite_request", + "payload_args": {"start": 0, "end": 10, "sourceName": "default", "origEnd": 10}})) + rows = json.loads(await ws.read_message()) + self.assertEqual(rows["length"], 10) + self.assertNotIn("error_info", rows) + table = pq.read_table(io.BytesIO(await ws.read_message())) + self.assertEqual(table.num_rows, 10) + ws.close() + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_deferred_delivery_runs_no_stat_query(self): + from buckaroo.pluggable_analysis_framework.xorq_stat_pipeline import XorqStatPipeline + builds_root = tempfile.mkdtemp() + stat_queries = [] + original = XorqStatPipeline._execute + + def spy(pipeline, query): + stat_queries.append(query) + return original(pipeline, query) + + try: + build_path = _build_expr_dir(builds_root) + with patch.object(XorqStatPipeline, "_execute", spy): + resp = await _post(self.get_http_port(), "/load_expr", + {"session": "sp-deferred-q", "build_dir": build_path, + "stats_delivery": "deferred"}) + self.assertEqual(resp.code, 200) + self.assertEqual(stat_queries, []) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_invalid_values_are_a_400(self): + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + for field, code in (("stats_tier", "invalid_stats_tier"), + ("stats_delivery", "invalid_stats_delivery")): + resp = await _post(self.get_http_port(), "/load_expr", + {"session": "sp-bad", "build_dir": build_path, field: "sometimes"}) + self.assertEqual(resp.code, 400) + self.assertEqual(json.loads(resp.body)["error_code"], code) + self.assertIsNone(self._session("sp-bad")) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_warm_repost_with_an_unchanged_pair_short_circuits(self): + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + body = {"session": "sp-warm", "build_dir": build_path, + "stats_tier": "schema", "stats_delivery": "deferred"} + self.assertEqual((await _post(self.get_http_port(), "/load_expr", body)).code, 200) + with patch.object(xorq_loading, "load_expr_build_dir", + side_effect=AssertionError("an unchanged pair must take the warm exit")): + same = await _post(self.get_http_port(), "/load_expr", body) + omitted = await _post(self.get_http_port(), "/load_expr", + {"session": "sp-warm", "build_dir": build_path}) + self.assertEqual(same.code, 200) + self.assertEqual(omitted.code, 200, "omitting the pair keeps the session's") + session = self._session("sp-warm") + self.assertEqual((session.stats_tier, session.stats_delivery), ("schema", "deferred")) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_the_default_pair_sent_on_every_post_does_not_defeat_the_warm_exit(self): + """has_config tests truthiness, so a host that always sends the pair + would rebuild on every POST if the fields were in it (#944).""" + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + body = {"session": "sp-warm-default", "build_dir": build_path, + "stats_tier": "full", "stats_delivery": "inline"} + self.assertEqual((await _post(self.get_http_port(), "/load_expr", body)).code, 200) + with patch.object(xorq_loading, "load_expr_build_dir", + side_effect=AssertionError("the warm exit must not see the pair as config")): + resp = await _post(self.get_http_port(), "/load_expr", body) + self.assertEqual(resp.code, 200) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_warm_repost_with_a_changed_pair_rebuilds(self): + builds_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + sid = "sp-changed" + original = xorq_loading.load_expr_build_dir + calls = [] + + def counting_loader(bd, **kwargs): + calls.append(bd) + return original(bd, **kwargs) + + with patch.object(xorq_loading, "load_expr_build_dir", side_effect=counting_loader): + await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path}) + self.assertEqual(len(calls), 1) + resp = await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, "stats_delivery": "deferred"}) + self.assertEqual(resp.code, 200) + self.assertEqual(len(calls), 2, "a changed delivery must rebuild") + self.assertEqual(self._session(sid).xorq_dataflow.stats_tier, "schema") + resp = await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, + "stats_tier": "schema", "stats_delivery": "deferred"}) + self.assertEqual(len(calls), 3, "a changed tier must rebuild") + resp = await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, + "stats_tier": "full", "stats_delivery": "inline"}) + self.assertEqual(len(calls), 4, "returning to the defaults must rebuild") + session = self._session(sid) + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "inline")) + self.assertEqual(session.xorq_dataflow.stats_tier, "full") + self.assertIn("mean", session.xorq_dataflow.merged_sd["a"]) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_reload_expr_replays_the_stored_pair(self): + builds_root = tempfile.mkdtemp() + project_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + sid = "sp-reload" + await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, "project_root": project_root, + "stats_delivery": "deferred"}) + before = self._session(sid).xorq_dataflow + resp = await _post(self.get_http_port(), f"/reload_expr/{sid}", {}) + self.assertEqual(resp.code, 200) + session = self._session(sid) + self.assertIsNot(session.xorq_dataflow, before) + self.assertEqual(session.xorq_dataflow.stats_tier, "schema") + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "deferred")) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + shutil.rmtree(project_root, ignore_errors=True) + + @tornado.testing.gen_test + async def test_reload_expr_accepts_a_new_pair_and_stores_it(self): + builds_root = tempfile.mkdtemp() + project_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + sid = "sp-reload-new" + await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, "project_root": project_root, + "stats_delivery": "deferred"}) + resp = await _post(self.get_http_port(), f"/reload_expr/{sid}", + {"stats_delivery": "inline"}) + self.assertEqual(resp.code, 200) + session = self._session(sid) + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "inline")) + self.assertEqual(session.xorq_dataflow.stats_tier, "full") + bad = await _post(self.get_http_port(), f"/reload_expr/{sid}", + {"stats_tier": "sometimes"}) + self.assertEqual(bad.code, 400) + self.assertEqual(json.loads(bad.body)["error_code"], "invalid_stats_tier") + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "inline")) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + shutil.rmtree(project_root, ignore_errors=True) + + class TestReloadExpr(tornado.testing.AsyncHTTPTestCase): def get_app(self): return make_app() From 5db0cfcb5b0499912950a574f35bad8e0fb68223 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 3 Oct 2026 23:14:52 -0400 Subject: [PATCH 03/14] feat(xorq): stats tier machinery and the xorq schema tier (rows-first s1) Add a dataflow-level stats_tier ("full" default, "schema"). The xorq schema tier builds identity and typing for every column from the expression's schema, with no data query beyond the cached row count, so a dataflow constructs in milliseconds rather than the stats' hundreds. The tier is part of _scope_cache_key, so a schema entry is never read as a full one, and _populate_sd_cache stores summary_sd under the filt key only if it was computed for the current frame, klass list and tier. add_analysis no longer builds DFStatsClass outside the hook when the tier is not full. The merged_sd observer body is extracted as the pure assemble_merged_sd, and _handle_widget_change is split into _build_df_data_dict and _build_df_display_args. /load_expr and /reload_expr accept stats_tier and stats_delivery, stored on the session beside dataflow_kwargs and replayed on reload. They stay out of the has_config tuple; the warm short-circuit compares the stored pair. stats_delivery="deferred" builds the schema-tier dataflow and publishes it. Defaults (full, inline) leave behaviour unchanged. Co-Authored-By: Claude Sonnet 5.5 --- buckaroo/customizations/xorq_stats_v2.py | 10 + buckaroo/dataflow/dataflow.py | 235 +++++++++++++++-------- buckaroo/server/handlers.py | 73 ++++++- buckaroo/server/session.py | 18 ++ buckaroo/xorq_buckaroo.py | 19 +- 5 files changed, 273 insertions(+), 82 deletions(-) diff --git a/buckaroo/customizations/xorq_stats_v2.py b/buckaroo/customizations/xorq_stats_v2.py index 239a261f4..74a2a2b19 100644 --- a/buckaroo/customizations/xorq_stats_v2.py +++ b/buckaroo/customizations/xorq_stats_v2.py @@ -139,6 +139,16 @@ def _type(is_bool: bool, is_integer: bool, is_float: bool, is_datetime: bool, is return "obj" +def schema_stats(dtype: str) -> dict: + """The stats a column's dtype alone determines: ``dtype``, the ``is_*`` + flags and ``_type``. This is what the xorq dataflow's schema tier + publishes, derived through the same stats the pipeline runs.""" + flags = typing_stats(dtype) + return {"dtype": dtype, **flags, "_type": _type( + flags["is_bool"], flags["is_integer"], flags["is_float"], flags["is_datetime"], + flags["is_string"])} + + # ============================================================ # Batched aggregates — one ibis.Expr each, folded into the batch query # ============================================================ diff --git a/buckaroo/dataflow/dataflow.py b/buckaroo/dataflow/dataflow.py index 8b6553621..161d4fccb 100644 --- a/buckaroo/dataflow/dataflow.py +++ b/buckaroo/dataflow/dataflow.py @@ -27,6 +27,63 @@ from .df_types import DataFrameT from .sd_cache import hash_chain, split_chain_by_scope +# Content levels a dataflow can be built at. ``full`` runs the summary-stats +# pipeline. ``schema`` publishes identity and typing keys derived from the +# schema alone and runs no data query (backends opt in, see ``_get_schema_sd``). +STATS_TIERS = ("full", "schema") + + +def assemble_merged_sd(init_sd: InitSD, cleaned_sd: SDType, raw_sd: SDType, processed_sd: SDType, + processed_df: TAny, chains: TDict[str, List[TAny]], + clean_sd: Optional[SDType] = None, filt_sd: Optional[SDType] = None) -> SDType: + """Build the wire ``merged_sd`` from the pieces ``_merged_sd`` reads. + + Bare keys come from the raw scope's sd (``raw_sd``, computed on + ``sampled_df``) merged under ``init_sd`` (keyed by original column name, + so rewritten here), ``cleaned_sd`` and ``processed_sd``. ``cleaned_*`` keys + are layered on top from ``clean_sd`` when cleaning is active, and + ``filtered_*`` keys from ``filt_sd`` when a search filter is active. + + Pure: the inputs are not mutated, so a caller holding an accumulating + ``raw_sd`` can assemble it repeatedly without the traitlets cascade. + ``chains`` is ``split_chain_by_scope(operations)``. + """ + if processed_df is None: + # on initial startup + return merge_sds(init_sd, cleaned_sd, raw_sd, processed_sd) + + # Gate each prefixed layer on a chain-shape diff between scopes. + # ``cleaned_*`` fires when the clean chain has ops the raw chain + # doesn't (cleaning is on); ``filtered_*`` fires when the filt + # chain has ops the clean chain doesn't (at least one + # quick-command op is present, i.e. a search filter is on). + # Keying off ``filt_sd_key != raw_sd_key`` would also fire + # ``filtered_*`` for cleaning-only states, mislabelling cleaned + # stats as filtered. + cleaning_active = chains['clean'] != chains['raw'] + filter_active = chains['filt'] != chains['clean'] + + #we do this to get rewrtten keys for init_sd + rewritten_init_sd = merge_sd_overrides({}, processed_df, init_sd) + intermediate_sd = merge_sds(rewritten_init_sd, cleaned_sd, raw_sd) + base = merge_sd_overrides(intermediate_sd, processed_df, processed_sd) + + # Layer ``cleaned_*`` keys on top when cleaning is active. + if cleaning_active and clean_sd: + for col, stats in clean_sd.items(): + col_dict = base.setdefault(col, {}) + for stat_name, val in stats.items(): + col_dict[f'cleaned_{stat_name}'] = val + + # Layer ``filtered_*`` keys on top when a filter is active. + if filter_active and filt_sd: + for col, stats in filt_sd.items(): + col_dict = base.setdefault(col, {}) + for stat_name, val in stats.items(): + col_dict[f'filtered_{stat_name}'] = val + + return base + class DfTrait(Any): """Any-trait for values that may contain a pandas/polars DataFrame. @@ -161,6 +218,10 @@ def __init__(self, raw_df): # are id()-based cache keys and the dead "foo"/"bar" test branches below, # all of which treat None and [] alike). analysis_klasses: List[Type[ColAnalysis]] = [] + # The content level the summary-stats cascade runs at (see STATS_TIERS). + # Instance-level on CustomizableDataflow; ``full`` here so the bare + # pipeline's dedupe key and cache keys can read it. + stats_tier: str = "full" summary_sd = Any() df_meta = Any() @@ -316,7 +377,7 @@ def _get_summary_sd(self, processed_df: DataFrameT, scope: str = 'filt') -> Tupl ret_summary[col] = {} return ret_summary, {} - _summary_sd_cache_key = (None, None) + _summary_sd_cache_key = (None, None, None) @observe('processed_result', 'analysis_klasses') @exception_protect('summary_sd-protector') @@ -326,7 +387,8 @@ def _summary_sd(self, change): # tuple wrapper, which makes this observer fire twice per widget # construction even when processed_df identity is unchanged. # Skip when neither the dataframe nor analysis_klasses has actually - # changed since the last run. See issue #709. + # changed since the last run. See issue #709. The tier is part of the + # key so a tier change reruns the summary even on the same frame. df = self.processed_df if df is None: # Nothing to summarize before the pipeline has produced a frame. @@ -334,9 +396,9 @@ def _summary_sd(self, change): # narrows Optional[DataFrameT] -> DataFrameT for _get_summary_sd. return klasses = self.analysis_klasses - if (id(df), id(klasses)) == self._summary_sd_cache_key: + if (id(df), id(klasses), self.stats_tier) == self._summary_sd_cache_key: return - self._summary_sd_cache_key = (id(df), id(klasses)) + self._summary_sd_cache_key = (id(df), id(klasses), self.stats_tier) result_summary_sd, errs = self._get_summary_sd(df, scope='filt') self.summary_sd = result_summary_sd self.errs = errs @@ -394,7 +456,11 @@ def __init__(self, orig_df, debug=False, pinned_rows:Union[Literal[None], PinnedRowConfig]=None, extra_grid_config=None, component_config:Union[Literal[None], ComponentConfig]=None, init_sd:Union[Literal[None], InitSD]=None, skip_main_serial=False, - skip_stat_columns=None): + skip_stat_columns=None, stats_tier:str="full"): + if stats_tier not in STATS_TIERS: + raise ValueError(f"stats_tier must be one of {STATS_TIERS}, got {stats_tier!r}") + # Set before super().__init__ — assigning raw_df runs the whole cascade. + self.stats_tier = stats_tier self.init_sd: InitSD if init_sd is None: self.init_sd = {} @@ -524,43 +590,9 @@ def _merged_sd(self, change): if filt_sd is None: filt_sd = self.summary_sd or {} - # Gate each prefixed layer on a chain-shape diff between scopes. - # ``cleaned_*`` fires when the clean chain has ops the raw chain - # doesn't (cleaning is on); ``filtered_*`` fires when the filt - # chain has ops the clean chain doesn't (at least one - # quick-command op is present, i.e. a search filter is on). - # Keying off ``filt_sd_key != raw_sd_key`` would also fire - # ``filtered_*`` for cleaning-only states, mislabelling cleaned - # stats as filtered. - chains = split_chain_by_scope(self.operations) - cleaning_active = chains['clean'] != chains['raw'] - filter_active = chains['filt'] != chains['clean'] - - if self.processed_df is None: - #on initial startup - self.merged_sd = merge_sds(self.init_sd, self.cleaned_sd, raw_sd, self.processed_sd) - return - - #we do this to get rewrtten keys for init_sd - rewritten_init_sd = merge_sd_overrides({}, self.processed_df, self.init_sd) - intermediate_sd = merge_sds(rewritten_init_sd, self.cleaned_sd, raw_sd) - base = merge_sd_overrides(intermediate_sd, self.processed_df, self.processed_sd) - - # Layer ``cleaned_*`` keys on top when cleaning is active. - if cleaning_active and clean_sd: - for col, stats in clean_sd.items(): - col_dict = base.setdefault(col, {}) - for stat_name, val in stats.items(): - col_dict[f'cleaned_{stat_name}'] = val - - # Layer ``filtered_*`` keys on top when a filter is active. - if filter_active and filt_sd: - for col, stats in filt_sd.items(): - col_dict = base.setdefault(col, {}) - for stat_name, val in stats.items(): - col_dict[f'filtered_{stat_name}'] = val - - self.merged_sd = base + self.merged_sd = assemble_merged_sd( + self.init_sd, self.cleaned_sd, raw_sd, self.processed_sd, self.processed_df, + split_chain_by_scope(self.operations), clean_sd=clean_sd, filt_sd=filt_sd) def _compute_scope_df(self, scope: str): """Return the df that scope's SD should be computed against. @@ -602,14 +634,14 @@ def _compute_scope_df(self, scope: str): return base return pp_result[0] if pp_result else base - def _scope_cache_key(self, chain): + def _scope_cache_key(self, chain, tier=None): """Hash that identifies a scope's SD-input identity. Includes the op chain *and* an identifier for the source dataframe (``id(sampled_df)``) *and* the post-processing method - *and* the analysis-klasses identity — all four are inputs to - the scope's SD, and a cache hit must mean "same SD-producing - inputs" not just "same chain". + *and* the analysis-klasses identity *and* the stats tier — all + five are inputs to the scope's SD, and a cache hit must mean + "same SD-producing inputs" not just "same chain". - sampled_df identity addresses codex P1 on #783: a ``raw_df`` swap with an unchanged chain must invalidate. @@ -621,11 +653,18 @@ def _scope_cache_key(self, chain): - analysis_klasses identity addresses codex P2 on #783: a klass-list swap with an unchanged chain must invalidate so new stat klasses surface in ``merged_sd``. + - the tier keeps a schema-tier sd from being read as a full one: + without it, assigning full stats after a schema-tier + construction hits the schema entries and never reaches + ``merged_sd``. ``tier`` defaults to the dataflow's own, and is + explicit so a caller can address the full-tier entry while the + dataflow is still at the schema tier. """ sampled_id = id(self.sampled_df) if self.sampled_df is not None else 0 pp = self.post_processing_method or '' klasses_id = id(self.analysis_klasses) - return hash_chain(chain, extra=f"{sampled_id}|{pp}|{klasses_id}") + tier = tier or self.stats_tier + return hash_chain(chain, extra=f"{sampled_id}|{pp}|{klasses_id}|{tier}") @observe('summary_sd', 'operations', 'analysis_klasses') @exception_protect('sd-cache-protector') @@ -657,9 +696,18 @@ def _populate_sd_cache(self, _change): new_cache = dict(self.summary_stats_cache) cache_grew = False - # filt scope reuses the SD that _summary_sd just produced. + # filt scope reuses the SD that _summary_sd just produced, but only if + # it was produced for this frame, klass list and tier. An sd left over + # from another (a tier changed since it ran, or an assignment ahead of + # the cascade) would be stored under this key and never repaired, since + # a present key is a hit. Compute the scope's SD instead. if keys['filt'] not in new_cache: - new_cache[keys['filt']] = dict(self.summary_sd or {}) + current = (id(self.processed_df), id(self.analysis_klasses), self.stats_tier) + if self._summary_sd_cache_key == current: + new_cache[keys['filt']] = dict(self.summary_sd or {}) + else: + filt_sd, _errs = self._get_summary_sd(self.processed_df, scope='filt') + new_cache[keys['filt']] = filt_sd cache_grew = True # raw + clean: fresh compute, but only on cache miss. @@ -715,6 +763,8 @@ def _build_error_dataframe(self, e) -> pd.DataFrame: #TAny closer to some error type @override def _get_summary_sd(self, processed_df: DataFrameT, scope: str = 'filt') -> Tuple[SDType, ErrDict]: + if self.stats_tier == "schema": + return self._get_schema_sd(processed_df), {} stats = self.DFStatsClass( processed_df, self.analysis_klasses, @@ -732,6 +782,16 @@ def _get_summary_sd(self, processed_df: DataFrameT, scope: str = 'filt') -> Tupl return sdf, {} + def _get_schema_sd(self, processed_df: DataFrameT) -> SDType: + """Identity and typing keys for every column, from the schema alone. + + ``orig_col_name``, ``rewritten_col_name``, ``dtype``, the ``is_*`` + flags, ``_type`` and ``length``: the keys column styling needs to pick + displayers, with no data query beyond the row count. A backend opts in + to ``stats_tier="schema"`` by overriding this. + """ + raise NotImplementedError(f"{type(self).__name__} has no schema stats tier") + # ### end summary stats block def _sd_to_jsondf(self, sd:SDType): @@ -749,35 +809,43 @@ def _sd_to_jsondf(self, sd:SDType): def _df_to_obj(self, df: DataFrameT) -> List[TDict[str, TAny]]: return pd_to_obj(self.sampling_klass.serialize_sample(df)) + def _analysis_klasses_with(self, analysis_klass:Type[ColAnalysis]) -> List[Type[ColAnalysis]]: + """The klass list the stats pipelines' ``add_stat`` builds: a class of the + same name is replaced, anything else is appended.""" + klasses = list(self.analysis_klasses) + if isinstance(analysis_klass, type): + klasses = [k for k in klasses + if not (isinstance(k, type) and k.__name__ == analysis_klass.__name__)] + klasses.append(analysis_klass) + return klasses + def add_analysis(self, analysis_klass:Type[ColAnalysis]) -> None: """ same as get_summary_sd, call whatever to set summary_sd and trigger further comps """ - stats = self.DFStatsClass( - self.processed_df, - self.analysis_klasses, - self.df_name, debug=self.debug) - stats.add_analysis(analysis_klass) - - self.analysis_klasses = stats.ap.ordered_a_objs + if self.stats_tier == "full": + stats = self.DFStatsClass( + self.processed_df, + self.analysis_klasses, + self.df_name, debug=self.debug) + stats.add_analysis(analysis_klass) + self.analysis_klasses = stats.ap.ordered_a_objs + else: + # Building DFStatsClass here would run the stats this tier exists + # to skip (twice, since the constructor and add_analysis each + # process the frame), and it does not go through _get_summary_sd. + self.analysis_klasses = self._analysis_klasses_with(analysis_klass) self.DFStatsClass.verify_analysis_objects(self.analysis_klasses) self.setup_options_from_analysis() #force recomputation self._handle_widget_change({}) - #final processing block - @observe('widget_args_tuple') - def _handle_widget_change(self, change): - """ - put together df_dict for consumption by the frontend - """ - # Tuple[TAny, pd.DataFrame, SDType] - _unused, processed_df, merged_sd = self.widget_args_tuple - if processed_df is None: - return - + def _build_df_data_dict(self, processed_df, merged_sd): + """The ``df_data_dict`` trait value: the main rows, the ``all_stats`` + payload and the empty placeholder. ``all_stats`` is the 5-11 ms part, + which is why it is built apart from ``_build_df_display_args``.""" # df_data_dict is still hardcoded for now # eventually processed_df will be able to add or alter values of df_data_dict # correlation would be added, filtered would probably be altered @@ -785,15 +853,15 @@ def _handle_widget_change(self, change): # to expedite processing maybe future provided dfs from # postprcoessing could default to empty until that is # selected, optionally - if self.skip_main_serial: - self.df_data_dict = {'main': [], - 'all_stats': self._sd_to_jsondf(merged_sd), - 'empty': []} - else: - self.df_data_dict = {'main': self._df_to_obj(processed_df), - 'all_stats': self._sd_to_jsondf(merged_sd), - 'empty': []} - + main = [] if self.skip_main_serial else self._df_to_obj(processed_df) + return {'main': main, + 'all_stats': self._sd_to_jsondf(merged_sd), + 'empty': []} + + def _build_df_display_args(self, processed_df, merged_sd): + """The ``df_display_args`` trait value: each display's column_config, + pinned_rows and keys, with the overrides applied. Reads ``merged_sd`` + but serializes none of it.""" temp_display_args = {} for display_name, A_Klass in self.df_display_klasses.items(): df_viewer_config = A_Klass.get_dfviewer_config(merged_sd, processed_df) @@ -811,8 +879,21 @@ def _handle_widget_change(self, change): temp_display_args['main']['df_viewer_config']['extra_grid_config'] = self.extra_grid_config if self.component_config: temp_display_args['main']['df_viewer_config']['component_config'] = self.component_config + return temp_display_args + + #final processing block + @observe('widget_args_tuple') + def _handle_widget_change(self, change): + """ + put together df_dict for consumption by the frontend + """ + # Tuple[TAny, pd.DataFrame, SDType] + _unused, processed_df, merged_sd = self.widget_args_tuple + if processed_df is None: + return - self.df_display_args = temp_display_args + self.df_data_dict = self._build_df_data_dict(processed_df, merged_sd) + self.df_display_args = self._build_df_display_args(processed_df, merged_sd) """ diff --git a/buckaroo/server/handlers.py b/buckaroo/server/handlers.py index 57e862e62..bb2cf848a 100644 --- a/buckaroo/server/handlers.py +++ b/buckaroo/server/handlers.py @@ -14,7 +14,9 @@ from buckaroo.compare import col_join_dfs from buckaroo.df_util import old_col_new_col from buckaroo.server.focus import find_or_create_session_window -from buckaroo.server.session import build_state_message +from buckaroo.dataflow.dataflow import STATS_TIERS +from buckaroo.server.session import ( + STATS_DELIVERIES, build_state_message, dataflow_stats_tier) from buckaroo.server import telemetry from buckaroo.pluggable_analysis_framework import perf_log @@ -396,6 +398,28 @@ async def post(self): self.write({"session": session_id, "server_pid": os.getpid(), "browser_action": browser_action, **metadata}) +def _stats_policy_from_body(body: dict, current_tier: str, current_delivery: str): + """Read ``stats_tier`` and ``stats_delivery`` from a request body. + + A field the body omits (or sends as null) keeps the current value (the + session's, or the default for a new session), as ``cache_dir`` does. Returns + ``(stats_tier, stats_delivery, None)``, or ``(None, None, error)`` with the + 400 response body for an unknown value.""" + stats_tier = body.get("stats_tier") + if stats_tier is None: + stats_tier = current_tier + if stats_tier not in STATS_TIERS: + return None, None, {"error_code": "invalid_stats_tier", + "message": f"stats_tier must be one of {list(STATS_TIERS)}, got {stats_tier!r}"} + stats_delivery = body.get("stats_delivery") + if stats_delivery is None: + stats_delivery = current_delivery + if stats_delivery not in STATS_DELIVERIES: + return None, None, {"error_code": "invalid_stats_delivery", + "message": f"stats_delivery must be one of {list(STATS_DELIVERIES)}, got {stats_delivery!r}"} + return stats_tier, stats_delivery, None + + class LoadExprHandler(tornado.web.RequestHandler): """POST /load_expr — load a xorq/ibis expression from a build dir and serve it via the xorq-backed buckaroo dataflow. @@ -454,7 +478,10 @@ async def post(self): # Config-bearing fields that change how the result is computed or # rendered. If the caller passes any of these on a warm POST we must # re-run the pipeline — returning cached metadata would silently - # ignore the new config. + # ignore the new config. stats_tier and stats_delivery are not in this + # tuple: it tests truthiness, and a host that sends the pair on every + # POST would never get the warm exit (#944). They are compared with the + # session's stored pair below instead. has_config = any(body.get(k) for k in ( "component_config", "column_config_overrides", "extra_grid_config", "init_sd", "skip_stat_columns")) @@ -479,6 +506,16 @@ async def post(self): # ~/.cache/xorq and recomputing the embedder's snapshots there. if cache_dir is None and existing is not None: cache_dir = existing.cache_dir + # The stats policy carries over the same way: a re-POST that omits it + # keeps the session's, and one that changes it rebuilds. + stats_tier, stats_delivery, policy_error = _stats_policy_from_body( + body, + existing.stats_tier if existing is not None else "full", + existing.stats_delivery if existing is not None else "inline") + if policy_error is not None: + self.set_status(400) + self.write(policy_error) + return # The identity of the rows the build reads (e.g. tallyman's snapshot # digest); it keys the summary-stat cache. Omitted, it carries over # like cache_dir, but only for the same build. A new value means the @@ -492,7 +529,8 @@ async def post(self): if (not force_reload and not has_config and existing and existing.backend == "xorq" and existing.build_dir == build_dir and existing.cache_dir == cache_dir and existing_kwargs.get("data_id") == data_id - and existing.metadata): + and existing.metadata + and (existing.stats_tier, existing.stats_delivery) == (stats_tier, stats_delivery)): # The pipeline is skipped, but the refreshed page still opens a new # WS and pulls a fresh time-to-first-rows (#944). telemetry.arm_session(existing, tele_sink) @@ -564,6 +602,7 @@ async def post(self): with perf_log.perf_span("firstpull.dataflow_construct", session=session_id): xorq_dataflow = xorq_loading.XorqServerDataflow( expr, skip_main_serial=True, extra_klasses=extra_klasses, + stats_tier=dataflow_stats_tier(stats_tier, stats_delivery), **dataflow_kwargs) # Spanning metadata too leaves only the small klass-load step # unmeasured inside the outer firstpull.load_expr total. @@ -589,6 +628,8 @@ async def post(self): session.cache_dir = cache_dir session.project_root = project_root session.dataflow_kwargs = dataflow_kwargs + session.stats_tier = stats_tier + session.stats_delivery = stats_delivery session.xorq_dataflow = xorq_dataflow # Clear pandas-side state left by a prior /load on the same # session so WS dispatch can no longer reach a stale dataflow. @@ -821,8 +862,22 @@ class ReloadExprHandler(tornado.web.RequestHandler): ``cache_storage_path`` store are cache hits, and column overrides, extra grid config, init_sd and skip_stat_columns survive the reload. + The session's stored ``stats_tier`` / ``stats_delivery`` are replayed too. + The body is optional; a pair in it replaces the stored one (and is stored) + when the reload succeeds. + Returns 404 when the session does not exist, 400 when it is not a xorq - session or has no project_root recorded, 501 when xorq is not installed.""" + session, has no project_root recorded or carries an invalid stats policy, + 501 when xorq is not installed.""" + + def _optional_body(self) -> dict: + """The JSON object body, or ``{}`` when there is none or it is not an + object: the endpoint took no body before the stats policy fields.""" + try: + body = json.loads(self.request.body) + except (json.JSONDecodeError, TypeError): + return {} + return body if isinstance(body, dict) else {} async def post(self, session_id): sessions = self.application.settings["sessions"] @@ -845,6 +900,13 @@ async def post(self, session_id): "message": "Session has no project_root — pass project_root to /load_expr first"}) return + stats_tier, stats_delivery, policy_error = _stats_policy_from_body( + self._optional_body(), session.stats_tier, session.stats_delivery) + if policy_error is not None: + self.set_status(400) + self.write(policy_error) + return + try: from buckaroo.server import xorq_loading except ImportError: @@ -861,6 +923,7 @@ async def post(self, session_id): + xorq_loading.load_project_display_klasses(session.project_root)) xorq_dataflow = xorq_loading.XorqServerDataflow( session.expr, skip_main_serial=True, extra_klasses=extra_klasses, + stats_tier=dataflow_stats_tier(stats_tier, stats_delivery), **session.dataflow_kwargs) except Exception: tb = traceback.format_exc() @@ -888,6 +951,8 @@ async def post(self, session_id): refreshed = get_buckaroo_display_state(xorq_dataflow) session.xorq_dataflow = xorq_dataflow + session.stats_tier = stats_tier + session.stats_delivery = stats_delivery session.df_display_args = refreshed["df_display_args"] session.df_data_dict = refreshed["df_data_dict"] session.df_meta = refreshed["df_meta"] diff --git a/buckaroo/server/session.py b/buckaroo/server/session.py index afb802d64..4602cff07 100644 --- a/buckaroo/server/session.py +++ b/buckaroo/server/session.py @@ -11,6 +11,17 @@ _DEFAULT_SESSION_TTL_S = 3600.0 # 1 hour idle before eviction _DEFAULT_EVICTION_INTERVAL_S = 300.0 # check every 5 minutes +# How a session reaches its stats tier: ``inline`` computes them in the +# dataflow constructor (today's behaviour); ``deferred`` builds the schema tier +# first and leaves the rest to later requests. +STATS_DELIVERIES = ("inline", "deferred") + + +def dataflow_stats_tier(stats_tier: str, stats_delivery: str) -> str: + """The tier a session's dataflow is constructed at. A deferred session + starts at the schema tier whatever tier it is headed for.""" + return "schema" if stats_delivery == "deferred" else stats_tier + @dataclass class SessionState: @@ -43,6 +54,13 @@ class SessionState: # extra_grid_config, init_sd, skip_stat_columns), replayed by /reload_expr so # a reload keeps stat caching and column config (#957). dataflow_kwargs: dict = field(default_factory=dict) + # The /load_expr stats policy. ``stats_tier`` is the tier the session is + # headed for (see dataflow.STATS_TIERS) and ``stats_delivery`` how it gets + # there (see STATS_DELIVERIES). Kept apart from dataflow_kwargs, which is + # splatted into the dataflow constructor, and replayed by /reload_expr. The + # tier the dataflow is built at follows from the pair (dataflow_stats_tier). + stats_tier: str = "full" + stats_delivery: str = "inline" # Companion telemetry sink (#943): a fire-and-forget POST callable, built # once from the /load_expr payload's telemetry_url on the IOLoop (where # make_http_sink captures AsyncHTTPClient/IOLoop.current()). Stored here so diff --git a/buckaroo/xorq_buckaroo.py b/buckaroo/xorq_buckaroo.py index 74cfb73da..f58ac1a44 100644 --- a/buckaroo/xorq_buckaroo.py +++ b/buckaroo/xorq_buckaroo.py @@ -27,7 +27,7 @@ from .buckaroo_widget import BuckarooInfiniteWidget, BuckarooWidget from .customizations.styling import DefaultMainStyling, DefaultSummaryStatsStyling from .customizations.xorq_autoclean_conf import NoCleaningConfXorq -from .customizations.xorq_stats_v2 import XORQ_STATS_V2 +from .customizations.xorq_stats_v2 import XORQ_STATS_V2, schema_stats from .dataflow.autocleaning import PandasAutocleaning from .dataflow.dataflow import CustomizableDataflow from .dataflow.dataflow_extras import Sampling @@ -220,6 +220,8 @@ def _get_summary_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str = 'orig_col_name': orig_col, 'rewritten_col_name': rewritten_col} return empty, {} + if self.stats_tier == "schema": + return self._get_schema_sd(processed_df), {} cache_storage = getattr(self, 'cache_storage', None) chain = self._scope_chain(scope) scope_id = self._stat_scope_id(chain) if cache_storage is not None else None @@ -266,6 +268,21 @@ def _get_summary_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str = rewritten[rewritten_col] = col_meta return rewritten, errs + def _get_schema_sd(self, processed_df: "XorqExpr | pd.DataFrame") -> dict: + """Identity, dtype, typing flags and row count for every column of an + expression, with no query beyond ``_expr_count`` (cached per + expression). Carries the same values full stats give those keys, so + styling that reads only them renders the same. ``_get_summary_sd`` + handles a pandas frame before it gets here.""" + expr = cast("XorqExpr", processed_df) + schema = expr.schema() + length = _expr_count(expr) + return { + rewritten_col: { + 'orig_col_name': orig_col, 'rewritten_col_name': rewritten_col, + **schema_stats(str(schema[str(orig_col)])), 'length': length} + for orig_col, rewritten_col in old_col_new_col(expr)} + _XORQ_ANALYSIS_KLASSES = list(XORQ_STATS_V2) + [DefaultSummaryStatsStyling, DefaultMainStyling] From f304948208d01646e3c73299cdfb261c5b0aa5e5 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 3 Oct 2026 23:24:11 -0400 Subject: [PATCH 04/14] test(dataflow): accept pandas 3's str dtype in the pandas characterization test (rows-first s1) The Max Versions jobs resolve pandas 3, which reports a string column's dtype as 'str' where pandas 2 says 'object'. The characterization test asserted 'object'. Verified in a Max Versions environment (pandas 3.0.6, polars 1.44.2, xorq 0.4.5): the unit suite passes. Co-Authored-By: Claude Sonnet 5.5 --- tests/unit/basic_widget_test.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/unit/basic_widget_test.py b/tests/unit/basic_widget_test.py index 9e63765e1..8852b8cdd 100644 --- a/tests/unit/basic_widget_test.py +++ b/tests/unit/basic_widget_test.py @@ -296,9 +296,11 @@ def test_default_tier_widget_change_sequence_and_merged_sd(): 'price': [12.5, 18.9, 7.4, 22.1, 14.0], 'qty': [1, 2, 1, 3, 2], 'category': ['a', 'b', 'a', 'c', 'b']})) msd = bw.dataflow.merged_sd - assert {col: (sd['length'], sd['_type'], sd['dtype']) for col, sd in msd.items()} == { - 'a': (5, 'float', 'float64'), 'b': (5, 'integer', 'int64'), - 'c': (5, 'string', 'object')} + assert {col: (sd['length'], sd['_type']) for col, sd in msd.items()} == { + 'a': (5, 'float'), 'b': (5, 'integer'), 'c': (5, 'string')} + # pandas 3 names the string dtype 'str' where pandas 2 says 'object'. + assert [msd[col]['dtype'] for col in 'ab'] == ['float64', 'int64'] + assert msd['c']['dtype'] in ('object', 'str') assert msd['a']['mean'] == pytest.approx(14.98) seen = [] From 0ef794e999c5e10866e56c2c69797d2c7b517670 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 3 Oct 2026 23:49:42 -0400 Subject: [PATCH 05/14] test(xorq): failing test for the schema tier overriding init_sd typing on a skipped column (rows-first s1) A column in skip_stat_columns gets only name, dtype and length from the full-tier pipeline, so its _type comes from init_sd. The schema tier layers the schema-derived _type and is_* keys over it, so an int64 column that init_sd types as float merges as integer and renders with zero fraction digits instead of the float displayer init_sd asked for. Co-Authored-By: Claude Sonnet 5.5 --- tests/unit/server/test_load_expr.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index c23ea3206..8d08af658 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -1138,6 +1138,29 @@ def test_init_sd_hints_and_overrides_still_apply(self): assert by_header["qty"]["displayer_args"]["max_length"] == 200 assert by_header["category"]["displayer_args"]["max_length"] == 5000 + def test_skipped_column_keeps_init_sd_typing(self): + """A column in ``skip_stat_columns`` gets only name, dtype and length + from the pipeline, so its ``_type`` comes from ``init_sd``. The schema + tier must not layer the schema-derived typing keys over it.""" + expr = _stats_tier_expr() + kwargs = { + "init_sd": {"qty": {"_type": "float", "mean": 1.8, "min": 1, "max": 3}}, + "skip_stat_columns": ["qty"]} + full = _build_dataflow(expr, **kwargs) + schema = _build_dataflow(expr, stats_tier="schema", **kwargs) + + def merged(dataflow, orig_col): + return next(v for v in dataflow.merged_sd.values() if v["orig_col_name"] == orig_col) + + assert merged(full, "qty")["_type"] == "float" + assert merged(schema, "qty")["_type"] == "float" + assert not {"is_numeric", "is_integer", "is_float"} & merged(schema, "qty").keys() + assert merged(schema, "price")["_type"] == "float" + assert merged(schema, "price")["is_float"] is True + full_cfg = full.df_display_args["main"]["df_viewer_config"]["column_config"] + schema_cfg = schema.df_display_args["main"]["df_viewer_config"]["column_config"] + assert _without_min_width(schema_cfg) == _without_min_width(full_cfg) + def test_sorted_infinite_request_works(self): dataflow = _build_dataflow(stats_tier="schema") qty = next(k for k, v in dataflow.merged_sd.items() if v["orig_col_name"] == "qty") From 91e1ff5d2a27f50af17295b7ccc8ec854c489bde Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 3 Oct 2026 23:56:33 -0400 Subject: [PATCH 06/14] fix(xorq): schema tier leaves a skipped column's typing to init_sd (rows-first s1) _get_schema_sd never read skip_stat_columns, so a skipped column's schema-derived _type and is_* keys overrode init_sd's _type once merged. The full tier gives a skipped column only name, dtype and length. The schema tier now does the same, so init_sd's _type decides the displayer at both tiers. Co-Authored-By: Claude Sonnet 5.5 --- buckaroo/xorq_buckaroo.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/buckaroo/xorq_buckaroo.py b/buckaroo/xorq_buckaroo.py index f58ac1a44..e2a03855b 100644 --- a/buckaroo/xorq_buckaroo.py +++ b/buckaroo/xorq_buckaroo.py @@ -272,16 +272,22 @@ def _get_schema_sd(self, processed_df: "XorqExpr | pd.DataFrame") -> dict: """Identity, dtype, typing flags and row count for every column of an expression, with no query beyond ``_expr_count`` (cached per expression). Carries the same values full stats give those keys, so - styling that reads only them renders the same. ``_get_summary_sd`` - handles a pandas frame before it gets here.""" + styling that reads only them renders the same. A column in + ``skip_stat_columns`` gets only name, dtype and length, as at the full + tier, so its typing comes from ``init_sd``. ``_get_summary_sd`` handles + a pandas frame before it gets here.""" expr = cast("XorqExpr", processed_df) schema = expr.schema() length = _expr_count(expr) - return { - rewritten_col: { + skip = getattr(self, 'skip_stat_columns', None) or () + sd: dict = {} + for orig_col, rewritten_col in old_col_new_col(expr): + dtype = str(schema[str(orig_col)]) + typing = {'dtype': dtype} if orig_col in skip else schema_stats(dtype) + sd[rewritten_col] = { 'orig_col_name': orig_col, 'rewritten_col_name': rewritten_col, - **schema_stats(str(schema[str(orig_col)])), 'length': length} - for orig_col, rewritten_col in old_col_new_col(expr)} + **typing, 'length': length} + return sd _XORQ_ANALYSIS_KLASSES = list(XORQ_STATS_V2) + [DefaultSummaryStatsStyling, DefaultMainStyling] From 7ea46262f5ba6f2586abd8e3c2f44ae3f7af6558 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 12:09:39 -0400 Subject: [PATCH 07/14] test(xorq): failing test for a schema-tier add_analysis that fails validation (rows-first s1) At the schema tier, add_analysis assigns the new klass list before verify_analysis_objects runs, so a klass that fails validation stays installed and every later add_analysis raises. Co-Authored-By: Claude Opus 5.5 --- tests/unit/server/test_load_expr.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 8d08af658..85aa849fd 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -1071,6 +1071,11 @@ def post_process_df(cls, expr): return [expr, {}] +class _V1Analysis(ColAnalysis): + """Fails DAG validation: v1 ColAnalysis classes are no longer supported.""" + provides_defaults = {"foo": 1} + + class TestStatsTierSchema: """``XorqServerDataflow(..., stats_tier="schema")`` (rows-first s1): the dataflow publishes identity and typing for every column, with no data query @@ -1121,6 +1126,15 @@ def test_issues_no_data_query_besides_the_cached_count(self, monkeypatch): assert stat_queries == [] assert ops and set(ops) == {"CountStar"} + def test_add_analysis_that_fails_validation_keeps_the_klasses(self): + dataflow = _build_dataflow(stats_tier="schema") + before = list(dataflow.analysis_klasses) + with pytest.raises(TypeError): + dataflow.add_analysis(_V1Analysis) + assert dataflow.analysis_klasses == before + dataflow.add_analysis(_NoopPostProcessing) + assert dataflow.analysis_klasses == [*before, _NoopPostProcessing] + def test_the_spy_sees_stat_queries_at_the_full_tier(self, monkeypatch): ops, stat_queries = _spy_data_queries(monkeypatch) _build_dataflow() From ba9e0819dbb194ea6740295c58fdd2773c296d0e Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 12:12:15 -0400 Subject: [PATCH 08/14] fix(xorq): schema-tier add_analysis validates the klass list before assigning it (rows-first s1) A klass that fails verify_analysis_objects no longer stays in analysis_klasses, so later add_analysis calls and a switch to the full tier keep working. Both tiers now validate the candidate list and only then assign it. Co-Authored-By: Claude Opus 5.5 --- buckaroo/dataflow/dataflow.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/buckaroo/dataflow/dataflow.py b/buckaroo/dataflow/dataflow.py index 161d4fccb..3e897e3bf 100644 --- a/buckaroo/dataflow/dataflow.py +++ b/buckaroo/dataflow/dataflow.py @@ -830,13 +830,15 @@ def add_analysis(self, analysis_klass:Type[ColAnalysis]) -> None: self.analysis_klasses, self.df_name, debug=self.debug) stats.add_analysis(analysis_klass) - self.analysis_klasses = stats.ap.ordered_a_objs + klasses = stats.ap.ordered_a_objs else: # Building DFStatsClass here would run the stats this tier exists # to skip (twice, since the constructor and add_analysis each # process the frame), and it does not go through _get_summary_sd. - self.analysis_klasses = self._analysis_klasses_with(analysis_klass) - self.DFStatsClass.verify_analysis_objects(self.analysis_klasses) + klasses = self._analysis_klasses_with(analysis_klass) + # Validate before assigning, so a klass that fails leaves the list as it was. + self.DFStatsClass.verify_analysis_objects(klasses) + self.analysis_klasses = klasses self.setup_options_from_analysis() #force recomputation self._handle_widget_change({}) From 227a6642b0f04adb8f0800519d7e7dafa8a71736 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 12:19:11 -0400 Subject: [PATCH 09/14] test(server): failing test for /reload_expr with a body that isn't UTF-8 (rows-first s1) _optional_body catches JSONDecodeError and TypeError, but json.loads on bytes that aren't valid UTF-8 raises UnicodeDecodeError, so the reload returns 500 where a body it can't parse should be ignored. Co-Authored-By: Claude Opus 5.5 --- tests/unit/server/test_load_expr.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 85aa849fd..98a1319e6 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -1579,6 +1579,30 @@ async def test_reload_expr_returns_200_with_zero_klasses(self): shutil.rmtree(builds_root, ignore_errors=True) shutil.rmtree(project_root, ignore_errors=True) + @tornado.testing.gen_test + async def test_reload_expr_ignores_a_body_it_cannot_parse(self): + """The body is optional, so one that isn't JSON (or isn't UTF-8) is + treated as absent rather than failing the reload.""" + builds_root = tempfile.mkdtemp() + project_root = tempfile.mkdtemp() + try: + build_path = _build_expr_dir(builds_root) + sid = "re-bad-body" + await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, + "project_root": project_root}) + client = tornado.httpclient.AsyncHTTPClient() + for raw in (b"not json", b"\xff"): + resp = await client.fetch( + f"http://localhost:{self.get_http_port()}/reload_expr/{sid}", + method="POST", body=raw, + headers={"Content-Type": "application/json"}, + raise_error=False) + self.assertEqual(resp.code, 200, raw) + finally: + shutil.rmtree(builds_root, ignore_errors=True) + shutil.rmtree(project_root, ignore_errors=True) + @tornado.testing.gen_test async def test_reload_expr_preserves_load_expr_config(self): """#957: /reload_expr must rebuild the dataflow with the same From a03bfa004c0f396d59084ac175fc783bd9953fef Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 12:20:01 -0400 Subject: [PATCH 10/14] fix(server): /reload_expr ignores a body that isn't UTF-8 (rows-first s1) _optional_body catches ValueError, which covers both JSONDecodeError and the UnicodeDecodeError json.loads raises for non-UTF-8 bytes. Co-Authored-By: Claude Opus 5.5 --- buckaroo/server/handlers.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/buckaroo/server/handlers.py b/buckaroo/server/handlers.py index bb2cf848a..ebafb74bd 100644 --- a/buckaroo/server/handlers.py +++ b/buckaroo/server/handlers.py @@ -875,7 +875,8 @@ def _optional_body(self) -> dict: object: the endpoint took no body before the stats policy fields.""" try: body = json.loads(self.request.body) - except (json.JSONDecodeError, TypeError): + # ValueError covers JSONDecodeError and UnicodeDecodeError (non-UTF-8 bytes). + except (ValueError, TypeError): return {} return body if isinstance(body, dict) else {} From 739f2c1d3e01b96d4d5dad9437928e245227422c Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 13:10:43 -0400 Subject: [PATCH 11/14] test(dataflow): failing tests for the stats_tier trait, set_stats_tier, add_analysis and /load (rows-first s1) - Assigning stats_tier changes nothing: it is a plain attribute, so a schema dataflow switched to full keeps its schema stats, and any string is accepted. - set_stats_tier, which installs a summary computed elsewhere (or one an earlier visit cached) without computing the filtered scope again, does not exist. The two tests that assigned summary_sd after flipping the tier now go through it or the trait. - add_analysis never reruns the summary stats, so an added stat never reaches merged_sd. - At the schema tier, an op under a search counts the rebuilt clean-scope frame as well as the filtered one. - /load leaves the stats policy on the session, so a later /load_expr that omits it inherits an earlier load's deferred delivery. Co-Authored-By: Claude Opus 5.5 --- .../dataflow/customizable_dataflow_test.py | 13 ++++ tests/unit/server/test_load_expr.py | 65 +++++++++++++++++-- 2 files changed, 74 insertions(+), 4 deletions(-) diff --git a/tests/unit/dataflow/customizable_dataflow_test.py b/tests/unit/dataflow/customizable_dataflow_test.py index 909302c7a..688155d81 100644 --- a/tests/unit/dataflow/customizable_dataflow_test.py +++ b/tests/unit/dataflow/customizable_dataflow_test.py @@ -21,6 +21,11 @@ def always_fail(ser: RawSeries) -> int: raise ZeroDivisionError("always fails") +@stat() +def double_length(length: int) -> int: + return length * 2 + + BASIC_DF = get_basic_df2() EMPTY_DF_JSON = { 'dfviewer_config': { @@ -295,6 +300,14 @@ class PostDCFC(ACDFC): assert p_dfc.cleaned_sd == {} assert p_dfc.df_display_args['main']['df_viewer_config'] == DFVIEWER_CONFIG_DEFAULT +def test_add_analysis_recomputes_the_summary_stats(): + """The summary stats rerun with the new klass list, so the added stat + reaches merged_sd.""" + bw = BuckarooWidget(BASIC_DF) + bw.add_analysis(double_length) + for col in ('a', 'b'): + assert bw.dataflow.merged_sd[col]['double_length'] == 2 * bw.dataflow.merged_sd[col]['length'] + class HidePostProcessingAnalysis2(ColAnalysis): provides_defaults = {} diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 98a1319e6..9e59caf4a 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -15,6 +15,7 @@ import tornado.httpclient import tornado.testing import tornado.websocket +from traitlets import TraitError xo = pytest.importorskip("xorq.api") @@ -1126,6 +1127,16 @@ def test_issues_no_data_query_besides_the_cached_count(self, monkeypatch): assert stat_queries == [] assert ops and set(ops) == {"CountStar"} + def test_an_op_under_a_search_counts_only_the_filtered_frame(self, monkeypatch): + """The clean scope's frame is rebuilt on every cache miss, so counting + it is a query of its own. The filtered frame's count is one df_meta + runs anyway.""" + dataflow = _build_dataflow(stats_tier="schema") + dataflow.quick_command_args = {"search": ["a"]} + ops, _stat_queries = _spy_data_queries(monkeypatch) + dataflow.operations = [[lisp_sym("fillna"), {"symbol": "df"}, "qty", 0]] + assert ops == ["CountStar"] + def test_add_analysis_that_fails_validation_keeps_the_klasses(self): dataflow = _build_dataflow(stats_tier="schema") before = list(dataflow.analysis_klasses) @@ -1193,7 +1204,7 @@ def test_pending_state_writes_no_full_tier_cache_key(self): assert dataflow.summary_stats_cache assert not full_keys & dataflow.summary_stats_cache.keys() - def test_later_full_assignment_reaches_merged_sd_for_all_scopes(self): + def test_switching_the_tier_computes_its_stats_for_all_scopes(self): expr = _stats_tier_expr() full = _three_scope_dataflow(expr) dataflow = _three_scope_dataflow(expr, stats_tier="schema") @@ -1201,10 +1212,32 @@ def test_later_full_assignment_reaches_merged_sd_for_all_scopes(self): assert {"mean", "cleaned_mean", "filtered_mean"} <= full.merged_sd["a"].keys() dataflow.stats_tier = "full" - dataflow.summary_sd = full.summary_sd assert _as_json(dataflow.merged_sd) == _as_json(full.merged_sd) + def test_an_unknown_tier_is_rejected(self): + dataflow = _build_dataflow(stats_tier="schema") + with pytest.raises(TraitError): + dataflow.stats_tier = "Full" + assert dataflow.stats_tier == "schema" + + def test_set_stats_tier_installs_the_summary_it_is_given(self, monkeypatch): + """Stats computed off the IOLoop go in as they are: the filtered scope + is not computed again, only the scopes the cascade still lacks.""" + expr = _stats_tier_expr() + full = _three_scope_dataflow(expr) + dataflow = _three_scope_dataflow(expr, stats_tier="schema") + scopes = [] + original = dataflow._get_summary_sd + monkeypatch.setattr(dataflow, "_get_summary_sd", + lambda processed_df, scope="filt": scopes.append(scope) or original(processed_df, scope=scope)) + + dataflow.set_stats_tier("full", (full.summary_sd, {})) + + assert scopes == ["raw", "clean"] + assert dataflow.stats_tier == "full" + assert _as_json(dataflow.merged_sd) == _as_json(full.merged_sd) + def test_a_summary_sd_from_another_tier_is_not_cached_under_the_new_tier(self): """An sd computed at the schema tier must not become the full tier's entry because the tier flipped before the next cascade: a present key @@ -1230,8 +1263,7 @@ def test_full_scope_sds_cached_first_are_used_without_a_stat_query(self, monkeyp dataflow.summary_stats_cache = cache _ops, stat_queries = _spy_data_queries(monkeypatch) - dataflow.stats_tier = "full" - dataflow.summary_sd = full.summary_sd + dataflow.set_stats_tier("full") assert stat_queries == [] assert _as_json(dataflow.merged_sd) == _as_json(full.merged_sd) @@ -1420,6 +1452,31 @@ def counting_loader(bd, **kwargs): finally: shutil.rmtree(builds_root, ignore_errors=True) + @tornado.testing.gen_test + async def test_a_load_in_between_resets_the_pair(self): + """/load replaces the session's xorq state, so a later /load_expr that + omits the pair starts from the defaults, not from an earlier load's.""" + builds_root = tempfile.mkdtemp() + csv_fd, csv_path = tempfile.mkstemp(suffix=".csv") + os.close(csv_fd) + try: + build_path = _build_expr_dir(builds_root) + pd.DataFrame({"x": [1, 2]}).to_csv(csv_path, index=False) + sid = "sp-load-between" + await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path, "stats_delivery": "deferred"}) + await _post(self.get_http_port(), "/load", + {"session": sid, "path": csv_path, "mode": "buckaroo"}) + resp = await _post(self.get_http_port(), "/load_expr", + {"session": sid, "build_dir": build_path}) + self.assertEqual(resp.code, 200) + session = self._session(sid) + self.assertEqual((session.stats_tier, session.stats_delivery), ("full", "inline")) + self.assertEqual(session.xorq_dataflow.stats_tier, "full") + finally: + shutil.rmtree(builds_root, ignore_errors=True) + os.unlink(csv_path) + @tornado.testing.gen_test async def test_reload_expr_replays_the_stored_pair(self): builds_root = tempfile.mkdtemp() From 55d70bb91251f4b57291cbe1221425f668d7847b Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 13:18:41 -0400 Subject: [PATCH 12/14] fix(dataflow): stats_tier is a trait, set_stats_tier installs stats in hand, add_analysis reruns the summary (rows-first s1) stats_tier is now a traitlets Enum that _summary_sd observes. Assigning it validates the value and computes the new tier's stats, so switching tiers no longer means assigning summary_sd by hand. set_stats_tier switches without that compute when the stats are already in hand: a summary computed elsewhere (stored under the state's key at the new tier), or the entry an earlier visit cached. It records the dedupe key first, so the observer doesn't redo them. _summary_sd records its dedupe key only after the stats are computed. The _populate_sd_cache guard no longer computes the filt scope itself: a summary_sd that isn't this frame's is left out of the cache, and the next run fills it. That removes the synchronous recompute on a tier switch and the errs it dropped. add_analysis builds the klass list with with_stat, the rule both pipelines' add_stat now share, validates it, then reruns _summary_sd. The full-tier branch used to run two stat pipelines and throw both away, so an added stat never reached merged_sd. The tier is checked once, in CustomizableDataflow._get_summary_sd, which dispatches to _get_full_sd or _get_schema_sd. XorqDataflow overrides _get_full_sd and keeps only its pandas-frame branch in _get_summary_sd. Its _get_schema_sd takes the scope and gives the clean scope no length: that frame is rebuilt on every cache miss, so its count was a query of its own. Co-Authored-By: Claude Opus 5.5 --- buckaroo/dataflow/dataflow.py | 122 ++++++++++-------- .../stat_pipeline.py | 20 +-- .../xorq_stat_pipeline.py | 15 +-- buckaroo/xorq_buckaroo.py | 29 +++-- 4 files changed, 100 insertions(+), 86 deletions(-) diff --git a/buckaroo/dataflow/dataflow.py b/buckaroo/dataflow/dataflow.py index 3e897e3bf..2ba5d133a 100644 --- a/buckaroo/dataflow/dataflow.py +++ b/buckaroo/dataflow/dataflow.py @@ -5,12 +5,13 @@ import six import warnings import pandas as pd -from traitlets import Unicode, Any, observe, Dict +from traitlets import Unicode, Any, observe, Dict, Enum from buckaroo.pluggable_analysis_framework.col_analysis import ColAnalysis, ErrDict, SDType from ..serialization_utils import pd_to_obj, sd_to_parquet_b64, project_sd from buckaroo.pluggable_analysis_framework.utils import (filter_analysis) from buckaroo.pluggable_analysis_framework.df_stats_v2 import DfStatsV2 +from buckaroo.pluggable_analysis_framework.stat_pipeline import with_stat from .autocleaning import SentinelAutocleaning from .dataflow_extras import (exception_protect, Sampling) from .styling_core import ( @@ -218,10 +219,11 @@ def __init__(self, raw_df): # are id()-based cache keys and the dead "foo"/"bar" test branches below, # all of which treat None and [] alike). analysis_klasses: List[Type[ColAnalysis]] = [] - # The content level the summary-stats cascade runs at (see STATS_TIERS). - # Instance-level on CustomizableDataflow; ``full`` here so the bare - # pipeline's dedupe key and cache keys can read it. - stats_tier: str = "full" + # The content level the summary-stats cascade runs at (see STATS_TIERS). A + # trait, so assigning it reruns _summary_sd at the new tier and an unknown + # value is refused. ``set_stats_tier`` switches without that compute when + # the new tier's stats are already in hand. + stats_tier = Enum(STATS_TIERS, default_value="full") summary_sd = Any() df_meta = Any() @@ -379,7 +381,7 @@ def _get_summary_sd(self, processed_df: DataFrameT, scope: str = 'filt') -> Tupl _summary_sd_cache_key = (None, None, None) - @observe('processed_result', 'analysis_klasses') + @observe('processed_result', 'analysis_klasses', 'stats_tier') @exception_protect('summary_sd-protector') def _summary_sd(self, change): # Dedupe: the autocleaning operations cascade re-fires @@ -395,11 +397,13 @@ def _summary_sd(self, change): # Matches the guards in _merged_sd and _populate_sd_cache, and # narrows Optional[DataFrameT] -> DataFrameT for _get_summary_sd. return - klasses = self.analysis_klasses - if (id(df), id(klasses), self.stats_tier) == self._summary_sd_cache_key: + key = (id(df), id(self.analysis_klasses), self.stats_tier) + if key == self._summary_sd_cache_key: return - self._summary_sd_cache_key = (id(df), id(klasses), self.stats_tier) result_summary_sd, errs = self._get_summary_sd(df, scope='filt') + # Recorded once the stats exist: after a run that raises, the key still + # names the frame summary_sd belongs to, which _populate_sd_cache checks. + self._summary_sd_cache_key = key self.summary_sd = result_summary_sd self.errs = errs @@ -697,17 +701,13 @@ def _populate_sd_cache(self, _change): cache_grew = False # filt scope reuses the SD that _summary_sd just produced, but only if - # it was produced for this frame, klass list and tier. An sd left over - # from another (a tier changed since it ran, or an assignment ahead of - # the cascade) would be stored under this key and never repaired, since - # a present key is a hit. Compute the scope's SD instead. - if keys['filt'] not in new_cache: - current = (id(self.processed_df), id(self.analysis_klasses), self.stats_tier) - if self._summary_sd_cache_key == current: - new_cache[keys['filt']] = dict(self.summary_sd or {}) - else: - filt_sd, _errs = self._get_summary_sd(self.processed_df, scope='filt') - new_cache[keys['filt']] = filt_sd + # it was produced for this frame, klass list and tier. One left over + # from an earlier frame (its run raised) stays out of the cache: stored + # under this key it would never be repaired, since a present key is a + # hit. The next _summary_sd run for this state fills it. + current = (id(self.processed_df), id(self.analysis_klasses), self.stats_tier) + if keys['filt'] not in new_cache and self._summary_sd_cache_key == current: + new_cache[keys['filt']] = dict(self.summary_sd or {}) cache_grew = True # raw + clean: fresh compute, but only on cache miss. @@ -763,8 +763,15 @@ def _build_error_dataframe(self, e) -> pd.DataFrame: #TAny closer to some error type @override def _get_summary_sd(self, processed_df: DataFrameT, scope: str = 'filt') -> Tuple[SDType, ErrDict]: + """``processed_df``'s summary stats at the dataflow's tier: the stats + pipeline's (``_get_full_sd``) or the schema's (``_get_schema_sd``).""" if self.stats_tier == "schema": - return self._get_schema_sd(processed_df), {} + return self._get_schema_sd(processed_df, scope), {} + return self._get_full_sd(processed_df, scope) + + def _get_full_sd(self, processed_df: DataFrameT, scope: str) -> Tuple[SDType, ErrDict]: + """The stats pipeline's summary of ``processed_df``, the frame of + ``scope``. A backend overrides this to run its own pipeline.""" stats = self.DFStatsClass( processed_df, self.analysis_klasses, @@ -782,17 +789,47 @@ def _get_summary_sd(self, processed_df: DataFrameT, scope: str = 'filt') -> Tupl return sdf, {} - def _get_schema_sd(self, processed_df: DataFrameT) -> SDType: + def _get_schema_sd(self, processed_df: DataFrameT, scope: str) -> SDType: """Identity and typing keys for every column, from the schema alone. ``orig_col_name``, ``rewritten_col_name``, ``dtype``, the ``is_*`` flags, ``_type`` and ``length``: the keys column styling needs to pick - displayers, with no data query beyond the row count. A backend opts in - to ``stats_tier="schema"`` by overriding this. + displayers, with no data query beyond the row count. ``scope`` is the + frame's scope, for a backend whose row count is a query. A backend opts + in to ``stats_tier="schema"`` by overriding this. """ raise NotImplementedError(f"{type(self).__name__} has no schema stats tier") - # ### end summary stats block + def set_stats_tier(self, tier: str, summary: Optional[Tuple[SDType, ErrDict]] = None) -> None: + """Switch to ``tier`` without computing stats that are already in hand. + + ``summary`` is the current frame's ``(sd, errs)`` at ``tier``, when the + caller computed it elsewhere (off the IOLoop, say). Without it, the + entry an earlier visit to this state left in ``summary_stats_cache`` is + used, and failing that this is plain assignment: the ``stats_tier`` + observer computes the tier's stats. The cascade then fills the raw and + clean scopes from the cache, computing the ones it lacks. + """ + filt_key = self._scope_cache_key(split_chain_by_scope(self.operations)['filt'], tier=tier) + if summary is None: + cached = self.summary_stats_cache.get(filt_key) + if cached is None: + self.stats_tier = tier + return + summary = (cached, {}) + else: + # Stored under the state's key at the new tier, so merged_sd reads + # the stats given rather than an entry already there. + self.summary_stats_cache = {**self.summary_stats_cache, filt_key: summary[0]} + sd, errs = summary + # Recorded first, so the stats_tier observer finds this frame's stats + # present and leaves them to the assignments below. + self._summary_sd_cache_key = (id(self.processed_df), id(self.analysis_klasses), tier) + self.stats_tier = tier + self.summary_sd = sd + self.errs = errs + + # ### end summary stats block def _sd_to_jsondf(self, sd:SDType): """Serialize summary stats to the wire payload (parquet-b64 tagged dict). @@ -809,38 +846,19 @@ def _sd_to_jsondf(self, sd:SDType): def _df_to_obj(self, df: DataFrameT) -> List[TDict[str, TAny]]: return pd_to_obj(self.sampling_klass.serialize_sample(df)) - def _analysis_klasses_with(self, analysis_klass:Type[ColAnalysis]) -> List[Type[ColAnalysis]]: - """The klass list the stats pipelines' ``add_stat`` builds: a class of the - same name is replaced, anything else is appended.""" - klasses = list(self.analysis_klasses) - if isinstance(analysis_klass, type): - klasses = [k for k in klasses - if not (isinstance(k, type) and k.__name__ == analysis_klass.__name__)] - klasses.append(analysis_klass) - return klasses - def add_analysis(self, analysis_klass:Type[ColAnalysis]) -> None: - """ - same as get_summary_sd, call whatever to set summary_sd and trigger further comps - """ - - if self.stats_tier == "full": - stats = self.DFStatsClass( - self.processed_df, - self.analysis_klasses, - self.df_name, debug=self.debug) - stats.add_analysis(analysis_klass) - klasses = stats.ap.ordered_a_objs - else: - # Building DFStatsClass here would run the stats this tier exists - # to skip (twice, since the constructor and add_analysis each - # process the frame), and it does not go through _get_summary_sd. - klasses = self._analysis_klasses_with(analysis_klass) + """Add ``analysis_klass``, replacing a class of the same name, and rerun + the summary stats with it at the dataflow's tier.""" + klasses = with_stat(self.analysis_klasses, analysis_klass) # Validate before assigning, so a klass that fails leaves the list as it was. self.DFStatsClass.verify_analysis_objects(klasses) self.analysis_klasses = klasses self.setup_options_from_analysis() - #force recomputation + # analysis_klasses is a plain attribute, so the cascade starts here; the + # new list misses _summary_sd's dedupe key. + self._summary_sd({}) + # A klass that changes only the displays (post-processing, styling) + # leaves summary_sd as it was, and nothing downstream would rebuild them. self._handle_widget_change({}) diff --git a/buckaroo/pluggable_analysis_framework/stat_pipeline.py b/buckaroo/pluggable_analysis_framework/stat_pipeline.py index 9823b8b0f..3404c748f 100644 --- a/buckaroo/pluggable_analysis_framework/stat_pipeline.py +++ b/buckaroo/pluggable_analysis_framework/stat_pipeline.py @@ -23,6 +23,16 @@ from .utils import PERVERSE_DF +def with_stat(inputs: list, stat_func_or_class) -> list: + """``inputs`` with ``stat_func_or_class`` added: a class of the same name is + replaced, anything else is appended. The rule both pipelines' ``add_stat`` + and ``CustomizableDataflow.add_analysis`` follow.""" + if isinstance(stat_func_or_class, type): + inputs = [inp for inp in inputs + if not (isinstance(inp, type) and inp.__name__ == stat_func_or_class.__name__)] + return [*inputs, stat_func_or_class] + + def _normalize_inputs(inputs: list) -> List[StatFunc]: """Convert a mixed list of StatFunc, @stat functions, and stat-group classes to StatFuncs. @@ -349,15 +359,7 @@ def add_stat(self, stat_func_or_class) -> Tuple[bool, List[StatError]]: Validates the DAG and runs unit test against PERVERSE_DF. """ - new_inputs = list(self._original_inputs) - - # Remove existing with same name if re-adding - if isinstance(stat_func_or_class, type): - new_inputs = [ - inp for inp in new_inputs - if not (isinstance(inp, type) and inp.__name__ == stat_func_or_class.__name__) - ] - new_inputs.append(stat_func_or_class) + new_inputs = with_stat(self._original_inputs, stat_func_or_class) try: new_funcs = _normalize_inputs(new_inputs) diff --git a/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py b/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py index 793779092..8abe74893 100644 --- a/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py +++ b/buckaroo/pluggable_analysis_framework/xorq_stat_pipeline.py @@ -38,7 +38,7 @@ from .stat_cache import (CachedScope, CachedStatError, StatCache, length_stat_id, make_scope_id, stat_hashes) from .stat_func import XorqColumn, XorqExpr, XorqExecute, RAW_MARKER_TYPES, StatFunc -from .stat_pipeline import _execute_stat_func, _normalize_inputs, errors_to_errdict +from .stat_pipeline import _execute_stat_func, _normalize_inputs, errors_to_errdict, with_stat from .stat_result import Err, Ok, StatError, StatResult, resolve_accumulator from .typed_dag import build_column_dag, build_typed_dag from .utils import PERVERSE_DF @@ -602,18 +602,7 @@ def add_stat(self, stat_func_or_class) -> Tuple[bool, List[StatError]]: against). Validates the DAG; returns ``(True, [])`` on success or ``(False, [config_error])`` if the DAG can't be built. """ - new_inputs = list(self._original_inputs) - - if isinstance(stat_func_or_class, type): - new_inputs = [ - inp - for inp in new_inputs - if not ( - isinstance(inp, type) - and inp.__name__ == stat_func_or_class.__name__ - ) - ] - new_inputs.append(stat_func_or_class) + new_inputs = with_stat(self._original_inputs, stat_func_or_class) try: new_funcs = _normalize_inputs(new_inputs) diff --git a/buckaroo/xorq_buckaroo.py b/buckaroo/xorq_buckaroo.py index e2a03855b..40bfbb2ca 100644 --- a/buckaroo/xorq_buckaroo.py +++ b/buckaroo/xorq_buckaroo.py @@ -144,7 +144,7 @@ class XorqDataflow(CustomizableDataflow["XorqExpr | pd.DataFrame"]): 1. ``populate_df_meta`` can't ``len(expr)`` — it issues an ``expr.count().execute()`` for the row count. - 2. ``_get_summary_sd`` re-keys the summary dict from original column + 2. ``_get_full_sd`` re-keys the summary dict from original column names (what ``XorqStatPipeline`` produces) to the rewritten ``a, b, c`` names that ``pd_to_obj`` and the styling layer expect. @@ -220,8 +220,9 @@ def _get_summary_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str = 'orig_col_name': orig_col, 'rewritten_col_name': rewritten_col} return empty, {} - if self.stats_tier == "schema": - return self._get_schema_sd(processed_df), {} + return super()._get_summary_sd(processed_df, scope) + + def _get_full_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str): cache_storage = getattr(self, 'cache_storage', None) chain = self._scope_chain(scope) scope_id = self._stat_scope_id(chain) if cache_storage is not None else None @@ -241,10 +242,10 @@ def _get_summary_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str = # signal only the server observes (#943). Carries the write side too # (snapshots/bytes/write_errors, #951) so a cache that stops writing — # or a run with write_errors > 0 — is visible to a telemetry consumer - # rather than only to a server log no deployment reads. The pandas - # branch above already returned, so stats here is always an - # XorqDfStatsV2, which always exposes cache_run_stats(); call it - # unconditionally so a missing signal fails loudly rather than + # rather than only to a server log no deployment reads. + # _get_summary_sd keeps pandas frames from reaching here, so stats is + # always an XorqDfStatsV2, which always exposes cache_run_stats(); call + # it unconditionally so a missing signal fails loudly rather than # silently vanishing. cs = stats.cache_run_stats() span.set_attr( @@ -268,17 +269,21 @@ def _get_summary_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str = rewritten[rewritten_col] = col_meta return rewritten, errs - def _get_schema_sd(self, processed_df: "XorqExpr | pd.DataFrame") -> dict: + def _get_schema_sd(self, processed_df: "XorqExpr | pd.DataFrame", scope: str) -> dict: """Identity, dtype, typing flags and row count for every column of an - expression, with no query beyond ``_expr_count`` (cached per - expression). Carries the same values full stats give those keys, so + expression. The count (``_expr_count``, cached per expression) is the + only query: the filtered frame's is the one ``populate_df_meta`` takes + anyway, and the raw frame changes only with the source or the + post-processing. The clean scope's frame is rebuilt on every cache miss, + so counting it would be a query of its own, and its sd has no + ``length``. Carries the same values full stats give those keys, so styling that reads only them renders the same. A column in ``skip_stat_columns`` gets only name, dtype and length, as at the full tier, so its typing comes from ``init_sd``. ``_get_summary_sd`` handles a pandas frame before it gets here.""" expr = cast("XorqExpr", processed_df) schema = expr.schema() - length = _expr_count(expr) + length = {} if scope == 'clean' else {'length': _expr_count(expr)} skip = getattr(self, 'skip_stat_columns', None) or () sd: dict = {} for orig_col, rewritten_col in old_col_new_col(expr): @@ -286,7 +291,7 @@ def _get_schema_sd(self, processed_df: "XorqExpr | pd.DataFrame") -> dict: typing = {'dtype': dtype} if orig_col in skip else schema_stats(dtype) sd[rewritten_col] = { 'orig_col_name': orig_col, 'rewritten_col_name': rewritten_col, - **typing, 'length': length} + **typing, **length} return sd From 9609772278ee56e7324670d4ec35d7ab7e10562f Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 13:18:48 -0400 Subject: [PATCH 13/14] fix(server): /load resets the session's stats policy, and its defaults are defined once (rows-first s1) /load clears the xorq state a /load_expr left but kept stats_tier and stats_delivery, which /load_expr reads as the defaults for a re-POST that omits them. A /load in between therefore handed an earlier load's deferred delivery to the next /load_expr. /load now resets the pair. The new-session defaults were hard-coded in LoadExprHandler as well as on SessionState. Both now read DEFAULT_STATS_TIER and DEFAULT_STATS_DELIVERY from session.py. Co-Authored-By: Claude Opus 5.5 --- buckaroo/server/handlers.py | 9 ++++++--- buckaroo/server/session.py | 7 +++++-- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/buckaroo/server/handlers.py b/buckaroo/server/handlers.py index ebafb74bd..86309ef07 100644 --- a/buckaroo/server/handlers.py +++ b/buckaroo/server/handlers.py @@ -16,7 +16,7 @@ from buckaroo.server.focus import find_or_create_session_window from buckaroo.dataflow.dataflow import STATS_TIERS from buckaroo.server.session import ( - STATS_DELIVERIES, build_state_message, dataflow_stats_tier) + DEFAULT_STATS_DELIVERY, DEFAULT_STATS_TIER, STATS_DELIVERIES, build_state_message, dataflow_stats_tier) from buckaroo.server import telemetry from buckaroo.pluggable_analysis_framework import perf_log @@ -326,6 +326,9 @@ async def post(self): session.backend = backend session.xorq_dataflow = None session.expr = None + # The stats policy is /load_expr's too, so a later /load_expr that + # omits it starts from the defaults rather than an earlier load's. + session.stats_tier, session.stats_delivery = DEFAULT_STATS_TIER, DEFAULT_STATS_DELIVERY session.prompt = prompt if component_config: session.component_config = component_config @@ -510,8 +513,8 @@ async def post(self): # keeps the session's, and one that changes it rebuilds. stats_tier, stats_delivery, policy_error = _stats_policy_from_body( body, - existing.stats_tier if existing is not None else "full", - existing.stats_delivery if existing is not None else "inline") + existing.stats_tier if existing is not None else DEFAULT_STATS_TIER, + existing.stats_delivery if existing is not None else DEFAULT_STATS_DELIVERY) if policy_error is not None: self.set_status(400) self.write(policy_error) diff --git a/buckaroo/server/session.py b/buckaroo/server/session.py index 4602cff07..a687dbed8 100644 --- a/buckaroo/server/session.py +++ b/buckaroo/server/session.py @@ -15,6 +15,9 @@ # dataflow constructor (today's behaviour); ``deferred`` builds the schema tier # first and leaves the rest to later requests. STATS_DELIVERIES = ("inline", "deferred") +# The policy of a session no /load_expr has set one for. +DEFAULT_STATS_TIER = "full" +DEFAULT_STATS_DELIVERY = "inline" def dataflow_stats_tier(stats_tier: str, stats_delivery: str) -> str: @@ -59,8 +62,8 @@ class SessionState: # there (see STATS_DELIVERIES). Kept apart from dataflow_kwargs, which is # splatted into the dataflow constructor, and replayed by /reload_expr. The # tier the dataflow is built at follows from the pair (dataflow_stats_tier). - stats_tier: str = "full" - stats_delivery: str = "inline" + stats_tier: str = DEFAULT_STATS_TIER + stats_delivery: str = DEFAULT_STATS_DELIVERY # Companion telemetry sink (#943): a fire-and-forget POST callable, built # once from the /load_expr payload's telemetry_url on the IOLoop (where # make_http_sink captures AsyncHTTPClient/IOLoop.current()). Stored here so From c61b22111f08b676164d9440ddd58c9516552057 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Tue, 6 Oct 2026 13:18:56 -0400 Subject: [PATCH 14/14] refactor: the infinite widget reuses the dataflow's builders, and test imports move to module level (rows-first s1) BuckarooInfiniteWidget._handle_widget_change repeated the df_data_dict and df_display_args assembly this PR moved into _build_df_data_dict and _build_df_display_args. It now calls them. Its dataflow is built with skip_main_serial, so 'main' stays empty as before. assemble_merged_sd, Expr and XorqStatPipeline were imported inside test functions, against the repo's rule. They are now module-level imports. Co-Authored-By: Claude Opus 5.5 --- buckaroo/buckaroo_widget.py | 39 +++---------------- .../dataflow/scoped_summary_stats_test.py | 5 +-- tests/unit/server/test_load_expr.py | 6 +-- 3 files changed, 8 insertions(+), 42 deletions(-) diff --git a/buckaroo/buckaroo_widget.py b/buckaroo/buckaroo_widget.py index 758b1b0f2..28c0225d9 100644 --- a/buckaroo/buckaroo_widget.py +++ b/buckaroo/buckaroo_widget.py @@ -31,7 +31,7 @@ from .serialization_utils import EMPTY_DF_WHOLE, check_and_fix_df, pd_to_obj, to_parquet, send_infinite_resp from .dataflow.dataflow import CustomizableDataflow from .dataflow.dataflow_extras import (Sampling, exception_protect) -from .dataflow.styling_core import (ComponentConfig, InitSD, OverrideColumnConfig, PinnedRowConfig, StylingAnalysis, merge_column_config, EMPTY_DFVIEWER_CONFIG) +from .dataflow.styling_core import (ComponentConfig, InitSD, OverrideColumnConfig, PinnedRowConfig, StylingAnalysis, EMPTY_DFVIEWER_CONFIG) from .dataflow.autocleaning import PandasAutocleaning from pathlib import Path @@ -359,39 +359,10 @@ def _handle_widget_change(self, change): _bk_flash("_handle_widget_change EXIT — processed_df is None") return - # df_data_dict is still hardcoded for now - # eventually processed_df will be able to add or alter values of df_data_dict - # correlation would be added, filtered would probably be altered - - # to expedite processing maybe future provided dfs from - # postprcoessing could default to empty until that is - # selected, optionally - - #note this needs to be empty so that we can do the infinite stuff - self.df_data_dict = {'main': [], - 'all_stats': self._sd_to_jsondf(merged_sd), - 'empty': []} - - temp_display_args = {} - for display_name, A_Klass in self.dataflow.df_display_klasses.items(): - df_viewer_config = A_Klass.get_dfviewer_config(merged_sd, processed_df) - base_column_config = df_viewer_config['column_config'] - df_viewer_config['column_config'] = merge_column_config( - base_column_config, self.dataflow.processed_df, self.dataflow.column_config_overrides) - disp_arg = {'data_key': A_Klass.data_key, - #'df_viewer_config': json.loads(json.dumps(df_viewer_config)), - 'df_viewer_config': df_viewer_config, - 'summary_stats_key': A_Klass.summary_stats_key} - temp_display_args[display_name] = disp_arg - - if self.dataflow.pinned_rows is not None: - temp_display_args['main']['df_viewer_config']['pinned_rows'] = self.dataflow.pinned_rows - if self.dataflow.extra_grid_config: - temp_display_args['main']['df_viewer_config']['extra_grid_config'] = self.dataflow.extra_grid_config - if self.dataflow.component_config: - temp_display_args['main']['df_viewer_config']['component_config'] = self.dataflow.component_config - - self.df_display_args = temp_display_args + # The dataflow is built with skip_main_serial, so 'main' stays empty for + # the infinite rows. + self.df_data_dict = self.dataflow._build_df_data_dict(processed_df, merged_sd) + self.df_display_args = self.dataflow._build_df_display_args(processed_df, merged_sd) _bk_flash("_handle_widget_change EXIT (df_display_args → JS)") diff --git a/tests/unit/dataflow/scoped_summary_stats_test.py b/tests/unit/dataflow/scoped_summary_stats_test.py index a8b0a5339..1e990c58e 100644 --- a/tests/unit/dataflow/scoped_summary_stats_test.py +++ b/tests/unit/dataflow/scoped_summary_stats_test.py @@ -24,7 +24,7 @@ from buckaroo.customizations.pd_autoclean_conf import NoCleaningConf from buckaroo.customizations.pd_stats_v2 import PD_ANALYSIS_V2, PD_AUTOCLEAN_DEFAULT_V2, cleaning_gen_ops from buckaroo.dataflow.autocleaning import (AutocleaningConfig, PandasAutocleaning) -from buckaroo.dataflow.dataflow import CustomizableDataflow, StylingAnalysis +from buckaroo.dataflow.dataflow import CustomizableDataflow, StylingAnalysis, assemble_merged_sd from buckaroo.dataflow.sd_cache import split_chain_by_scope from buckaroo.pluggable_analysis_framework.stat_func import stat @@ -279,7 +279,6 @@ def test_assemble_merged_sd_equals_merged_sd_with_init_sd_and_cleaning(): """``assemble_merged_sd`` is the body of the ``merged_sd`` observer as a pure function, so a server can assemble wire stats from an accumulating raw sd without going through the traitlets cascade.""" - from buckaroo.dataflow.dataflow import assemble_merged_sd dfc = _assembly_dataflow() sd = dfc.merged_sd['a'] assert sd['init_only'] == 1, "init_sd override must reach merged_sd under the rewritten name" @@ -291,7 +290,6 @@ def test_assemble_merged_sd_equals_merged_sd_with_init_sd_and_cleaning(): def test_assemble_merged_sd_without_a_processed_df_merges_the_unrewritten_sds(): - from buckaroo.dataflow.dataflow import assemble_merged_sd assembled = assemble_merged_sd( init_sd={'x': {'init': 1}}, cleaned_sd={'x': {'cleaned': 2}}, raw_sd={'x': {'raw': 3}}, processed_sd={'x': {'processed': 4}}, processed_df=None, @@ -300,7 +298,6 @@ def test_assemble_merged_sd_without_a_processed_df_merges_the_unrewritten_sds(): def test_assemble_merged_sd_does_not_mutate_its_inputs(): - from buckaroo.dataflow.dataflow import assemble_merged_sd dfc = _assembly_dataflow() inputs = _scope_inputs(dfc) sd_names = ('init_sd', 'cleaned_sd', 'raw_sd', 'processed_sd', 'clean_sd', 'filt_sd') diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 9e59caf4a..65872e3a7 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -25,10 +25,12 @@ from xorq.common.utils.graph_utils import replace_nodes, walk_nodes # noqa: E402 from xorq.common.utils.provenance_utils import read_parquet_provenance # noqa: E402 from xorq.expr.relations import CachedNode # noqa: E402 +from xorq.vendor.ibis.expr.types.core import Expr # noqa: E402 from buckaroo.dataflow.sd_cache import split_chain_by_scope # noqa: E402 from buckaroo.jlisp.lisp_utils import s as lisp_sym # noqa: E402 from buckaroo.pluggable_analysis_framework.col_analysis import ColAnalysis # noqa: E402 +from buckaroo.pluggable_analysis_framework.xorq_stat_pipeline import XorqStatPipeline # noqa: E402 from buckaroo.server import telemetry, xorq_loading # noqa: E402 from buckaroo.server.app import make_app as _make_app # noqa: E402 @@ -1028,9 +1030,6 @@ def _three_scope_dataflow(expr=None, **kwargs): def _spy_data_queries(monkeypatch): """Record the outermost op of every materialisation (``execute`` and ``to_pyarrow``), and every stat query XorqStatPipeline sends.""" - from xorq.vendor.ibis.expr.types.core import Expr - - from buckaroo.pluggable_analysis_framework.xorq_stat_pipeline import XorqStatPipeline ops, stat_queries = [], [] original_execute, original_to_pyarrow = Expr.execute, Expr.to_pyarrow original_stat_execute = XorqStatPipeline._execute @@ -1343,7 +1342,6 @@ async def test_deferred_delivery_publishes_a_schema_dataflow_and_serves_rows(sel @tornado.testing.gen_test async def test_deferred_delivery_runs_no_stat_query(self): - from buckaroo.pluggable_analysis_framework.xorq_stat_pipeline import XorqStatPipeline builds_root = tempfile.mkdtemp() stat_queries = [] original = XorqStatPipeline._execute