diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index f4efc3fa8..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({ @@ -308,6 +327,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..a9508c156 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,22 @@ 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 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 — +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 @@ -30,21 +46,41 @@ 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 datetime +import sys +import threading 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 +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 @@ -63,9 +99,124 @@ 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). + # 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)], + '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: + # 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)}) + + +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, + '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 class SmokeFinding: stat_name: str + frame: str column: str kind: str # 'time' | 'memory' measured: float # seconds for 'time', bytes for 'memory' @@ -74,20 +225,155 @@ 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. + ``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 +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.""" +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] - ser.value_counts() - ser.sort_values() - ser.nunique() + for op in _canonical_ops(df[col]): + try: + op() + except Exception: + pass work() # warm caches so the measured pass isn't paying first-call cost t0 = time.perf_counter() @@ -96,12 +382,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') @@ -112,56 +401,76 @@ 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, + native_peak_bytes=native_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: + 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, 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=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 a {self.rows}-row frame (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) -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 +478,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 +490,63 @@ 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(): + 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( + rows=n, + time_limit=max(time_floor_seconds, max_time_ratio * baseline), + memory_limit=max(memory_floor_bytes, + max_memory_ratio * _frame_bytes(df))) + 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, 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 + 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, 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'| {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..51dc12178 --- /dev/null +++ b/buckaroo/pluggable_analysis_framework/perf_smoke_xorq.py @@ -0,0 +1,128 @@ +"""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(): + try: + pdf = maker(n) + table = xo.memtable(pdf) + table.count().execute() + except Exception: + continue # frame shape not representable in arrow (e.g. Period, #799) + 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 new file mode 100644 index 000000000..b4e8e8ccf --- /dev/null +++ b/scripts/perf_smoke_report.py @@ -0,0 +1,70 @@ +#!/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 / 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 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 +""" +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) + +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: + 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 0 if all_passed else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tests/unit/perf_smoke_test.py b/tests/unit/perf_smoke_test.py index 1248487a9..da2bec514 100644 --- a/tests/unit/perf_smoke_test.py +++ b/tests/unit/perf_smoke_test.py @@ -7,9 +7,15 @@ wrong. """ +import datetime + import numpy as np +import pandas as pd -from buckaroo.pluggable_analysis_framework.perf_smoke import perf_smoke_test +from buckaroo.customizations.pl_stats_v2 import PL_ANALYSIS_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.stat_func import stat, RawSeries @@ -64,3 +70,81 @@ 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 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', '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(): + @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('|') + 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] + 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 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