From bb932c63c5c0f0c1a4d44d4891f323dfb8c2dd25 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 11 Jun 2026 15:14:34 -0400 Subject: [PATCH 1/8] test(stats): failing tests for smoke frame suite + measurements report (#920) --- tests/unit/perf_smoke_test.py | 42 ++++++++++++++++++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/tests/unit/perf_smoke_test.py b/tests/unit/perf_smoke_test.py index 1248487a9..a70066f22 100644 --- a/tests/unit/perf_smoke_test.py +++ b/tests/unit/perf_smoke_test.py @@ -8,8 +8,10 @@ """ import numpy as np +import pandas as pd -from buckaroo.pluggable_analysis_framework.perf_smoke import perf_smoke_test +from buckaroo.pluggable_analysis_framework.perf_smoke import ( + SMOKE_FRAME_MAKERS, measurements_markdown, perf_smoke_test, run_perf_smoke) from buckaroo.pluggable_analysis_framework.stat_func import stat, RawSeries @@ -64,3 +66,41 @@ def memory_hog(ser: RawSeries) -> float: mem_findings = [f for f in findings if f.kind == 'memory' and f.stat_name == 'memory_hog'] assert mem_findings assert 'memory_hog' in mem_findings[0].message + + +def test_smoke_frames_cover_known_killers(): + # Shapes with bug history: unhashable cells (#843), int64 past 2^53 + # (#800/#632), Decimal (#801), tz-aware datetimes (#277), all-null, + # categorical. + expected = {'base', 'unhashable', 'big_int64', 'decimal', 'tz_datetime', + 'all_null', 'categorical'} + assert expected <= set(SMOKE_FRAME_MAKERS) + frames = {name: maker(100) for name, maker in SMOKE_FRAME_MAKERS.items()} + for name, df in frames.items(): + assert len(df) == 100, name + assert isinstance(frames['unhashable'].iloc[0, 0], list) + assert (frames['big_int64'] > 2**53).any().any() + assert frames['all_null'].isna().all().all() + assert isinstance(frames['categorical'].dtypes.iloc[0], pd.CategoricalDtype) + + +def test_findings_name_the_frame(): + @stat() + def memory_hog_frames(ser: RawSeries) -> float: + scratch = np.ones((len(ser), 4_000)) + return float(scratch.sum()) + + passed, findings = perf_smoke_test([memory_hog_frames], rows=2_000) + assert passed is False + finding = findings[0] + assert finding.frame in SMOKE_FRAME_MAKERS + assert finding.frame in finding.message + + +def test_run_perf_smoke_measurements_and_markdown(): + result = run_perf_smoke([smoke_length], rows=2_000) + assert result.passed is True + assert {m.frame for m in result.measurements} == set(SMOKE_FRAME_MAKERS) + md = measurements_markdown(result.measurements) + assert 'smoke_length' in md + assert md.startswith('|') From c296d6661d04138cbeaccd6afe8c7b84633a682a Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 11 Jun 2026 15:21:28 -0400 Subject: [PATCH 2/8] feat(stats): smoke frame suite, per-call measurements, CI perf report (#920) --- .github/workflows/checks.yml | 19 ++ .../perf_smoke.py | 191 +++++++++++++++--- scripts/perf_smoke_report.py | 39 ++++ 3 files changed, 217 insertions(+), 32 deletions(-) create mode 100644 scripts/perf_smoke_report.py diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index f4efc3fa8..df6db2f63 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -308,6 +308,25 @@ jobs: run: scripts/run_pytest_tolerant.sh uv run --with pytest-cov pytest ./tests/unit -m "not slow" --color=yes --cov anywidget --cov-report xml - uses: codecov/codecov-action@v6 + StatsPerfSmoke: + name: Stats / Perf Smoke Report + runs-on: depot-ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v6 + - name: Install uv + uses: astral-sh/setup-uv@v7 + with: + enable-cache: true + python-version: "3.13" + prune-cache: false + - name: Install the project + run: uv sync --all-extras --dev + - name: Run perf smoke over shipped stats + # Per-stat worst-case time/memory table lands in the job summary; + # the job only fails on the loose smoke limits. See #920. + run: uv run python scripts/perf_smoke_report.py + TestPythonMaxVersions: name: Python / Test (Max Versions) runs-on: depot-ubuntu-latest diff --git a/buckaroo/pluggable_analysis_framework/perf_smoke.py b/buckaroo/pluggable_analysis_framework/perf_smoke.py index 369541e74..d6725cb6f 100644 --- a/buckaroo/pluggable_analysis_framework/perf_smoke.py +++ b/buckaroo/pluggable_analysis_framework/perf_smoke.py @@ -7,8 +7,8 @@ surface as a hung widget or an OOM on real data. User- and LLM/MCP-authored stats (``add_analysis``, project stat dirs) are the usual source. -``perf_smoke_test()`` runs each stat against a seeded synthetic frame and -flags per-stat wall-time and peak-memory blowups:: +``perf_smoke_test()`` runs each stat against a suite of seeded synthetic +frames and flags per-stat wall-time and peak-memory blowups:: from buckaroo.pluggable_analysis_framework.perf_smoke import perf_smoke_test @@ -19,6 +19,15 @@ Pass the same stat list you would hand to a widget or pipeline — stats that ``require`` keys from other stats need their providers in the list. +The frame suite (``SMOKE_FRAME_MAKERS``) covers shapes with bug history: +unhashable object cells (#843), int64 beyond 2^53 (#800, #632), Decimal +columns (#801), tz-aware datetimes (#277), all-null columns, and categorical +dtype. Findings name the frame that triggered them. + +``run_perf_smoke()`` additionally returns every per-call measurement, and +``measurements_markdown()`` renders them as a per-stat worst-case table — +used by ``scripts/perf_smoke_report.py`` to write a CI job summary. + Design notes: - Limits are machine-relative: the time limit is a ratio against canonical @@ -40,7 +49,8 @@ import time import tracemalloc from dataclasses import dataclass -from typing import List, Tuple +from decimal import Decimal +from typing import Callable, Dict, List, Optional, Tuple import numpy as np import pandas as pd @@ -63,9 +73,58 @@ def make_smoke_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: 'str_high_card': [f'id_{i:08d}' for i in range(rows)]}) +def _make_unhashable_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # 843: cells holding lists/dicts break hash-based ops (nunique raises) and + # make others crawl (value_counts succeeds but takes seconds at 10k rows) + return pd.DataFrame({'list_cells': [[i, i + 1] for i in range(rows)], + 'dict_cells': [{'k': i} for i in range(rows)]}) + + +def _make_big_int64_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # 800/632: int64 beyond 2^53 silently loses precision through float casts + return pd.DataFrame({'beyond_2_53': np.arange(rows, dtype=np.int64) + 2**53 + 1, + 'int64_extremes': np.where(np.arange(rows) % 2 == 0, + np.int64(2**63 - 1), np.int64(-(2**63) + 1))}) + + +def _make_decimal_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # 801: Decimal cells are object dtype and break numeric fast paths + return pd.DataFrame({'decimal_col': [Decimal(i) / Decimal(7) for i in range(rows)]}) + + +def _make_tz_datetime_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # 277: tz-aware datetimes break naive datetime arithmetic/serialization + return pd.DataFrame({ + 'utc_ts': pd.date_range('2020-01-01', periods=rows, freq='min', tz='UTC'), + 'ny_ts': pd.date_range('2020-01-01', periods=rows, freq='min', + tz='America/New_York')}) + + +def _make_all_null_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + return pd.DataFrame({'none_obj': pd.Series([None] * rows, dtype='object'), + 'nan_float': pd.Series(np.full(rows, np.nan)), + 'nat_ts': pd.Series([pd.NaT] * rows)}) + + +def _make_categorical_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + rng = np.random.default_rng(seed) + return pd.DataFrame({ + 'cat_col': pd.Categorical.from_codes(rng.integers(0, 5, rows), + categories=[f'c{i}' for i in range(5)]), + 'ordered_cat': pd.Categorical.from_codes(rng.integers(0, 3, rows), + categories=['low', 'mid', 'high'], + ordered=True)}) + + +SMOKE_FRAME_MAKERS: Dict[str, Callable[..., pd.DataFrame]] = {'base': make_smoke_df, 'unhashable': _make_unhashable_df, + 'big_int64': _make_big_int64_df, 'decimal': _make_decimal_df, 'tz_datetime': _make_tz_datetime_df, + 'all_null': _make_all_null_df, 'categorical': _make_categorical_df} + + @dataclass class SmokeFinding: stat_name: str + frame: str column: str kind: str # 'time' | 'memory' measured: float # seconds for 'time', bytes for 'memory' @@ -74,20 +133,43 @@ class SmokeFinding: message: str +@dataclass +class SmokeMeasurement: + """One instrumented stat call: which stat, on which frame/column/scale, + and what it cost. Collected for every call, not just limit exceedances.""" + stat_name: str + frame: str + column: str + rows: int + seconds: float + peak_bytes: float + + +@dataclass +class SmokeResult: + passed: bool + findings: List[SmokeFinding] + measurements: List[SmokeMeasurement] + + class _SmokeSkip(Exception): """Raised inside a wrapped stat that was already flagged, so the pipeline records an Err instead of running the pathological code again.""" def _baseline_seconds(df: pd.DataFrame) -> float: - """Time canonical per-column pandas work on the smoke frame. A reasonable - stat does far less than this; the time limit is a multiple of it.""" + """Time canonical per-column pandas work on a smoke frame. A reasonable + stat does far less than this; the time limit is a multiple of it. Ops that + raise on a shape (value_counts on unhashable cells) are skipped — the + absolute floor keeps the limit sane when most ops skip.""" def work(): for col in df.columns: ser = df[col] - ser.value_counts() - ser.sort_values() - ser.nunique() + for op in (ser.value_counts, ser.sort_values, ser.nunique): + try: + op() + except Exception: + pass work() # warm caches so the measured pass isn't paying first-call cost t0 = time.perf_counter() @@ -96,12 +178,15 @@ def work(): class _Recorder: - """Shared state between perf_smoke_test and the wrapped stat funcs: - current column/scale/limits going in, findings and flagged stats out.""" + """Shared state between run_perf_smoke and the wrapped stat funcs: + current frame/column/scale/limits going in, measurements, findings and + flagged stats out.""" def __init__(self): self.findings: List[SmokeFinding] = [] + self.measurements: List[SmokeMeasurement] = [] self.flagged: set = set() + self.frame = '' self.column = '' self.rows = 0 self.time_limit = float('inf') @@ -113,24 +198,28 @@ def configure(self, rows: int, time_limit: float, memory_limit: float) -> None: self.memory_limit = memory_limit def record(self, stat_name: str, seconds: float, peak_bytes: float) -> None: + self.measurements.append(SmokeMeasurement( + stat_name=stat_name, frame=self.frame, column=self.column, + rows=self.rows, seconds=seconds, peak_bytes=peak_bytes)) if seconds > self.time_limit: self.flagged.add(stat_name) self.findings.append(SmokeFinding( - stat_name=stat_name, column=self.column, kind='time', measured=seconds, - limit=self.time_limit, rows=self.rows, + stat_name=stat_name, frame=self.frame, column=self.column, + kind='time', measured=seconds, limit=self.time_limit, rows=self.rows, message=( f"stat '{stat_name}' took {seconds:.2f}s on column '{self.column}' " - f"of a {self.rows}-row frame (limit {self.time_limit:.2f}s). " + f"of the {self.rows}-row '{self.frame}' frame (limit {self.time_limit:.2f}s). " f"Likely a per-row python loop or an O(n^2) algorithm — " f"vectorize with pandas/numpy operations."))) if peak_bytes > self.memory_limit: self.flagged.add(stat_name) self.findings.append(SmokeFinding( - stat_name=stat_name, column=self.column, kind='memory', measured=peak_bytes, - limit=self.memory_limit, rows=self.rows, + stat_name=stat_name, frame=self.frame, column=self.column, + kind='memory', measured=peak_bytes, limit=self.memory_limit, rows=self.rows, message=( f"stat '{stat_name}' allocated a peak of {peak_bytes / 1e6:.0f}MB on column " - f"'{self.column}' of a {self.rows}-row frame (limit {self.memory_limit / 1e6:.0f}MB). " + f"'{self.column}' of the {self.rows}-row '{self.frame}' frame " + f"(limit {self.memory_limit / 1e6:.0f}MB). " f"Avoid materializing large intermediates (pairwise matrices, " f"cross joins, full copies per row)."))) @@ -154,14 +243,16 @@ def measured(*args, **kwargs): return dataclasses.replace(sf, func=measured) -def perf_smoke_test(stat_funcs: list, rows: int = DEFAULT_ROWS, max_time_ratio: float = 20.0, - time_floor_seconds: float = 0.25, max_memory_ratio: float = 20.0, - memory_floor_bytes: float = MEMORY_FLOOR_BYTES) -> Tuple[bool, List[SmokeFinding]]: +def run_perf_smoke(stat_funcs: list, rows: int = DEFAULT_ROWS, + frames: Optional[Dict[str, Callable[..., pd.DataFrame]]] = None, + max_time_ratio: float = 20.0, time_floor_seconds: float = 0.25, + max_memory_ratio: float = 20.0, + memory_floor_bytes: float = MEMORY_FLOOR_BYTES) -> SmokeResult: """Smoke-test stats for perf/memory pathologies on synthetic data. Accepts the same mixed input as StatPipeline (StatFunc, @stat functions, - stat group classes, ColAnalysis subclasses). Returns - ``(passed, findings)`` mirroring ``unit_test()``'s shape. + stat group classes, ColAnalysis subclasses). ``frames`` maps frame name to + a ``maker(rows) -> DataFrame`` callable; defaults to SMOKE_FRAME_MAKERS. A stat is flagged when a single per-column call exceeds ``max(time_floor_seconds, max_time_ratio * baseline)`` wall-time or @@ -169,6 +260,7 @@ def perf_smoke_test(stat_funcs: list, rows: int = DEFAULT_ROWS, max_time_ratio: memory. Defaults are deliberately loose — this catches order-of-magnitude blowups, not 2x regressions. """ + frame_makers = SMOKE_FRAME_MAKERS if frames is None else frames recorder = _Recorder() wrapped = [_wrap(sf, recorder) for sf in _normalize_inputs(stat_funcs)] pipeline = StatPipeline(wrapped, unit_test=False) @@ -180,18 +272,53 @@ def perf_smoke_test(stat_funcs: list, rows: int = DEFAULT_ROWS, max_time_ratio: small = max(500, rows // 10) scales = [small, rows] if small < rows else [rows] for n in scales: - df = make_smoke_df(n) - baseline = _baseline_seconds(df) - recorder.configure( - rows=n, - time_limit=max(time_floor_seconds, max_time_ratio * baseline), - memory_limit=max(memory_floor_bytes, - max_memory_ratio * float(df.memory_usage(deep=True).sum()))) - for col in df.columns: - recorder.column = col - pipeline.process_df(df[[col]]) + for frame_name, maker in frame_makers.items(): + df = maker(n) + recorder.frame = frame_name + baseline = _baseline_seconds(df) + recorder.configure( + rows=n, + time_limit=max(time_floor_seconds, max_time_ratio * baseline), + memory_limit=max(memory_floor_bytes, + max_memory_ratio * float(df.memory_usage(deep=True).sum()))) + for col in df.columns: + recorder.column = col + pipeline.process_df(df[[col]]) finally: if not was_tracing: tracemalloc.stop() - return (not recorder.findings), recorder.findings + return SmokeResult(passed=(not recorder.findings), findings=recorder.findings, + measurements=recorder.measurements) + + +def perf_smoke_test(stat_funcs: list, rows: int = DEFAULT_ROWS, + frames: Optional[Dict[str, Callable[..., pd.DataFrame]]] = None, + max_time_ratio: float = 20.0, time_floor_seconds: float = 0.25, + max_memory_ratio: float = 20.0, + memory_floor_bytes: float = MEMORY_FLOOR_BYTES) -> Tuple[bool, List[SmokeFinding]]: + """``run_perf_smoke`` returning just ``(passed, findings)``, mirroring + ``unit_test()``'s shape.""" + result = run_perf_smoke(stat_funcs, rows=rows, frames=frames, + max_time_ratio=max_time_ratio, time_floor_seconds=time_floor_seconds, + max_memory_ratio=max_memory_ratio, memory_floor_bytes=memory_floor_bytes) + return result.passed, result.findings + + +def measurements_markdown(measurements: List[SmokeMeasurement]) -> str: + """Per-stat markdown table: worst wall-time and worst peak memory across + all frames/columns/scales, naming where each worst case happened.""" + worst_time: Dict[str, SmokeMeasurement] = {} + worst_mem: Dict[str, SmokeMeasurement] = {} + for m in measurements: + if m.stat_name not in worst_time or m.seconds > worst_time[m.stat_name].seconds: + worst_time[m.stat_name] = m + if m.stat_name not in worst_mem or m.peak_bytes > worst_mem[m.stat_name].peak_bytes: + worst_mem[m.stat_name] = m + lines = ['| stat | worst time | worst peak memory |', '| --- | --- | --- |'] + for name in sorted(worst_time): + wt, wm = worst_time[name], worst_mem[name] + lines.append( + f'| {name} | {wt.seconds * 1000:.1f}ms ({wt.frame}.{wt.column}, {wt.rows} rows) ' + f'| {wm.peak_bytes / 1e6:.1f}MB ({wm.frame}.{wm.column}, {wm.rows} rows) |') + return '\n'.join(lines) diff --git a/scripts/perf_smoke_report.py b/scripts/perf_smoke_report.py new file mode 100644 index 000000000..0cf55c7cd --- /dev/null +++ b/scripts/perf_smoke_report.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +"""Run the perf/memory smoke harness over the shipped stat lists and report. + +Used by the `Stats / Perf Smoke Report` CI job: prints a per-stat worst-case +time / peak-memory markdown table, appends it to the job summary when +$GITHUB_STEP_SUMMARY is set, and exits 1 if any stat blows the loose smoke +limits. The table is for spotting drift between runs; only order-of-magnitude +blowups fail the job. See issue #920. + +Usage: + uv run python scripts/perf_smoke_report.py +""" +import os +import sys + +from buckaroo.customizations.pd_stats_v2 import PD_ANALYSIS_V2 +from buckaroo.pluggable_analysis_framework.perf_smoke import ( + measurements_markdown, run_perf_smoke) + + +def main() -> int: + result = run_perf_smoke(PD_ANALYSIS_V2) + sections = ['## Summary stat perf smoke (PD_ANALYSIS_V2)', '', + measurements_markdown(result.measurements), ''] + if result.findings: + sections += ['### Findings', ''] + sections += [f'- {finding.message}' for finding in result.findings] + sections += [''] + report = '\n'.join(sections) + print(report) + summary_path = os.environ.get('GITHUB_STEP_SUMMARY') + if summary_path: + with open(summary_path, 'a') as fh: + fh.write(report + '\n') + return 1 if not result.passed else 0 + + +if __name__ == '__main__': + sys.exit(main()) From 75e45d982f1b66633a1c4c80aa36ebd094eac36a Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 11 Jun 2026 17:56:36 -0400 Subject: [PATCH 3/8] feat(ci): link the perf smoke report from the TestPyPI PR comment (#920) --- .github/workflows/checks.yml | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index df6db2f63..802de529c 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -116,6 +116,7 @@ jobs: permissions: id-token: write pull-requests: write + actions: read steps: - name: Download build artifacts uses: actions/download-artifact@v8 @@ -139,6 +140,22 @@ jobs: const pr = context.issue.number; const rtdSlug = 'buckaroo-data'; const docsUrl = `https://${rtdSlug}--${pr}.org.readthedocs.build/en/${pr}/`; + // Deep-link to the Stats / Perf Smoke Report job summary in this + // run (the per-stat time/memory table). Falls back to the run page. + const runUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`; + let perfSmokeUrl = runUrl; + try { + const { data: { jobs } } = await github.rest.actions.listJobsForWorkflowRun({ + owner: context.repo.owner, + repo: context.repo.repo, + run_id: context.runId, + per_page: 100, + }); + const smoke = jobs.find(j => j.name === 'Stats / Perf Smoke Report'); + if (smoke) perfSmokeUrl = `${runUrl}#summary-${smoke.id}`; + } catch (e) { + core.warning(`perf smoke job lookup failed: ${e.message}`); + } const body = [ '## :package: TestPyPI package published', '', @@ -161,6 +178,8 @@ jobs: `### :book: [Docs preview](${docsUrl})`, '', `### :art: [Storybook preview](${docsUrl}storybook/)`, + '', + `### :stopwatch: [Perf smoke report](${perfSmokeUrl})`, ].join('\n'); const { data: comments } = await github.rest.issues.listComments({ From 3f3d3671bf2d40125a4eeaff10e588d85eed930b Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Fri, 12 Jun 2026 05:52:22 -0400 Subject: [PATCH 4/8] feat(stats): native-memory tracking + polars/xorq perf smoke runners (#920) --- .../perf_smoke.py | 217 +++++++++++++++--- .../perf_smoke_pl.py | 28 +++ .../perf_smoke_xorq.py | 127 ++++++++++ scripts/perf_smoke_report.py | 53 ++++- tests/unit/perf_smoke_test.py | 55 +++++ 5 files changed, 436 insertions(+), 44 deletions(-) create mode 100644 buckaroo/pluggable_analysis_framework/perf_smoke_pl.py create mode 100644 buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py diff --git a/buckaroo/pluggable_analysis_framework/perf_smoke.py b/buckaroo/pluggable_analysis_framework/perf_smoke.py index d6725cb6f..bedc413ca 100644 --- a/buckaroo/pluggable_analysis_framework/perf_smoke.py +++ b/buckaroo/pluggable_analysis_framework/perf_smoke.py @@ -39,13 +39,21 @@ never run again — a quadratic stat can't make the smoke test itself run away. - Errors raised by stats are ignored here; that's ``unit_test()``'s job. -- Peak memory is measured with ``tracemalloc``, which sees numpy/pandas - allocations. Polars/Arrow-native allocations are invisible to it; this - harness targets the pandas pipeline. +- Peak memory is measured two ways per call: ``tracemalloc`` (exact, but only + sees python-heap allocations like numpy/pandas) and a native peak-RSS + tracker (``_NativePeak``) that catches what tracemalloc can't — polars, + Arrow and DataFusion allocate in native memory. A memory finding fires when + either signal exceeds the limit. +- The harness is engine-agnostic where it can be: ``run_perf_smoke`` accepts + polars frames (``perf_smoke_pl.PL_SMOKE_FRAME_MAKERS``) with a polars stat + list, since polars stats run through the same ``StatPipeline``. The xorq + pipeline executes differently — see ``perf_smoke_xorq``. """ from __future__ import annotations import dataclasses +import sys +import threading import time import tracemalloc from dataclasses import dataclass @@ -55,6 +63,16 @@ import numpy as np import pandas as pd +try: + import psutil +except ImportError: # pragma: no cover - optional, used only as a fallback tracker + psutil = None + +try: + import resource +except ImportError: # pragma: no cover - windows + resource = None + from .stat_pipeline import StatPipeline, _normalize_inputs from .stat_func import StatFunc @@ -136,13 +154,16 @@ class SmokeFinding: @dataclass class SmokeMeasurement: """One instrumented stat call: which stat, on which frame/column/scale, - and what it cost. Collected for every call, not just limit exceedances.""" + and what it cost. Collected for every call, not just limit exceedances. + ``peak_bytes`` is tracemalloc's python-heap peak; ``native_peak_bytes`` + is the process peak-RSS delta (catches polars/Arrow allocations).""" stat_name: str frame: str column: str rows: int seconds: float peak_bytes: float + native_peak_bytes: float = 0.0 @dataclass @@ -157,15 +178,124 @@ class _SmokeSkip(Exception): records an Err instead of running the pathological code again.""" -def _baseline_seconds(df: pd.DataFrame) -> float: - """Time canonical per-column pandas work on a smoke frame. A reasonable - stat does far less than this; the time limit is a multiple of it. Ops that - raise on a shape (value_counts on unhashable cells) are skipped — the - absolute floor keeps the limit sane when most ops skip.""" +def _read_status_bytes(field: str) -> float: + """Read a kB-valued field (VmRSS, VmHWM) from /proc/self/status.""" + with open('/proc/self/status') as fh: + for line in fh: + if line.startswith(field + ':'): + return float(line.split()[1]) * 1024 + return 0.0 + + +def _maxrss_bytes() -> float: + # ru_maxrss is kB on linux, bytes on macOS + val = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + return float(val) * (1024 if sys.platform.startswith('linux') else 1) + + +def _choose_native_strategy() -> str: + """Pick the best available native peak-memory tracker once at import. + + linux: the kernel's peak-RSS counter (VmHWM), reset per call by writing + '5' to /proc/self/clear_refs — exact and dependency-free. + psutil: a ~1ms sampling thread reading current RSS — approximate, but + blowups big enough to matter (>20MB) live long enough to be sampled. + maxrss: ru_maxrss high-water only — registers a peak only when it exceeds + the process's previous maximum. + """ + if sys.platform.startswith('linux'): + try: + with open('/proc/self/clear_refs', 'w') as fh: + fh.write('5') + if _read_status_bytes('VmHWM') > 0: + return 'linux' + except Exception: + pass + if psutil is not None: + return 'psutil' + if resource is not None: + return 'maxrss' + return 'none' + + +_NATIVE_STRATEGY = _choose_native_strategy() +_PSUTIL_PROC = psutil.Process() if psutil is not None else None + + +class _NativePeak: + """Peak process-RSS delta across one call — the native-memory complement + to tracemalloc. Catches polars/Arrow/DataFusion allocations, which never + touch the python heap. RSS is confounded by allocator pooling (memory + reused from a freed pool shows no delta), so this under-reports repeats; + with flag-and-skip the first blowup is the one that matters.""" + + def start(self) -> None: + self.before = 0.0 + if _NATIVE_STRATEGY == 'linux': + with open('/proc/self/clear_refs', 'w') as fh: + fh.write('5') + self.before = _read_status_bytes('VmRSS') + elif _NATIVE_STRATEGY == 'psutil': + self.before = float(_PSUTIL_PROC.memory_info().rss) + self.peak = self.before + self._stop_evt = threading.Event() + self._thread = threading.Thread(target=self._sample, daemon=True) + self._thread.start() + elif _NATIVE_STRATEGY == 'maxrss': + self.before = _maxrss_bytes() + + def _sample(self) -> None: + while not self._stop_evt.wait(0.001): + rss = float(_PSUTIL_PROC.memory_info().rss) + if rss > self.peak: + self.peak = rss + + def stop(self) -> float: + if _NATIVE_STRATEGY == 'linux': + return max(0.0, _read_status_bytes('VmHWM') - self.before) + if _NATIVE_STRATEGY == 'psutil': + self._stop_evt.set() + self._thread.join() + rss = float(_PSUTIL_PROC.memory_info().rss) + if rss > self.peak: + self.peak = rss + return max(0.0, self.peak - self.before) + if _NATIVE_STRATEGY == 'maxrss': + return max(0.0, _maxrss_bytes() - self.before) + return 0.0 + + +def _frame_bytes(df) -> float: + """Deep size of a pandas or polars frame.""" + if isinstance(df, pd.DataFrame): + return float(df.memory_usage(deep=True).sum()) + return float(df.estimated_size()) + + +# (pandas name, polars name) pairs for the canonical baseline ops +_CANONICAL_OP_NAMES = (('value_counts', 'value_counts'), ('sort_values', 'sort'), + ('nunique', 'n_unique')) + + +def _canonical_ops(ser) -> list: + ops = [] + for pd_name, pl_name in _CANONICAL_OP_NAMES: + op = getattr(ser, pd_name, None) + if op is None: + op = getattr(ser, pl_name, None) + if op is not None: + ops.append(op) + return ops + + +def _baseline_seconds(df) -> float: + """Time canonical per-column work on a smoke frame (pandas or polars). A + reasonable stat does far less than this; the time limit is a multiple of + it. Ops that raise on a shape (value_counts on unhashable cells) are + skipped — the absolute floor keeps the limit sane when most ops skip.""" def work(): for col in df.columns: - ser = df[col] - for op in (ser.value_counts, ser.sort_values, ser.nunique): + for op in _canonical_ops(df[col]): try: op() except Exception: @@ -197,10 +327,12 @@ def configure(self, rows: int, time_limit: float, memory_limit: float) -> None: self.time_limit = time_limit self.memory_limit = memory_limit - def record(self, stat_name: str, seconds: float, peak_bytes: float) -> None: + def record(self, stat_name: str, seconds: float, peak_bytes: float, + native_peak_bytes: float = 0.0) -> None: self.measurements.append(SmokeMeasurement( stat_name=stat_name, frame=self.frame, column=self.column, - rows=self.rows, seconds=seconds, peak_bytes=peak_bytes)) + rows=self.rows, seconds=seconds, peak_bytes=peak_bytes, + native_peak_bytes=native_peak_bytes)) if seconds > self.time_limit: self.flagged.add(stat_name) self.findings.append(SmokeFinding( @@ -211,34 +343,46 @@ def record(self, stat_name: str, seconds: float, peak_bytes: float) -> None: f"of the {self.rows}-row '{self.frame}' frame (limit {self.time_limit:.2f}s). " f"Likely a per-row python loop or an O(n^2) algorithm — " f"vectorize with pandas/numpy operations."))) - if peak_bytes > self.memory_limit: + worst_mem = max(peak_bytes, native_peak_bytes) + if worst_mem > self.memory_limit: + signal = 'python-heap' if peak_bytes >= native_peak_bytes else 'native' self.flagged.add(stat_name) self.findings.append(SmokeFinding( stat_name=stat_name, frame=self.frame, column=self.column, - kind='memory', measured=peak_bytes, limit=self.memory_limit, rows=self.rows, + kind='memory', measured=worst_mem, limit=self.memory_limit, rows=self.rows, message=( - f"stat '{stat_name}' allocated a peak of {peak_bytes / 1e6:.0f}MB on column " - f"'{self.column}' of the {self.rows}-row '{self.frame}' frame " - f"(limit {self.memory_limit / 1e6:.0f}MB). " + f"stat '{stat_name}' allocated a peak of {worst_mem / 1e6:.0f}MB " + f"({signal}) on column '{self.column}' of the {self.rows}-row " + f"'{self.frame}' frame (limit {self.memory_limit / 1e6:.0f}MB). " f"Avoid materializing large intermediates (pairwise matrices, " f"cross joins, full copies per row)."))) +def _measure(recorder: _Recorder, stat_name: str, fn, *args, **kwargs): + """Run ``fn`` under wall-time + tracemalloc + native-RSS instrumentation + and record the measurement. Exceptions propagate after recording; the + flagged check is the caller's job.""" + before_bytes = tracemalloc.get_traced_memory()[0] + tracemalloc.reset_peak() + native = _NativePeak() + native.start() + t0 = time.perf_counter() + try: + return fn(*args, **kwargs) + finally: + elapsed = time.perf_counter() - t0 + native_peak = native.stop() + peak = tracemalloc.get_traced_memory()[1] - before_bytes + recorder.record(stat_name, elapsed, peak, native_peak) + + def _wrap(sf: StatFunc, recorder: _Recorder) -> StatFunc: inner = sf.func def measured(*args, **kwargs): if sf.name in recorder.flagged: raise _SmokeSkip(f"'{sf.name}' skipped after an earlier perf finding") - before_bytes = tracemalloc.get_traced_memory()[0] - tracemalloc.reset_peak() - t0 = time.perf_counter() - try: - return inner(*args, **kwargs) - finally: - elapsed = time.perf_counter() - t0 - peak = tracemalloc.get_traced_memory()[1] - before_bytes - recorder.record(sf.name, elapsed, peak) + return _measure(recorder, sf.name, inner, *args, **kwargs) return dataclasses.replace(sf, func=measured) @@ -280,7 +424,7 @@ def run_perf_smoke(stat_funcs: list, rows: int = DEFAULT_ROWS, rows=n, time_limit=max(time_floor_seconds, max_time_ratio * baseline), memory_limit=max(memory_floor_bytes, - max_memory_ratio * float(df.memory_usage(deep=True).sum()))) + max_memory_ratio * _frame_bytes(df))) for col in df.columns: recorder.column = col pipeline.process_df(df[[col]]) @@ -306,19 +450,26 @@ def perf_smoke_test(stat_funcs: list, rows: int = DEFAULT_ROWS, def measurements_markdown(measurements: List[SmokeMeasurement]) -> str: - """Per-stat markdown table: worst wall-time and worst peak memory across - all frames/columns/scales, naming where each worst case happened.""" + """Per-stat markdown table: worst wall-time, worst python-heap peak and + worst native (RSS) peak across all frames/columns/scales, naming where + each worst case happened.""" worst_time: Dict[str, SmokeMeasurement] = {} worst_mem: Dict[str, SmokeMeasurement] = {} + worst_native: Dict[str, SmokeMeasurement] = {} for m in measurements: if m.stat_name not in worst_time or m.seconds > worst_time[m.stat_name].seconds: worst_time[m.stat_name] = m if m.stat_name not in worst_mem or m.peak_bytes > worst_mem[m.stat_name].peak_bytes: worst_mem[m.stat_name] = m - lines = ['| stat | worst time | worst peak memory |', '| --- | --- | --- |'] + if (m.stat_name not in worst_native + or m.native_peak_bytes > worst_native[m.stat_name].native_peak_bytes): + worst_native[m.stat_name] = m + lines = ['| stat | worst time | worst python-heap peak | worst native peak |', + '| --- | --- | --- | --- |'] for name in sorted(worst_time): - wt, wm = worst_time[name], worst_mem[name] + wt, wm, wn = worst_time[name], worst_mem[name], worst_native[name] lines.append( f'| {name} | {wt.seconds * 1000:.1f}ms ({wt.frame}.{wt.column}, {wt.rows} rows) ' - f'| {wm.peak_bytes / 1e6:.1f}MB ({wm.frame}.{wm.column}, {wm.rows} rows) |') + f'| {wm.peak_bytes / 1e6:.1f}MB ({wm.frame}.{wm.column}, {wm.rows} rows) ' + f'| {wn.native_peak_bytes / 1e6:.1f}MB ({wn.frame}.{wn.column}, {wn.rows} rows) |') return '\n'.join(lines) diff --git a/buckaroo/pluggable_analysis_framework/perf_smoke_pl.py b/buckaroo/pluggable_analysis_framework/perf_smoke_pl.py new file mode 100644 index 000000000..19c3087c1 --- /dev/null +++ b/buckaroo/pluggable_analysis_framework/perf_smoke_pl.py @@ -0,0 +1,28 @@ +"""Polars frame suite for the perf/memory smoke harness. + +``run_perf_smoke`` is engine-agnostic: pass ``PL_SMOKE_FRAME_MAKERS`` as its +``frames`` with a polars stat list (``PL_ANALYSIS_V2``) and the same wrap, +scales and limits apply — polars stats run through the same ``StatPipeline``. +Peak-memory findings on polars rely on the native RSS tracker in +``perf_smoke``; tracemalloc cannot see polars' Rust-side allocations. + +Frames mirror ``SMOKE_FRAME_MAKERS`` via ``pl.from_pandas`` so both engines +see the same data. The shapes translate to native polars dtypes: unhashable +cells become ``List``/``Struct`` columns, Decimal becomes ``pl.Decimal``, +categorical becomes ``pl.Categorical``. +""" +from typing import Callable, Dict + +import polars as pl + +from .perf_smoke import DEFAULT_ROWS, SMOKE_FRAME_MAKERS + + +def _from_pandas_maker(pd_maker: Callable) -> Callable: + def make(rows: int = DEFAULT_ROWS, seed: int = 42) -> pl.DataFrame: + return pl.from_pandas(pd_maker(rows, seed)) + return make + + +PL_SMOKE_FRAME_MAKERS: Dict[str, Callable[..., pl.DataFrame]] = { + name: _from_pandas_maker(maker) for name, maker in SMOKE_FRAME_MAKERS.items()} diff --git a/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py b/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py new file mode 100644 index 000000000..33f4b9a3f --- /dev/null +++ b/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py @@ -0,0 +1,127 @@ +"""Xorq runner for the perf/memory smoke harness. + +The xorq pipeline doesn't call one python function per stat, so the pandas +harness's wrap-the-func approach can't time it alone. Batched aggregate stats +(``XorqColumn -> ibis.Expr``) only *build* expressions; ``XorqStatPipeline`` +folds them into a single ``table.aggregate`` and the engine executes them +together. ``run_xorq_perf_smoke`` therefore measures in two passes per frame: + +1. each batch stat's expression executed as its own single-expression + aggregate — per-stat engine time and memory attribution, and +2. the full ``XorqStatPipeline`` per column with wrapped funcs, which times + post-batch stats — including expression stats like ``histogram`` that run + real queries through their injected ``execute``. + +Frames are the pandas ``SMOKE_FRAME_MAKERS`` wrapped as ``xo.memtable``. +Engine allocations (DataFusion/Arrow) live in native memory; the RSS tracker +in ``perf_smoke`` is the signal that sees them, tracemalloc cannot. +""" +import time +import tracemalloc +from typing import Callable, Dict, Optional + +import xorq.api as xo + +from .perf_smoke import (DEFAULT_ROWS, MEMORY_FLOOR_BYTES, SMOKE_FRAME_MAKERS, SmokeResult, + _frame_bytes, _measure, _Recorder, _wrap) +from .stat_pipeline import _normalize_inputs +from .xorq_stat_pipeline import XorqStatPipeline, XorqColumn, _is_batch_func + + +def _xorq_baseline_seconds(table) -> float: + """Time canonical engine work on the memtable: a row count plus min/max + aggregates per column. Plays the role ``_baseline_seconds`` plays for + pandas — the time limit is a multiple of this on the same machine. + Columns whose dtype can't build min/max are skipped; the absolute floor + keeps the limit sane when most skip.""" + def work(): + try: + table.count().execute() + except Exception: + pass + for col in table.columns: + try: + table.aggregate([table[col].max().name('mx'), + table[col].min().name('mn')]).execute() + except Exception: + pass + + work() # warm caches so the measured pass isn't paying first-call cost + t0 = time.perf_counter() + work() + return time.perf_counter() - t0 + + +def run_xorq_perf_smoke(stat_funcs: list, rows: int = DEFAULT_ROWS, + frames: Optional[Dict[str, Callable]] = None, + max_time_ratio: float = 20.0, time_floor_seconds: float = 0.25, + max_memory_ratio: float = 20.0, + memory_floor_bytes: float = MEMORY_FLOOR_BYTES) -> SmokeResult: + """Smoke-test a xorq stat list for perf/memory pathologies. + + ``frames`` maps frame name to a ``maker(rows) -> pandas.DataFrame`` + callable (the frame is wrapped as a memtable); defaults to + SMOKE_FRAME_MAKERS. Limits and return shape match ``run_perf_smoke``. + Engine errors are tolerated — that's ``unit_test()``'s job. + """ + frame_makers = SMOKE_FRAME_MAKERS if frames is None else frames + recorder = _Recorder() + normalized = _normalize_inputs(stat_funcs) + batch_funcs = [sf for sf in normalized if _is_batch_func(sf)] + wrapped = [_wrap(sf, recorder) for sf in normalized] + pipeline = XorqStatPipeline(wrapped, unit_test=False) + + was_tracing = tracemalloc.is_tracing() + if not was_tracing: + tracemalloc.start() + try: + small = max(500, rows // 10) + scales = [small, rows] if small < rows else [rows] + for n in scales: + for frame_name, maker in frame_makers.items(): + pdf = maker(n) + try: + table = xo.memtable(pdf) + except Exception: + continue # frame shape not representable in arrow + recorder.frame = frame_name + baseline = _xorq_baseline_seconds(table) + recorder.configure( + rows=n, + time_limit=max(time_floor_seconds, max_time_ratio * baseline), + memory_limit=max(memory_floor_bytes, + max_memory_ratio * _frame_bytes(pdf))) + schema = table.schema() + # Pass 1: batch aggregates, one engine round-trip per stat + for sf in batch_funcs: + param = next(r.name for r in sf.requires if r.type is XorqColumn) + for col in table.columns: + if sf.name in recorder.flagged: + continue + if sf.column_filter is not None and not sf.column_filter(schema[col]): + continue + recorder.column = col + try: + expr = sf.func(**{param: table[col]}) + except Exception: + continue + if expr is None: + continue + query = table.aggregate([expr.name('v')]) + try: + _measure(recorder, sf.name, query.execute) + except Exception: + pass + # Pass 2: full pipeline per column — post-batch stats + for col in table.columns: + recorder.column = col + try: + pipeline.process_table(xo.memtable(pdf[[col]])) + except Exception: + pass + finally: + if not was_tracing: + tracemalloc.stop() + + return SmokeResult(passed=(not recorder.findings), findings=recorder.findings, + measurements=recorder.measurements) diff --git a/scripts/perf_smoke_report.py b/scripts/perf_smoke_report.py index 0cf55c7cd..b4e8e8ccf 100644 --- a/scripts/perf_smoke_report.py +++ b/scripts/perf_smoke_report.py @@ -2,10 +2,11 @@ """Run the perf/memory smoke harness over the shipped stat lists and report. Used by the `Stats / Perf Smoke Report` CI job: prints a per-stat worst-case -time / peak-memory markdown table, appends it to the job summary when +time / memory markdown table per engine (pandas, polars, xorq — the latter +two when installed), appends them to the job summary when $GITHUB_STEP_SUMMARY is set, and exits 1 if any stat blows the loose smoke -limits. The table is for spotting drift between runs; only order-of-magnitude -blowups fail the job. See issue #920. +limits. The tables are for spotting drift between runs; only +order-of-magnitude blowups fail the job. See issue #920. Usage: uv run python scripts/perf_smoke_report.py @@ -17,22 +18,52 @@ from buckaroo.pluggable_analysis_framework.perf_smoke import ( measurements_markdown, run_perf_smoke) +try: + from buckaroo.customizations.pl_stats_v2 import PL_ANALYSIS_V2 + from buckaroo.pluggable_analysis_framework.perf_smoke_pl import PL_SMOKE_FRAME_MAKERS +except ImportError: + PL_ANALYSIS_V2 = None + +try: + from buckaroo.customizations.xorq_stats_v2 import XORQ_STATS_V2 + from buckaroo.pluggable_analysis_framework.perf_smoke_xorq import run_xorq_perf_smoke +except ImportError: + XORQ_STATS_V2 = None + + +def _suites(): + suites = [('PD_ANALYSIS_V2 (pandas)', lambda: run_perf_smoke(PD_ANALYSIS_V2))] + if PL_ANALYSIS_V2 is not None: + suites.append(('PL_ANALYSIS_V2 (polars)', + lambda: run_perf_smoke(PL_ANALYSIS_V2, frames=PL_SMOKE_FRAME_MAKERS))) + else: + print('polars not installed — skipping polars suite', file=sys.stderr) + if XORQ_STATS_V2 is not None: + suites.append(('XORQ_STATS_V2 (xorq)', lambda: run_xorq_perf_smoke(XORQ_STATS_V2))) + else: + print('xorq not installed — skipping xorq suite', file=sys.stderr) + return suites + def main() -> int: - result = run_perf_smoke(PD_ANALYSIS_V2) - sections = ['## Summary stat perf smoke (PD_ANALYSIS_V2)', '', - measurements_markdown(result.measurements), ''] - if result.findings: - sections += ['### Findings', ''] - sections += [f'- {finding.message}' for finding in result.findings] - sections += [''] + all_passed = True + sections = [] + for title, runner in _suites(): + result = runner() + all_passed = all_passed and result.passed + sections += [f'## Summary stat perf smoke — {title}', '', + measurements_markdown(result.measurements), ''] + if result.findings: + sections += ['### Findings', ''] + sections += [f'- {finding.message}' for finding in result.findings] + sections += [''] report = '\n'.join(sections) print(report) summary_path = os.environ.get('GITHUB_STEP_SUMMARY') if summary_path: with open(summary_path, 'a') as fh: fh.write(report + '\n') - return 1 if not result.passed else 0 + return 0 if all_passed else 1 if __name__ == '__main__': diff --git a/tests/unit/perf_smoke_test.py b/tests/unit/perf_smoke_test.py index a70066f22..588c77ab3 100644 --- a/tests/unit/perf_smoke_test.py +++ b/tests/unit/perf_smoke_test.py @@ -10,8 +10,12 @@ import numpy as np import pandas as pd +from buckaroo.customizations.pl_stats_v2 import PL_ANALYSIS_V2 +from buckaroo.customizations.xorq_stats_v2 import XORQ_STATS_V2 from buckaroo.pluggable_analysis_framework.perf_smoke import ( SMOKE_FRAME_MAKERS, measurements_markdown, perf_smoke_test, run_perf_smoke) +from buckaroo.pluggable_analysis_framework.perf_smoke_pl import PL_SMOKE_FRAME_MAKERS +from buckaroo.pluggable_analysis_framework.perf_smoke_xorq import run_xorq_perf_smoke from buckaroo.pluggable_analysis_framework.stat_func import stat, RawSeries @@ -104,3 +108,54 @@ def test_run_perf_smoke_measurements_and_markdown(): md = measurements_markdown(result.measurements) assert 'smoke_length' in md assert md.startswith('|') + assert 'native' in md + + +def test_polars_native_memory_hog_flagged(): + # ~32MB allocated by polars' Rust-side allocator — invisible to + # tracemalloc, which is why the harness also tracks peak RSS. + @stat() + def pl_native_hog(ser: RawSeries) -> int: + big = ser.sample(len(ser) * 4_000, with_replacement=True, seed=1) + return len(big) + + passed, findings = perf_smoke_test([pl_native_hog], rows=2_000, + frames={'base': PL_SMOKE_FRAME_MAKERS['base']}) + assert passed is False + mem_findings = [f for f in findings if f.kind == 'memory' and f.stat_name == 'pl_native_hog'] + assert mem_findings + assert 'native' in mem_findings[0].message + + +def test_polars_stats_over_polars_frames(): + result = run_perf_smoke(PL_ANALYSIS_V2, rows=2_000, frames=PL_SMOKE_FRAME_MAKERS) + assert result.passed is True, [f.message for f in result.findings] + assert {m.frame for m in result.measurements} == set(PL_SMOKE_FRAME_MAKERS) + + +def test_xorq_batch_stats_measured(): + # Batch aggregate stats never execute inside their own python func; the + # xorq runner times each one as its own single-expression aggregate. + result = run_xorq_perf_smoke(XORQ_STATS_V2, rows=2_000, + frames={'base': SMOKE_FRAME_MAKERS['base']}) + measured = {m.stat_name for m in result.measurements} + assert {'min', 'mean'} <= measured # batch aggregates, individually executed + assert 'histogram' in measured # post-batch expression stat via the pipeline + + +def test_xorq_slow_stat_flagged(): + # Deterministic CPU burn standing in for a pathological post-batch stat. + @stat() + def xorq_slow_post(length: int) -> int: + total = 0 + for _ in range(length): + for _ in range(5_000): + total += 1 + return total + + result = run_xorq_perf_smoke(list(XORQ_STATS_V2) + [xorq_slow_post], rows=2_000, + frames={'base': SMOKE_FRAME_MAKERS['base']}) + assert result.passed is False + time_findings = [f for f in result.findings + if f.kind == 'time' and f.stat_name == 'xorq_slow_post'] + assert time_findings From 133f8ec19e1d9bd2082c5e2fbc48e626893b88b5 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Fri, 12 Jun 2026 06:01:28 -0400 Subject: [PATCH 5/8] fix(tests): move xorq perf smoke tests behind the importorskip guard (3.14 has no xorq) --- tests/unit/perf_smoke_test.py | 30 -------------------------- tests/unit/test_xorq_stats_v2.py | 37 ++++++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 30 deletions(-) diff --git a/tests/unit/perf_smoke_test.py b/tests/unit/perf_smoke_test.py index 588c77ab3..05186e082 100644 --- a/tests/unit/perf_smoke_test.py +++ b/tests/unit/perf_smoke_test.py @@ -11,11 +11,9 @@ import pandas as pd from buckaroo.customizations.pl_stats_v2 import PL_ANALYSIS_V2 -from buckaroo.customizations.xorq_stats_v2 import XORQ_STATS_V2 from buckaroo.pluggable_analysis_framework.perf_smoke import ( SMOKE_FRAME_MAKERS, measurements_markdown, perf_smoke_test, run_perf_smoke) from buckaroo.pluggable_analysis_framework.perf_smoke_pl import PL_SMOKE_FRAME_MAKERS -from buckaroo.pluggable_analysis_framework.perf_smoke_xorq import run_xorq_perf_smoke from buckaroo.pluggable_analysis_framework.stat_func import stat, RawSeries @@ -131,31 +129,3 @@ def test_polars_stats_over_polars_frames(): result = run_perf_smoke(PL_ANALYSIS_V2, rows=2_000, frames=PL_SMOKE_FRAME_MAKERS) assert result.passed is True, [f.message for f in result.findings] assert {m.frame for m in result.measurements} == set(PL_SMOKE_FRAME_MAKERS) - - -def test_xorq_batch_stats_measured(): - # Batch aggregate stats never execute inside their own python func; the - # xorq runner times each one as its own single-expression aggregate. - result = run_xorq_perf_smoke(XORQ_STATS_V2, rows=2_000, - frames={'base': SMOKE_FRAME_MAKERS['base']}) - measured = {m.stat_name for m in result.measurements} - assert {'min', 'mean'} <= measured # batch aggregates, individually executed - assert 'histogram' in measured # post-batch expression stat via the pipeline - - -def test_xorq_slow_stat_flagged(): - # Deterministic CPU burn standing in for a pathological post-batch stat. - @stat() - def xorq_slow_post(length: int) -> int: - total = 0 - for _ in range(length): - for _ in range(5_000): - total += 1 - return total - - result = run_xorq_perf_smoke(list(XORQ_STATS_V2) + [xorq_slow_post], rows=2_000, - frames={'base': SMOKE_FRAME_MAKERS['base']}) - assert result.passed is False - time_findings = [f for f in result.findings - if f.kind == 'time' and f.stat_name == 'xorq_slow_post'] - assert time_findings diff --git a/tests/unit/test_xorq_stats_v2.py b/tests/unit/test_xorq_stats_v2.py index c09bd5389..7fee1cd85 100644 --- a/tests/unit/test_xorq_stats_v2.py +++ b/tests/unit/test_xorq_stats_v2.py @@ -20,6 +20,10 @@ from buckaroo.pluggable_analysis_framework.stat_func import stat # noqa: E402 from buckaroo.customizations.xorq_stats_v2 import ( # noqa: E402 XORQ_STATS_V2) +from buckaroo.pluggable_analysis_framework.perf_smoke import ( # noqa: E402 + SMOKE_FRAME_MAKERS) +from buckaroo.pluggable_analysis_framework.perf_smoke_xorq import ( # noqa: E402 + run_xorq_perf_smoke) def _make_table(): @@ -977,3 +981,36 @@ def test_chain_source_is_materialized(self, shape, tmp_path): stats, _ = self._run(src, tmp_path) assert stats["materialized"] is True, ( f"a {shape} source carries an expensive chain — materialize once (#915)") + + +# ============================================================ +# Perf smoke (see perf_smoke_test.py for the pandas/polars half — +# these live here for the module-level xorq importorskip guard) +# ============================================================ + +def test_xorq_batch_stats_measured(): + # Batch aggregate stats never execute inside their own python func; the + # xorq runner times each one as its own single-expression aggregate. + result = run_xorq_perf_smoke(XORQ_STATS_V2, rows=2_000, + frames={"base": SMOKE_FRAME_MAKERS["base"]}) + measured = {m.stat_name for m in result.measurements} + assert {"min", "mean"} <= measured # batch aggregates, individually executed + assert "histogram" in measured # post-batch expression stat via the pipeline + + +def test_xorq_slow_stat_flagged(): + # Deterministic CPU burn standing in for a pathological post-batch stat. + @stat() + def xorq_slow_post(length: int) -> int: + total = 0 + for _ in range(length): + for _ in range(5_000): + total += 1 + return total + + result = run_xorq_perf_smoke(list(XORQ_STATS_V2) + [xorq_slow_post], rows=2_000, + frames={"base": SMOKE_FRAME_MAKERS["base"]}) + assert result.passed is False + time_findings = [f for f in result.findings + if f.kind == "time" and f.stat_name == "xorq_slow_post"] + assert time_findings From 1dfa054d5666532d189e68e8aaa1c126866b3eb6 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Fri, 12 Jun 2026 06:25:06 -0400 Subject: [PATCH 6/8] feat(stats): smoke frames for buckaroo/tallyman problem dtypes (timedelta, period, date/time/binary, uint, inf, big python ints, list) --- .../perf_smoke.py | 93 +++++++++++++++++-- .../perf_smoke_xorq.py | 5 +- tests/unit/perf_smoke_test.py | 29 +++++- 3 files changed, 112 insertions(+), 15 deletions(-) diff --git a/buckaroo/pluggable_analysis_framework/perf_smoke.py b/buckaroo/pluggable_analysis_framework/perf_smoke.py index bedc413ca..a9508c156 100644 --- a/buckaroo/pluggable_analysis_framework/perf_smoke.py +++ b/buckaroo/pluggable_analysis_framework/perf_smoke.py @@ -19,10 +19,17 @@ Pass the same stat list you would hand to a widget or pipeline — stats that ``require`` keys from other stats need their providers in the list. -The frame suite (``SMOKE_FRAME_MAKERS``) covers shapes with bug history: -unhashable object cells (#843), int64 beyond 2^53 (#800, #632), Decimal -columns (#801), tz-aware datetimes (#277), all-null columns, and categorical -dtype. Findings name the frame that triggered them. +The frame suite (``SMOKE_FRAME_MAKERS``) covers shapes with bug history in +buckaroo and shapes tallyman learned to stop emitting because the viewer +choked: unhashable object cells (#843), list (#842), int64 beyond +2^53 (#800, #632), Decimal (#801), tz-aware datetimes (#277), all-null, +categorical, timedelta/duration (#622 and tallyman's cast-to-int64 +guidance), Period/Interval (#799), date-only/time/binary (#918, tallyman's +date-not-timestamp guidance), uint8/uint64 (#791), inf/-inf extremes and +python ints past int64 (ddd_library ancestry). Findings name the frame that +triggered them. A frame whose maker or engine conversion raises is skipped — +shapes an engine rejects outright are unit-test territory, not perf +territory. ``run_perf_smoke()`` additionally returns every per-call measurement, and ``measurements_markdown()`` renders them as a per-stat worst-case table — @@ -52,6 +59,7 @@ from __future__ import annotations import dataclasses +import datetime import sys import threading import time @@ -93,9 +101,11 @@ def make_smoke_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: def _make_unhashable_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: # 843: cells holding lists/dicts break hash-based ops (nunique raises) and - # make others crawl (value_counts succeeds but takes seconds at 10k rows) + # make others crawl (value_counts succeeds but takes seconds at 10k rows). + # 842: list columns were 80-100x slower through ServerDataflow. return pd.DataFrame({'list_cells': [[i, i + 1] for i in range(rows)], - 'dict_cells': [{'k': i} for i in range(rows)]}) + 'dict_cells': [{'k': i} for i in range(rows)], + 'list_str_cells': [[f's{i}', f's{i + 1}'] for i in range(rows)]}) def _make_big_int64_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: @@ -134,9 +144,73 @@ def _make_categorical_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFra ordered=True)}) +def _make_timedelta_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # Durations were so problematic downstream (mean() raises, parquet write + # fails, polars Duration #622) that tallyman's MCP guidance tells LLMs to + # cast them to int64 before buckaroo ever sees them. The viewer still + # meets them from every other source. + rng = np.random.default_rng(seed) + return pd.DataFrame({ + 'timedelta_col': pd.to_timedelta(rng.integers(0, 10**9, rows), unit='us'), + 'mixed_magnitude_td': pd.to_timedelta( + rng.choice([1, 10**3, 10**6, 86_400 * 10**6], rows), unit='us')}) + + +def _make_period_interval_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # 799: xo.memtable rejects Period; ddd_library weird_types ancestry. + # polars/xorq conversion rejects this frame — it skips there by design + # and exercises the pandas pipeline only. + return pd.DataFrame({ + 'period_col': pd.period_range('2020-01', periods=rows, freq='M'), + 'interval_col': pd.arrays.IntervalArray.from_breaks(np.arange(rows + 1))}) + + +def _make_date_time_binary_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # date-only columns: tallyman steers schemas to 'date' (timestamp shows + # off-by-one across timezones). time + binary: 918 missed the approx + # distinct path for both. + base_date = datetime.date(2020, 1, 1) + return pd.DataFrame({ + 'date_col': [base_date + datetime.timedelta(days=int(i % 3650)) for i in range(rows)], + 'time_col': [datetime.time((i // 3600) % 24, (i // 60) % 60, i % 60) + for i in range(rows)], + 'binary_col': [b'\x00\x01' + str(i).encode() for i in range(rows)]}) + + +def _make_uint_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # 791: uint8 dictionary indices broke /load; uint64 beyond int64 range + # breaks anything that round-trips through int64. + rng = np.random.default_rng(seed) + return pd.DataFrame({ + 'uint8_col': rng.integers(0, 255, rows, dtype=np.uint8), + 'uint64_beyond_int64': np.uint64(2**63) + rng.integers(0, 1_000, rows).astype(np.uint64)}) + + +def _make_extreme_floats_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # df_with_infinity ancestry: inf/-inf/nan mixed, plus magnitudes at both + # float64 extremes (overflow bait for sums/squares). + rng = np.random.default_rng(seed) + vals = rng.random(rows) + vals[::7] = np.inf + vals[1::11] = -np.inf + vals[2::13] = np.nan + return pd.DataFrame({'inf_floats': vals, + 'tiny_huge': np.where(rng.random(rows) < 0.5, 1e-308, 1e308)}) + + +def _make_python_big_int_df(rows: int = DEFAULT_ROWS, seed: int = 42) -> pd.DataFrame: + # df_with_really_big_number ancestry: python ints past int64 stay object + # dtype and overflow any int64/float cast. + return pd.DataFrame({ + 'huge_int_obj': [9_999_999_999_999_999_999 + i for i in range(rows)]}) + + SMOKE_FRAME_MAKERS: Dict[str, Callable[..., pd.DataFrame]] = {'base': make_smoke_df, 'unhashable': _make_unhashable_df, 'big_int64': _make_big_int64_df, 'decimal': _make_decimal_df, 'tz_datetime': _make_tz_datetime_df, - 'all_null': _make_all_null_df, 'categorical': _make_categorical_df} + 'all_null': _make_all_null_df, 'categorical': _make_categorical_df, + 'timedelta': _make_timedelta_df, 'period_interval': _make_period_interval_df, + 'date_time_binary': _make_date_time_binary_df, 'uint': _make_uint_df, + 'extreme_floats': _make_extreme_floats_df, 'python_big_int': _make_python_big_int_df} @dataclass @@ -417,7 +491,10 @@ def run_perf_smoke(stat_funcs: list, rows: int = DEFAULT_ROWS, scales = [small, rows] if small < rows else [rows] for n in scales: for frame_name, maker in frame_makers.items(): - df = maker(n) + try: + df = maker(n) + except Exception: + continue # engine can't represent this shape (e.g. Period in polars) recorder.frame = frame_name baseline = _baseline_seconds(df) recorder.configure( diff --git a/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py b/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py index 33f4b9a3f..51dc12178 100644 --- a/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py +++ b/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py @@ -79,11 +79,12 @@ def run_xorq_perf_smoke(stat_funcs: list, rows: int = DEFAULT_ROWS, scales = [small, rows] if small < rows else [rows] for n in scales: for frame_name, maker in frame_makers.items(): - pdf = maker(n) try: + pdf = maker(n) table = xo.memtable(pdf) + table.count().execute() except Exception: - continue # frame shape not representable in arrow + continue # frame shape not representable in arrow (e.g. Period, #799) recorder.frame = frame_name baseline = _xorq_baseline_seconds(table) recorder.configure( diff --git a/tests/unit/perf_smoke_test.py b/tests/unit/perf_smoke_test.py index 05186e082..da2bec514 100644 --- a/tests/unit/perf_smoke_test.py +++ b/tests/unit/perf_smoke_test.py @@ -7,6 +7,8 @@ wrong. """ +import datetime + import numpy as np import pandas as pd @@ -71,19 +73,31 @@ def memory_hog(ser: RawSeries) -> float: def test_smoke_frames_cover_known_killers(): - # Shapes with bug history: unhashable cells (#843), int64 past 2^53 - # (#800/#632), Decimal (#801), tz-aware datetimes (#277), all-null, - # categorical. + # Shapes with bug history in buckaroo — unhashable cells (#843), + # list (#842), int64 past 2^53 (#800/#632), Decimal (#801), + # tz-aware datetimes (#277), Period (#799), uint8 (#791), time/binary + # (#918), all-null, categorical — plus shapes tallyman learned to stop + # emitting (timedelta/duration, timestamp-for-date) and ddd_library + # ancestry (inf, python ints past int64). expected = {'base', 'unhashable', 'big_int64', 'decimal', 'tz_datetime', - 'all_null', 'categorical'} + 'all_null', 'categorical', 'timedelta', 'period_interval', + 'date_time_binary', 'uint', 'extreme_floats', 'python_big_int'} assert expected <= set(SMOKE_FRAME_MAKERS) frames = {name: maker(100) for name, maker in SMOKE_FRAME_MAKERS.items()} for name, df in frames.items(): assert len(df) == 100, name assert isinstance(frames['unhashable'].iloc[0, 0], list) + assert isinstance(frames['unhashable']['list_str_cells'].iloc[0][0], str) assert (frames['big_int64'] > 2**53).any().any() assert frames['all_null'].isna().all().all() assert isinstance(frames['categorical'].dtypes.iloc[0], pd.CategoricalDtype) + assert frames['timedelta']['timedelta_col'].dtype.kind == 'm' + assert isinstance(frames['period_interval']['period_col'].dtype, pd.PeriodDtype) + assert isinstance(frames['date_time_binary']['date_col'].iloc[0], datetime.date) + assert isinstance(frames['date_time_binary']['binary_col'].iloc[0], bytes) + assert int(frames['uint']['uint64_beyond_int64'].iloc[0]) > 2**63 - 1 + assert np.isinf(frames['extreme_floats']['inf_floats']).any() + assert frames['python_big_int']['huge_int_obj'].iloc[0] > np.iinfo(np.int64).max def test_findings_name_the_frame(): @@ -128,4 +142,9 @@ def pl_native_hog(ser: RawSeries) -> int: def test_polars_stats_over_polars_frames(): result = run_perf_smoke(PL_ANALYSIS_V2, rows=2_000, frames=PL_SMOKE_FRAME_MAKERS) assert result.passed is True, [f.message for f in result.findings] - assert {m.frame for m in result.measurements} == set(PL_SMOKE_FRAME_MAKERS) + measured_frames = {m.frame for m in result.measurements} + # Frames whose pandas->polars conversion raises (Period/Interval) skip by + # design; everything else must be measured. + assert measured_frames <= set(PL_SMOKE_FRAME_MAKERS) + assert {'base', 'unhashable', 'big_int64', 'decimal', 'tz_datetime', + 'all_null', 'categorical', 'timedelta', 'uint'} <= measured_frames From 4d57a56a53b6a1fbce646a5e75a4b0a0c673a152 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Fri, 12 Jun 2026 06:33:19 -0400 Subject: [PATCH 7/8] chore: retrigger CI (readthedocs 502 infra flake) From 3bb2103f8a7117552020cbbaf3e3259ac2d91dc5 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sat, 13 Jun 2026 08:07:48 -0400 Subject: [PATCH 8/8] chore: retrigger RTD (stale infra-stuck build 33110420)