From cef1f8a0c5930e43bb717246c1007c3cd24f50e8 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Fri, 22 May 2026 17:03:33 -0400 Subject: [PATCH 1/3] feat(server): /load gains backend=xorq for XorqInfiniteBuckaroo over parquet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lets `/load` route a parquet file through the same xorq push-down path that `/load_expr` already uses for build-dir-driven expressions. Removes the build-dir requirement when a host already has a materialised parquet on disk (e.g. a desktop that materialised a catalog entry via `xorq catalog run`) and wants `XorqInfiniteBuckaroo` viewing — constant-memory grid scrolls via xorq query execution, instead of `pd.read_parquet`-ing the whole file and computing summary stats up front. ## What's new `POST /load` now accepts an optional `backend` field: - `"pandas"` (default): unchanged. `pd.read_parquet` + eager stats; same behaviour as before. - `"xorq"`: wraps the file in `xo.deferred_read_parquet(...)` and hands the resulting expression to `XorqServerDataflow`. Server then answers every `infinite_request` from the grid via `handle_infinite_request_xorq`. Requires `mode == "buckaroo"` (rejected with 400 otherwise) and `buckaroo[xorq]` (501 if missing). A new helper `xorq_loading.load_expr_parquet_path(path)` is the counterpart to `load_expr_build_dir`. It uses `xo.deferred_read_parquet` so the file is *never* eagerly read on the server — schema + row count come from the lazy-evaluated expression. ## Why Hosts that have already materialised a snapshot parquet (xorq-desktop is the immediate example, via `catalog_run_parquet`) currently have to choose between: - `mode="buckaroo"` (pandas eager) — easy to wire, but 5 GB on disk becomes ~7 GB resident in the sidecar's Python process + heavy stats-up-front cost on first scroll. - `mode="lazy"` (polars `scan_parquet`) — lazy, but loses the xorq semantics (no push-down beyond polars). - `/load_expr` — gets `XorqInfiniteBuckaroo`, but the host has to keep the entry's `build_dir` around, not just the materialised parquet. This bridges the gap: a path is enough to opt into XorqInfiniteBuckaroo. ## Verified End-to-end against the running sidecar with a real 20M-row parquet: curl -X POST http://127.0.0.1:/load -d '{ "path": "", "mode": "buckaroo", "backend": "xorq", "no_browser": true }' Returns session id, row count, and column schema; no upfront materialisation; subsequent WS `infinite_request`s flow through `handle_infinite_request_xorq` exactly as they do for `/load_expr`. Co-Authored-By: Claude Opus 4.7 (1M context) --- buckaroo/server/handlers.py | 100 ++++++++++++++++++++++++++++++-- buckaroo/server/xorq_loading.py | 26 +++++++++ 2 files changed, 122 insertions(+), 4 deletions(-) diff --git a/buckaroo/server/handlers.py b/buckaroo/server/handlers.py index bbbe93f67..efc452fd8 100644 --- a/buckaroo/server/handlers.py +++ b/buckaroo/server/handlers.py @@ -233,13 +233,30 @@ async def post(self): if session_id is None: return + # New since the path-only /load_expr distinction: callers can opt + # /load into the xorq push-down backend (XorqInfiniteBuckaroo) + # without having a build_dir handy — they provide a parquet path + # and we wrap it in a deferred-read expression. Default stays + # "pandas" so existing callers are unaffected. + backend_arg = body.get("backend", "pandas") + if backend_arg not in ("pandas", "xorq"): + self.set_status(400) + self.write({"error_code": "invalid_backend", + "message": f"backend must be 'pandas' or 'xorq' (got {backend_arg!r})"}) + return + if backend_arg == "xorq" and mode != "buckaroo": + self.set_status(400) + self.write({"error_code": "invalid_mode_for_backend", + "message": "backend='xorq' requires mode='buckaroo' (XorqInfiniteBuckaroo)"}) + return + sessions = self.application.settings["sessions"] session = sessions.get_or_create(session_id, path) session.mode = mode - # Loading via /load is always pandas — clear any xorq state left - # by a prior /load_expr on the same session so WS dispatch routes - # to the new pandas dataflow rather than a stale xorq one. - session.backend = "pandas" + session.backend = backend_arg + # Reset any xorq/pandas state left by a prior load on the same + # session so WS dispatch routes to the new dataflow rather than a + # stale one. The branches below repopulate the relevant fields. session.xorq_dataflow = None session.expr = None # Reset the live-typed row-fetch filter so a search term carried @@ -251,6 +268,81 @@ async def post(self): if component_config: session.component_config = component_config + if backend_arg == "xorq": + # XorqInfiniteBuckaroo over a materialised parquet. Mirrors + # LoadExprHandler's xorq-branch session setup but sourced from + # a file path instead of a build_dir. + try: + from buckaroo.server import xorq_loading # noqa: PLC0415 + except ImportError: + self.set_status(501) + self.write({"error_code": "xorq_not_installed", + "message": "xorq is not installed on this server. " + "Install with `pip install buckaroo[xorq]`."}) + return + + if not os.path.exists(path): + self.set_status(404) + self.write({"error_code": "file_not_found", + "message": f"File not found: {path}"}) + return + + try: + expr = xorq_loading.load_expr_parquet_path(path) + xorq_dataflow = xorq_loading.XorqServerDataflow( + expr, skip_main_serial=True) + metadata = xorq_loading.get_xorq_metadata(xorq_dataflow, path) + except Exception: + tb = traceback.format_exc() + log.error("load (xorq) error path=%s: %s", path, tb) + resp: dict = {"error_code": "load_error", + "message": "Failed to load parquet via xorq backend"} + if _BUCKAROO_DEBUG: + resp["details"] = tb + self.set_status(500) + self.write(resp) + return + + session.expr = expr + session.xorq_dataflow = xorq_dataflow + session.df = None + session.dataflow = None + session.ldf = None + session.metadata = metadata + session.df_display_args = xorq_dataflow.df_display_args + session.df_data_dict = xorq_dataflow.df_data_dict + session.df_meta = xorq_dataflow.df_meta + session.buckaroo_state = { + "cleaning_method": "", "post_processing": "", "sampled": False, + "show_commands": False, "df_display": "main", + "search_string": "", "quick_command_args": {}} + session.buckaroo_options = xorq_dataflow.buckaroo_options + session.command_config = xorq_dataflow.command_config + session.operation_results = { + "transformed_df": {"schema": {"fields": []}, "data": []}, + "generated_py_code": "# server mode (xorq backend via /load)"} + session.operations = [] + + if component_config and session.df_display_args: + for key in session.df_display_args: + dvc = session.df_display_args[key].get("df_viewer_config") + if dvc is not None: + dvc["component_config"] = { + **dvc.get("component_config", {}), + **component_config, + } + + self._push_state_to_clients(session, metadata) + browser_action = "skipped" if no_browser else self._handle_browser_window(session_id) + + log.info("load session=%s path=%s rows=%d backend=xorq browser=%s", + session_id, path, metadata["rows"], browser_action) + self.write({"session": session_id, "server_pid": os.getpid(), + "browser_action": browser_action, **metadata}) + return + + # Pandas-default / lazy-polars path. Identical to before — the + # session.backend assignment above already set "pandas". # Load data in appropriate mode file_obj, metadata = self._load_file_with_error_handling(path, is_lazy=(mode == "lazy")) if file_obj is None: diff --git a/buckaroo/server/xorq_loading.py b/buckaroo/server/xorq_loading.py index 5765920d2..51e478ac7 100644 --- a/buckaroo/server/xorq_loading.py +++ b/buckaroo/server/xorq_loading.py @@ -82,6 +82,32 @@ def get_xorq_metadata(xorq_dataflow: XorqServerDataflow, build_dir: str) -> dict return {"path": build_dir, "rows": _expr_count(expr), "columns": columns} +def load_expr_parquet_path(path: str): + """Wrap a local parquet file in a xorq deferred-read expression. + + Counterpart to ``load_expr_build_dir`` for the case where the caller + holds a materialised parquet (e.g. a host that materialised a + catalog entry to a snapshot file via ``xorq catalog run``) and wants + XorqInfiniteBuckaroo's push-down query behaviour rather than the + eager pandas/polars load that ``data_loading.load_file`` does. + + ``deferred_read_parquet`` wires the file into xorq's datafusion + backend without materialising it; ``XorqServerDataflow`` then takes + that expression and answers every ``infinite_request`` via + ``handle_infinite_request_xorq`` (the same code path the + build-dir-driven xorq loader uses). + """ + from pathlib import Path # noqa: PLC0415 + + from xorq.api import connect, deferred_read_parquet # noqa: PLC0415 + from xorq.vendor import ibis # noqa: PLC0415 + + if ibis.options.default_backend is None: + ibis.options.default_backend = connect() + table_name = Path(path).stem.replace("-", "_") or "parquet" + return deferred_read_parquet(path, table_name=table_name) + + # --------------------------------------------------------------------------- # project-authored summary stats (loaded from /stats/*.py) # --------------------------------------------------------------------------- From fc07f17637522f280377da312fb20d09709b8fdb Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sun, 24 May 2026 14:23:57 -0400 Subject: [PATCH 2/3] test(server): cover /load backend=xorq + minor cleanups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses review feedback on #840: - Add 3 coverage tests in `TestLoadBackendXorq` against the new branch: 400 invalid backend, 400 invalid mode/backend combo, and the happy-path 200 + WS `infinite_request` round-trip proving dispatch routes through `handle_infinite_request_xorq`. - Hoist the redundant `from pathlib import Path` out of `load_expr_parquet_path` — `Path` is already imported at module top (xorq_loading.py:15). - Trim the multi-line comment introducing the `backend` arg parse in `LoadHandler.post` to one short paragraph; the PR body carries the long-form rationale. The collision concern raised in review (two parquets with the same stem registering against the shared datafusion default backend) was investigated and dropped — `xorq.deferred_read_parquet` binds the source path into the expression, so re-registering under the same `table_name` does not cross-contaminate. Verified with two distinct schemas in the same process. Co-Authored-By: Claude Opus 4.7 (1M context) --- buckaroo/server/handlers.py | 8 ++-- buckaroo/server/xorq_loading.py | 2 - tests/unit/server/test_load_expr.py | 64 +++++++++++++++++++++++++++++ 3 files changed, 67 insertions(+), 7 deletions(-) diff --git a/buckaroo/server/handlers.py b/buckaroo/server/handlers.py index efc452fd8..55bdf538b 100644 --- a/buckaroo/server/handlers.py +++ b/buckaroo/server/handlers.py @@ -233,11 +233,9 @@ async def post(self): if session_id is None: return - # New since the path-only /load_expr distinction: callers can opt - # /load into the xorq push-down backend (XorqInfiniteBuckaroo) - # without having a build_dir handy — they provide a parquet path - # and we wrap it in a deferred-read expression. Default stays - # "pandas" so existing callers are unaffected. + # Optional backend="xorq" routes /load through the same push-down + # path as /load_expr but sourced from a parquet rather than a + # build_dir. Default "pandas" keeps existing callers unaffected. backend_arg = body.get("backend", "pandas") if backend_arg not in ("pandas", "xorq"): self.set_status(400) diff --git a/buckaroo/server/xorq_loading.py b/buckaroo/server/xorq_loading.py index 51e478ac7..9fada704d 100644 --- a/buckaroo/server/xorq_loading.py +++ b/buckaroo/server/xorq_loading.py @@ -97,8 +97,6 @@ def load_expr_parquet_path(path: str): ``handle_infinite_request_xorq`` (the same code path the build-dir-driven xorq loader uses). """ - from pathlib import Path # noqa: PLC0415 - from xorq.api import connect, deferred_read_parquet # noqa: PLC0415 from xorq.vendor import ibis # noqa: PLC0415 diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 73d32f7b9..8115ae1ed 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -373,3 +373,67 @@ async def test_session_reuse_xorq_then_pandas(self): finally: shutil.rmtree(builds_root, ignore_errors=True) os.unlink(csv_path) + + +class TestLoadBackendXorq(tornado.testing.AsyncHTTPTestCase): + """POST /load with backend="xorq" — wraps a parquet path in a + deferred-read xorq expression and serves it through the same + push-down path as /load_expr.""" + + def get_app(self): + return make_app() + + def test_invalid_backend_returns_400(self): + resp = self.fetch("/load", method="POST", + body=json.dumps({"session": "lb-bad", "path": "/tmp/x.parquet", + "backend": "bogus"}), + headers={"Content-Type": "application/json"}) + self.assertEqual(resp.code, 400) + self.assertEqual(json.loads(resp.body)["error_code"], "invalid_backend") + + def test_xorq_backend_requires_buckaroo_mode(self): + resp = self.fetch("/load", method="POST", + body=json.dumps({"session": "lb-mode", "path": "/tmp/x.parquet", + "backend": "xorq", "mode": "lazy"}), + headers={"Content-Type": "application/json"}) + self.assertEqual(resp.code, 400) + self.assertEqual(json.loads(resp.body)["error_code"], + "invalid_mode_for_backend") + + @tornado.testing.gen_test + async def test_load_parquet_via_xorq_backend(self): + """Happy path: POST /load with backend=xorq, then issue an + infinite_request — the row count must come from the parquet via + XorqServerDataflow, proving routing went through the xorq path.""" + d = tempfile.mkdtemp() + parquet_path = os.path.join(d, "lb_fixture.parquet") + try: + pd.DataFrame({"idx": list(range(7)), "name": ["a", "b", "c", "d", "e", "f", "g"]}).to_parquet(parquet_path) + + resp = await _post(self.get_http_port(), "/load", + {"session": "lb-ok", "path": parquet_path, + "mode": "buckaroo", "backend": "xorq"}) + self.assertEqual(resp.code, 200) + body = json.loads(resp.body) + self.assertEqual(body["session"], "lb-ok") + self.assertEqual(body["rows"], 7) + self.assertEqual({c["name"] for c in body["columns"]}, {"idx", "name"}) + + ws = await tornado.websocket.websocket_connect( + f"ws://localhost:{self.get_http_port()}/ws/lb-ok") + await ws.read_message() # initial_state + + ws.write_message(json.dumps({ + "type": "infinite_request", + "payload_args": {"start": 0, "end": 10, + "sourceName": "default", "origEnd": 10}})) + r = json.loads(await ws.read_message()) + self.assertEqual(r["type"], "infinite_resp") + self.assertEqual(r["length"], 7) + + binary_frame = await ws.read_message() + table = pq.read_table(io.BytesIO(binary_frame)) + self.assertEqual(table.num_rows, 7) + ws.close() + finally: + shutil.rmtree(d, ignore_errors=True) From 2e557a4b627467adc999231f4631040a5d3828e1 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sun, 24 May 2026 17:42:46 -0400 Subject: [PATCH 3/3] test(server): pin codex feedback on /load backend=xorq (#840) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two failing tests reproducing Codex's review on #840: - ``test_xorq_load_failure_preserves_session_backend`` — pandas /load, then xorq /load with a non-existent path; asserts session.backend stays "pandas" so the WS state-change dispatch (websocket_handler.py:59) still reaches a real dataflow. - ``test_xorq_not_installed_returns_501`` — hides ``xorq.api`` via ``sys.modules[..] = None`` and asserts the handler's documented 501 ``xorq_not_installed`` branch is reachable. Uses a real parquet path so the result is unambiguous (not masked as 404). Tests-only — fix lands in a follow-up commit per repo TDD policy (test must be seen failing on CI before the fix lands). Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/unit/server/test_load_expr.py | 73 +++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/tests/unit/server/test_load_expr.py b/tests/unit/server/test_load_expr.py index 8115ae1ed..a2e06923d 100644 --- a/tests/unit/server/test_load_expr.py +++ b/tests/unit/server/test_load_expr.py @@ -400,6 +400,79 @@ def test_xorq_backend_requires_buckaroo_mode(self): self.assertEqual(json.loads(resp.body)["error_code"], "invalid_mode_for_backend") + @tornado.testing.gen_test + async def test_xorq_load_failure_preserves_session_backend(self): + """Codex P1 on #840: a failed xorq /load must not flip + session.backend to 'xorq' before the load is known to succeed. + Otherwise a session previously serving pandas ends up with + backend='xorq' + xorq_dataflow=None, and the WS state-change + handler silently drops further updates (websocket_handler.py:59).""" + d = tempfile.mkdtemp() + good_path = os.path.join(d, "good.parquet") + bogus_path = os.path.join(d, "nonexistent.parquet") + try: + pd.DataFrame({"idx": list(range(3))}).to_parquet(good_path) + + # 1. Successful pandas load establishes prior session state. + resp = await _post(self.get_http_port(), "/load", + {"session": "lb-roll", "path": good_path, "mode": "buckaroo"}) + self.assertEqual(resp.code, 200) + + sessions = self._app.settings["sessions"] + session = sessions.get("lb-roll") + self.assertEqual(session.backend, "pandas") + prior_dataflow = session.dataflow + self.assertIsNotNone(prior_dataflow) + + # 2. Failed xorq load (path doesn't exist → 404). + resp = await _post(self.get_http_port(), "/load", + {"session": "lb-roll", "path": bogus_path, + "mode": "buckaroo", "backend": "xorq"}) + self.assertEqual(resp.code, 404) + + # 3. Session must not have been half-mutated. backend stays + # 'pandas'; the existing pandas dataflow stays reachable so + # WS dispatch can still answer buckaroo_state_change. + self.assertEqual(session.backend, "pandas", + "session.backend must not flip to 'xorq' on failed xorq load") + self.assertIs(session.dataflow, prior_dataflow, + "session.dataflow must survive a failed xorq load") + self.assertIsNone(session.xorq_dataflow) + finally: + shutil.rmtree(d, ignore_errors=True) + + def test_xorq_not_installed_returns_501(self): + """Codex P2 on #840: when xorq.api is not importable the handler + must return 501 xorq_not_installed, not a generic 500 load_error. + The probe in handlers.py has to be explicit — importing + ``buckaroo.server.xorq_loading`` succeeds even without xorq + because the transitive ``import xorq.api`` calls in + ``xorq_stats_v2`` and ``xorq_stat_pipeline`` are guarded with + try/except. + + Uses a real parquet path so the path-exists check passes; that + way a 500 result here proves the bug isn't masked as file-not- + found and is genuinely the unreachable 501 branch.""" + from unittest.mock import patch + d = tempfile.mkdtemp() + parquet_path = os.path.join(d, "p.parquet") + try: + pd.DataFrame({"idx": [0]}).to_parquet(parquet_path) + # sys.modules[name] = None forces a subsequent `import name` + # to raise ImportError. patch.dict restores the mapping so + # other tests still see the real xorq.api. + with patch.dict(sys.modules, {"xorq.api": None}): + resp = self.fetch("/load", method="POST", + body=json.dumps({"session": "lb-noxorq", + "path": parquet_path, + "mode": "buckaroo", "backend": "xorq"}), + headers={"Content-Type": "application/json"}) + self.assertEqual(resp.code, 501) + self.assertEqual(json.loads(resp.body)["error_code"], + "xorq_not_installed") + finally: + shutil.rmtree(d, ignore_errors=True) + @tornado.testing.gen_test async def test_load_parquet_via_xorq_backend(self): """Happy path: POST /load with backend=xorq, then issue an