From a750dbddf6ff4fe71aef97de010053de34298ea1 Mon Sep 17 00:00:00 2001 From: ryandamour Date: Sun, 4 Oct 2026 05:55:18 -0700 Subject: [PATCH 1/2] Stream filtered reads of cached exports instead of loading them whole A filtered get() read the entire cached export into memory and filtered afterwards, and Arrow's allocator kept the freed memory, so a worker backfilling a large non-templated export (liquid_universe_prices) window by window ratcheted up until it was OOM-killed. - Push start/end/symbols into a batch-by-batch pyarrow.dataset scan for feather (Arrow IPC) and parquet exports; memory now scales with the rows returned. Feather V1 (FX per-pair exports) and unsupported column types fall back to the previous full-read-then-filter path. - Return freed Arrow memory to the OS after every read. Year-by-year loop over a 5.26M-row synthetic export: peak RSS 1,224 MB -> 604 MB, identical rows. Tests pin the memory bound (metered Arrow pool, with an unfiltered control), that filtered reads never load a whole export, result equivalence with the old path across formats and column types, the V1 fallback, memory release, and that results are never shared between get() calls (live seam values must not leak into later bulk reads; mutating a result must not affect the next one) -- the two failure modes of an in-memory frame cache. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 8 + src/rwpytools/formats.py | 184 +++++++++++++++------- tests/test_session.py | 112 ++++++++++++++ tests/test_streaming_reads.py | 279 ++++++++++++++++++++++++++++++++++ 4 files changed, 531 insertions(+), 52 deletions(-) create mode 100644 tests/test_streaming_reads.py diff --git a/README.md b/README.md index 2f1e086..8d3523e 100644 --- a/README.md +++ b/README.md @@ -237,6 +237,14 @@ never served as a cache hit. Eviction is LRU by fetch time against a 10 GiB budget (`RWPYTOOLS_CACHE_MAX_BYTES`). The cache is thread-safe within a process; it is not coordinated across processes. +Reading a cached export is memory-bounded: `start`, `end` and `symbols` are +pushed into a batch-by-batch scan of feather and parquet exports, so a narrow +window over a multi-GB export holds only the matching rows, never the whole +file. Iterating a large export one window at a time (e.g. a year-by-year +backfill of `liquid_universe_prices`) therefore stays flat, and freed Arrow +memory is returned to the OS after every read so long-running workers do not +ratchet up. The per-pair FX exports (Feather V1) are small and read whole. + ```python client.cache.total_bytes() client.cache.entries() diff --git a/src/rwpytools/formats.py b/src/rwpytools/formats.py index 0e50228..5bc9bc1 100644 --- a/src/rwpytools/formats.py +++ b/src/rwpytools/formats.py @@ -42,14 +42,18 @@ -------------- :func:`read_frame` narrows by date range and symbol while reading. For the -Arrow formats the predicate runs on the Arrow table *before* pandas -conversion, so ``symbols=["BTCUSDT"]`` against a 20M-row export only ever -materialises BTCUSDT's rows. :func:`max_timestamp` finds an export's -watermark by reading just its date column. +Arrow formats the predicate is pushed into a batch-by-batch dataset scan, +so ``start="2024-01-01", symbols=["BTCUSDT"]`` against a 20M-row export +never holds more than the matching rows plus one record batch — the whole +export is never materialised. :func:`max_timestamp` finds an export's +watermark by reading just its date column. After each read, Arrow's +allocator is asked to return freed memory to the OS +(:func:`release_arrow_memory`), so long-running processes do not ratchet up. """ from __future__ import annotations +import contextlib import datetime as _dt import enum import logging @@ -61,6 +65,7 @@ import pandas as pd import pyarrow as pa import pyarrow.compute as pc +import pyarrow.dataset as ds import pyarrow.feather import pyarrow.parquet @@ -344,8 +349,8 @@ def _arrow_bound(column_type: Any, day: _dt.date) -> Any: return None -def _filter_table( - table: Any, +def _filter_expression( + schema: Any, *, date_column: str | None, start: _dt.date | None, @@ -353,50 +358,109 @@ def _filter_table( symbol_column: str | None, symbols: Sequence[str] | None, ) -> tuple[Any, bool]: - """Apply what filters Arrow can evaluate natively. + """Translate the date / symbol filters into an Arrow dataset expression. - Returns ``(table, fully_filtered)``; when ``fully_filtered`` is False - the caller finishes the job in pandas (e.g. dictionary-encoded symbols - or a date column stored in an unusual type). + Returns ``(expression, complete)``. ``expression`` is ``None`` when no + filter applies to this schema. ``complete`` is False when some filter + could not be expressed (e.g. a dictionary-encoded symbol column or a + date column in an unusual type); the caller then finishes the job in + pandas, so the result is always fully filtered. """ complete = True - mask: Any = None + expression: Any = None def _and(clause: Any) -> None: - nonlocal mask - mask = clause if mask is None else pc.and_(mask, clause) + nonlocal expression + expression = clause if expression is None else (expression & clause) - if symbols and symbol_column and symbol_column in table.column_names: - column = table[symbol_column] - try: - value_set = pa.array(list(symbols)).cast(column.type) - _and(pc.is_in(column, value_set=value_set)) - except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + names = set(schema.names) + if symbols and symbol_column and symbol_column in names: + column_type = schema.field(symbol_column).type + if pa.types.is_dictionary(column_type): complete = False - - if (start is not None or end is not None) and date_column and date_column in table.column_names: - column = table[date_column] - try: - if start is not None: - low = _arrow_bound(column.type, start) - if low is None: - raise pa.ArrowNotImplementedError(str(column.type)) - _and(pc.greater_equal(column, low)) - if end is not None: - high = _arrow_bound(column.type, end + _dt.timedelta(days=1)) - if high is None: - raise pa.ArrowNotImplementedError(str(column.type)) - _and(pc.less(column, high)) - except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + else: + try: + value_set = pa.array(list(symbols)).cast(column_type) + _and(pc.field(symbol_column).isin(value_set)) + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + complete = False + + if (start is not None or end is not None) and date_column and date_column in names: + column_type = schema.field(date_column).type + low = _arrow_bound(column_type, start) if start is not None else None + high = _arrow_bound(column_type, end + _dt.timedelta(days=1)) if end is not None else None + if (start is not None and low is None) or (end is not None and high is None): complete = False + else: + if low is not None: + _and(pc.field(date_column) >= low) + if high is not None: + _and(pc.field(date_column) < high) + + return expression, complete + + +def _scan_filtered( + path: Path, + fmt: DataFormat, + *, + date_column: str | None, + start: _dt.date | None, + end: _dt.date | None, + symbol_column: str | None, + symbols: Sequence[str] | None, + memory_pool: Any = None, +) -> tuple[Any, bool] | None: + """Scan an Arrow-format export with the filter pushed into the scan. + + The file is read record batch by record batch and each batch is + filtered as it is decoded, so peak memory is the *matching* rows plus + one batch — never the whole export. That is what keeps a narrow + ``start`` / ``end`` cheap on a multi-GB file, on the first read and on + every different window after it. + + Returns ``(table, complete)``, or ``None`` when the file cannot be + scanned as a dataset (Feather V1, which the per-pair FX exports use, is + not an Arrow IPC file) and the caller must fall back to a full read. + """ - if mask is not None: - # Null comparisons yield null; treat them as "not selected". - table = table.filter(pc.fill_null(mask, False)) + try: + dataset = ds.dataset(str(path), format="ipc" if fmt is DataFormat.FEATHER else "parquet") + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, OSError): + return None + expression, complete = _filter_expression( + dataset.schema, + date_column=date_column, + start=start, + end=end, + symbol_column=symbol_column, + symbols=symbols, + ) + try: + table = dataset.to_table(filter=expression, memory_pool=memory_pool) + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError): + # An expression the scanner rejects for this file: read it all and + # let pandas filter instead. + table = dataset.to_table(memory_pool=memory_pool) + complete = False return table, complete +def release_arrow_memory() -> None: + """Return Arrow's freed-but-cached memory to the operating system. + + Arrow's allocator (jemalloc / mimalloc) keeps freed memory for reuse, so + after reading a large export a process's RSS stays near its peak even + once every table is gone — and a long-running worker that reads export + after export ratchets upward until it hits its memory limit. Called + after every read of a cached export. + """ + + with contextlib.suppress(AttributeError, NotImplementedError): + pa.default_memory_pool().release_unused() + + def _read_arrow_table(path: Path, fmt: DataFormat, columns: Sequence[str] | None = None) -> Any: cols = list(columns) if columns is not None else None if fmt is DataFormat.FEATHER: @@ -419,25 +483,38 @@ def read_frame( With no filters this is :func:`read_auto`. ``end`` is inclusive of the whole day, so an intraday column keeps every bar on ``end``. + + Feather (Arrow IPC) and parquet exports are scanned with the filter + pushed down (see :func:`_scan_filtered`), so memory scales with the + rows returned rather than with the size of the export. """ fmt = detect_format(object_name) filtering = bool(symbols) or start is not None or end is not None - if not filtering: - return read(path, fmt, max_bytes=max_bytes) - _check_memory_guard(path, max_bytes=max_bytes) - kwargs: dict[str, Any] = { - "date_column": date_column, - "start": start, - "end": end, - "symbol_column": symbol_column, - "symbols": symbols, - } - if fmt in (DataFormat.FEATHER, DataFormat.PARQUET): - table, complete = _filter_table(_read_arrow_table(path, fmt), **kwargs) - frame = _arrow_to_df(table) - return frame if complete else filter_frame(frame, **kwargs) - return filter_frame(_BY_FORMAT[fmt].reader(path), **kwargs) + try: + if not filtering: + return read(path, fmt, max_bytes=max_bytes) + _check_memory_guard(path, max_bytes=max_bytes) + kwargs: dict[str, Any] = { + "date_column": date_column, + "start": start, + "end": end, + "symbol_column": symbol_column, + "symbols": symbols, + } + if fmt in (DataFormat.FEATHER, DataFormat.PARQUET): + scanned = _scan_filtered(path, fmt, **kwargs) + if scanned is None: # Feather V1: no batch index, read it whole + table = _read_arrow_table(path, fmt) + complete = False + else: + table, complete = scanned + frame = _arrow_to_df(table) + del table + return frame if complete else filter_frame(frame, **kwargs) + return filter_frame(_BY_FORMAT[fmt].reader(path), **kwargs) + finally: + release_arrow_memory() def max_timestamp(path: Path, *, object_name: str, column: str) -> pd.Timestamp | None: @@ -455,6 +532,8 @@ def max_timestamp(path: Path, *, object_name: str, column: str) -> pd.Timestamp except (KeyError, pa.ArrowInvalid): return None series = _arrow_to_df(table)[column] + del table + release_arrow_memory() else: compression: CsvCompression = "gzip" if fmt is DataFormat.CSV_GZ else None try: @@ -535,5 +614,6 @@ def read_csv_chunks( "read_frame", "read_head", "register_dtype_hints", + "release_arrow_memory", "to_timestamps", ] diff --git a/tests/test_session.py b/tests/test_session.py index 809bb96..80755e0 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -690,3 +690,115 @@ def test_snapshot_export_is_served_from_cache_without_replanning_a_download( assert not live.called assert again.route.bulk_action is not None and again.route.bulk_action.value == "cache" assert "no date watermark" in again.route.reason + + +# ---- results are never shared between calls ------------------------------------------ +# +# Each get() must return data read fresh for that call. Any in-memory reuse of +# a parsed export (e.g. an LRU of DataFrames) has to keep these passing: the +# seam refresh writes live values into the bulk half in place, and callers +# are free to mutate the frame they are handed. + + +def _mock_provisional_prices(respx_mock: respx.MockRouter) -> None: + table = pa.table( + { + "ticker": ["HBM", "HBM"], + "date": pa.array([dt.date(2026, 9, 21), dt.date(2026, 9, 22)], pa.date32()), + "close": [27.0, 27.5], + "volume": [4_000_000.0, 3_176_000.0], + } + ) + buf = io.BytesIO() + pyarrow.feather.write_feather(table, buf) + _mock_file( + respx_mock, + namespace="equities", + dataset="statarb", + schema="prices", + bucket="equity_factors_research_pod", + obj="statarb/prices.feather", + payload=buf.getvalue(), + fmt="feather", + ) + respx_mock.get("/v1/equities/statarb/prices").respond( + json=_page([{"ticker": "HBM", "date": "2026-09-22", "close": 27.74, "volume": 5_836_000.0}]) + ) + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_a_bulk_only_read_is_never_changed_by_an_earlier_live_top_up( + respx_mock: respx.MockRouter, client: Client +) -> None: + _mock_provisional_prices(respx_mock) + topped_up = client.get("statarb_prices") + assert topped_up.seam_updates == 1 # live values were applied to that result... + + # ...but the export itself, read again for the same window, is untouched. + bulk = client.get("statarb_prices", end="2026-09-22", source="bulk").df + assert bulk["volume"].tolist() == [4_000_000.0, 3_176_000.0] + assert bulk["close"].tolist() == [27.0, 27.5] + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_mutating_a_result_never_changes_the_next_one( + respx_mock: respx.MockRouter, client: Client +) -> None: + _mock_provisional_prices(respx_mock) + first = client.get("statarb_prices", source="bulk") + first.df.loc[0, "close"] = -1.0 + first.df["extra"] = 1 + first.df.sort_values("close", ascending=False, inplace=True) + + second = client.get("statarb_prices", source="bulk") + assert second.df["close"].tolist() == [27.0, 27.5] + assert "extra" not in second.df.columns + assert second.df is not first.df + + +@respx.mock(base_url=FAKE_API, assert_all_called=False) +def test_windowed_gets_scan_the_export_instead_of_loading_it( + respx_mock: respx.MockRouter, client: Client, monkeypatch: pytest.MonkeyPatch +) -> None: + """End to end: a year-by-year backfill through Client.get() never reads a + whole non-templated export into memory (only the one-column watermark + scan touches every row).""" + + import rwpytools.formats as formats_module + + days = pd.bdate_range("2015-01-01", "2018-12-31") + table = pa.table( + { + "ticker": ["AAPL"] * len(days) + ["MSFT"] * len(days), + "date": pa.array(list(days.date) * 2, pa.date32()), + "close": list(range(len(days) * 2)), + } + ) + buf = io.BytesIO() + pyarrow.feather.write_feather(table, buf, chunksize=100) + _mock_file( + respx_mock, + namespace="equities", + dataset="equity.factors", + schema="prices", + bucket="equity_factors_research_pod", + obj="equity_factors/prices.feather", + payload=buf.getvalue(), + fmt="feather", + ) + real = formats_module._read_arrow_table + + def _only_single_columns(path: Any, fmt: Any, columns: Any = None) -> Any: + assert columns is not None and len(columns) == 1, "a whole export was loaded" + return real(path, fmt, columns) + + monkeypatch.setattr(formats_module, "_read_arrow_table", _only_single_columns) + + rows = 0 + for year in range(2015, 2019): + result = client.get( + "liquid_universe_prices", start=f"{year}-01-01", end=f"{year}-12-31", source="bulk" + ) + assert set(pd.to_datetime(result.df["date"]).dt.year) == {year} + rows += len(result) + assert rows == len(days) * 2 diff --git a/tests/test_streaming_reads.py b/tests/test_streaming_reads.py new file mode 100644 index 0000000..ac9535e --- /dev/null +++ b/tests/test_streaming_reads.py @@ -0,0 +1,279 @@ +"""Filtered reads of cached exports must not load the whole export. + +A long-running worker backfilling a large export one window at a time +(e.g. ``liquid_universe_prices``, ~5M rows, year by year) used to read the +*entire* export into memory on every ``get()`` and filter afterwards, and +Arrow's allocator kept the freed memory — RSS ratcheted up until the +worker was OOM-killed. These tests pin the fix: the filter is pushed into +a batch-by-batch scan, results match the old full-read-then-filter path +exactly, and freed Arrow memory is handed back to the OS after every read. +""" + +from __future__ import annotations + +import datetime as dt +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd +import pyarrow as pa +import pyarrow.feather +import pyarrow.parquet +import pytest + +from rwpytools import formats +from rwpytools.errors import RwPyToolsError +from rwpytools.formats import DataFormat, filter_frame, read_auto, read_frame + +N_TICKERS = 200 +N_DAYS = 2_000 # 400k rows + + +def _wide_table(*, tz: str | None = None) -> pa.Table: + """A long export: every ticker on every day, like the equity price exports.""" + + rng = np.random.default_rng(0) + days = pd.date_range("2015-01-01", periods=N_DAYS, freq="D") + tickers = np.repeat([f"T{i:03d}" for i in range(N_TICKERS)], N_DAYS) + stamps = pd.DatetimeIndex(np.tile(days.values, N_TICKERS)).as_unit("us") + if tz: + stamps = stamps.tz_localize(tz) + n = N_TICKERS * N_DAYS + return pa.table( + { + "ticker": tickers, + "date": pa.array(stamps.date, pa.date32()), + "Datetime": pa.array(stamps.to_pydatetime(), pa.timestamp("us", tz=tz)), + "iso": pa.array(np.tile(days.strftime("%Y-%m-%d").to_numpy(), N_TICKERS)), + "close": rng.random(n), + "volume": rng.random(n), + } + ) + + +@pytest.fixture(scope="module") +def export_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: + """The same data in every format the exports use, written once.""" + + root = tmp_path_factory.mktemp("exports") + table = _wide_table() + # pandas' to_feather default: Arrow IPC (Feather V2), lz4, many batches. + pyarrow.feather.write_feather(table, root / "v2.feather", chunksize=2_000) + pyarrow.feather.write_feather( + table, root / "v2_raw.feather", chunksize=2_000, compression="uncompressed" + ) + pyarrow.feather.write_feather(table.drop_columns(["Datetime"]), root / "v1.feather", version=1) + pyarrow.parquet.write_table(table, root / "x.parquet", row_group_size=20_000) + pyarrow.feather.write_feather(_wide_table(tz="UTC"), root / "tz.feather", chunksize=2_000) + dictionary = table.set_column(0, "ticker", pa.compute.dictionary_encode(table.column("ticker"))) + pyarrow.feather.write_feather(dictionary, root / "dict.feather", chunksize=2_000) + table.to_pandas().to_csv(root / "x.csv", index=False) + return root + + +# ---- the memory property -------------------------------------------------------------- + + +def test_a_narrow_window_never_loads_the_whole_export(export_dir: Path) -> None: + """The fix itself: peak Arrow memory for a one-year window over a multi- + batch export is a small fraction of the export, not all of it.""" + + path = export_dir / "v2.feather" + full_bytes = pyarrow.feather.read_table(path).nbytes + + def peak(**filters: Any) -> tuple[int, int]: + pool = pa.proxy_memory_pool(pa.system_memory_pool()) + scanned = formats._scan_filtered( + path, + DataFormat.FEATHER, + date_column=filters.get("date_column"), + start=filters.get("start"), + end=filters.get("end"), + symbol_column=None, + symbols=None, + memory_pool=pool, + ) + assert scanned is not None + table, complete = scanned + assert complete + measured = (pool.max_memory(), table.num_rows) + # The table's buffers belong to ``pool``: free them while the pool + # is still alive, or local teardown order can crash the interpreter. + del table, scanned + return measured + + # Control: an unfiltered scan really does allocate the whole export, so + # the meter is measuring what we think it is. + whole_peak, whole_rows = peak() + assert whole_rows == N_TICKERS * N_DAYS + assert whole_peak >= 0.9 * full_bytes + + year_peak, year_rows = peak( + date_column="date", start=dt.date(2016, 1, 1), end=dt.date(2016, 12, 31) + ) + assert year_rows == N_TICKERS * 366 + assert year_peak < 0.35 * full_bytes, ( + f"one-year window peaked at {year_peak:,} bytes of a {full_bytes:,}-byte export" + ) + + +def test_filtered_reads_do_not_fall_back_to_a_full_table_read( + export_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def _full_read_forbidden(*_: Any, **__: Any) -> Any: + raise AssertionError("filtered read loaded the whole export") + + monkeypatch.setattr(formats, "_read_arrow_table", _full_read_forbidden) + monkeypatch.setattr(pyarrow.feather, "read_table", _full_read_forbidden) + monkeypatch.setattr(pyarrow.parquet, "read_table", _full_read_forbidden) + + for name in ("v2.feather", "x.parquet"): + out = read_frame( + export_dir / name, + object_name=name, + date_column="date", + start=dt.date(2016, 1, 1), + end=dt.date(2016, 1, 31), + symbol_column="ticker", + symbols=["T007"], + ) + assert len(out) == 31 + assert set(out["ticker"]) == {"T007"} + + +def test_arrow_memory_is_released_after_every_read( + export_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Without this, freed Arrow memory stays in the allocator and a worker's + RSS ratchets up read after read.""" + + calls: list[str] = [] + monkeypatch.setattr(formats, "release_arrow_memory", lambda: calls.append("released")) + + read_frame( + export_dir / "v2.feather", + object_name="v2.feather", + start=dt.date(2016, 1, 1), + date_column="date", + ) + read_frame(export_dir / "v2.feather", object_name="v2.feather") # unfiltered + formats.max_timestamp(export_dir / "v2.feather", object_name="v2.feather", column="date") + assert calls == ["released"] * 3 + + calls.clear() + with pytest.raises(RwPyToolsError, match="max_bytes"): # released on the error path too + read_frame( + export_dir / "v2.feather", + object_name="v2.feather", + date_column="date", + start=dt.date(2016, 1, 1), + max_bytes=1, + ) + assert calls == ["released"] + + +def test_release_arrow_memory_is_safe_to_call() -> None: + formats.release_arrow_memory() + + +# ---- results identical to the old full-read-then-filter path --------------------------- + +CASES = [ + pytest.param( + {"date_column": "date", "start": dt.date(2016, 3, 1), "end": dt.date(2016, 3, 31)}, + id="date32-range", + ), + pytest.param({"date_column": "Datetime", "start": dt.date(2017, 1, 1)}, id="timestamp-start"), + pytest.param( + {"date_column": "Datetime", "end": dt.date(2015, 2, 1)}, id="timestamp-end-inclusive" + ), + pytest.param( + {"date_column": "iso", "start": dt.date(2016, 1, 1), "end": dt.date(2016, 1, 2)}, + id="iso-string", + ), + pytest.param({"symbol_column": "ticker", "symbols": ["T001", "T150"]}, id="symbols"), + pytest.param( + { + "date_column": "date", + "start": dt.date(2016, 6, 1), + "end": dt.date(2016, 6, 30), + "symbol_column": "ticker", + "symbols": ["T042"], + }, + id="window-and-symbol", + ), + pytest.param({"date_column": "date", "start": dt.date(2030, 1, 1)}, id="empty-result"), +] + + +def _reference(path: Path, object_name: str, **filters: Any) -> pd.DataFrame: + return filter_frame(read_auto(path, object_name=object_name), **filters) + + +@pytest.mark.parametrize("filters", CASES) +@pytest.mark.parametrize( + "name", ["v2.feather", "v2_raw.feather", "x.parquet", "dict.feather", "x.csv"] +) +def test_filtered_read_matches_full_read_then_filter( + export_dir: Path, name: str, filters: dict[str, Any] +) -> None: + if name == "x.csv" and filters.get("date_column") == "Datetime": + pytest.skip("CSV round-trips timestamps as text; covered by the iso case") + out = read_frame(export_dir / name, object_name=name, **filters) + expected = _reference(export_dir / name, name, **filters) + pd.testing.assert_frame_equal( + out.reset_index(drop=True), expected.reset_index(drop=True), check_categorical=False + ) + + +@pytest.mark.parametrize( + "filters", [c for c in CASES if c.values[0].get("date_column") != "Datetime"] +) +def test_feather_v1_falls_back_and_still_filters(export_dir: Path, filters: Any) -> None: + """Feather V1 (the per-pair FX exports) is not an IPC file, so it cannot + be scanned; the fallback must still return exactly the filtered rows.""" + + assert ( + formats._scan_filtered( + export_dir / "v1.feather", + DataFormat.FEATHER, + date_column=None, + start=None, + end=None, + symbol_column=None, + symbols=None, + ) + is None + ) + out = read_frame(export_dir / "v1.feather", object_name="v1.feather", **filters) + expected = _reference(export_dir / "v1.feather", "v1.feather", **filters) + pd.testing.assert_frame_equal(out.reset_index(drop=True), expected.reset_index(drop=True)) + + +def test_timezone_aware_timestamps_filter_by_utc_day(export_dir: Path) -> None: + filters = {"date_column": "Datetime", "start": dt.date(2016, 1, 1), "end": dt.date(2016, 1, 3)} + out = read_frame(export_dir / "tz.feather", object_name="tz.feather", **filters) + expected = _reference(export_dir / "tz.feather", "tz.feather", **filters) + assert len(out) == N_TICKERS * 3 + pd.testing.assert_frame_equal(out.reset_index(drop=True), expected.reset_index(drop=True)) + + +def test_a_year_by_year_backfill_returns_every_row_exactly_once(export_dir: Path) -> None: + """The workload that OOM'd: iterate the export one calendar year at a time.""" + + path = export_dir / "v2.feather" + years = range(2015, 2021) + parts = [ + read_frame( + path, + object_name="v2.feather", + date_column="date", + start=dt.date(y, 1, 1), + end=dt.date(y, 12, 31), + ) + for y in years + ] + combined = pd.concat(parts, ignore_index=True) + assert len(combined) == N_TICKERS * N_DAYS + assert not combined.duplicated(["ticker", "date"]).any() From fb20309d7ccfe548d96136fb58e7f681726eb0e1 Mon Sep 17 00:00:00 2001 From: ryandamour Date: Mon, 5 Oct 2026 06:45:08 -0700 Subject: [PATCH 2/2] Keep metered Arrow pools alive in the memory-bound test The scanner's background read-ahead can release buffers to the test's proxy pool after to_table() returns. When the pool was destroyed first, a later, unrelated test crashed the interpreter (about 1 run in 10). Keep the metered pools for the life of the process. The library is not affected: it always uses Arrow's process-wide default pool. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_streaming_reads.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/tests/test_streaming_reads.py b/tests/test_streaming_reads.py index ac9535e..272f566 100644 --- a/tests/test_streaming_reads.py +++ b/tests/test_streaming_reads.py @@ -26,6 +26,11 @@ from rwpytools.errors import RwPyToolsError from rwpytools.formats import DataFormat, filter_frame, read_auto, read_frame +#: Metered pools must outlive every buffer allocated from them; see the +#: memory-bound test. (The library itself always uses Arrow's process-wide +#: default pool, so it has no such lifetime hazard.) +_METERED_POOLS: list[Any] = [] + N_TICKERS = 200 N_DAYS = 2_000 # 400k rows @@ -84,6 +89,10 @@ def test_a_narrow_window_never_loads_the_whole_export(export_dir: Path) -> None: def peak(**filters: Any) -> tuple[int, int]: pool = pa.proxy_memory_pool(pa.system_memory_pool()) + # The scanner's background read-ahead can still release buffers to + # this pool after the scan returns; a pool destroyed first crashes the + # interpreter later, in some unrelated test. Keep it for the process. + _METERED_POOLS.append(pool) scanned = formats._scan_filtered( path, DataFormat.FEATHER,