From 4cbbfcb9e739eff8773ec15579c56226213556b8 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Mon, 29 Jun 2026 12:57:37 -0700 Subject: [PATCH 1/7] draft move to segment based cache indexing --- .../_internal/low_level_wrappers/data.py | 592 +++++++++++------- .../_internal/low_level_wrappers/test_data.py | 562 ++++++++++++----- .../sift_client/_tests/test_client_cache.py | 61 +- python/lib/sift_client/util/cache.py | 46 +- 4 files changed, 845 insertions(+), 416 deletions(-) diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data.py b/python/lib/sift_client/_internal/low_level_wrappers/data.py index 13246f41f..309eec3cc 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data.py @@ -3,7 +3,7 @@ import asyncio import logging from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Tuple, cast import pandas as pd from pydantic import BaseModel, ConfigDict @@ -55,50 +55,63 @@ def _new_cache_entry( ) +TimeRange = Tuple[datetime, datetime] + + +class SegmentRef(BaseModel): + """Pointer to one cached segment, stored on the per-bucket index.""" + + model_config = ConfigDict(frozen=True) + seg_id: int + start_time: datetime + end_time: datetime + size_bytes: int + + +class SegmentIndex(BaseModel): + """Per-(channel, run) index of cached segments.""" + + model_config = ConfigDict(arbitrary_types_allowed=True) + schema_version: int = 1 + next_seg_id: int = 0 + segments: list[SegmentRef] = [] + + class ChannelDataCache: """Channel-side adapter over the shared :class:`DiskCache` store. - The store is owned by :class:`~sift_client.client.SiftClient` and - shared by every cache-aware resource; this adapter is the typed, - namespaced view of it that the channel data path uses. - - Responsibilities the adapter holds onto: - - * **Key namespacing.** Every read/write goes through :meth:`_key`, - which prefixes the channel id with ``channel:`` and folds in the - ``run_id`` it was queried under. Run id is part of the cache - dimension because the same channel can have wholly different data - under different runs (different timelines, different points). A - bare ``channel:`` key would conflate runs and serve run-A's - data to a run-B query. The ``channel:`` prefix also keeps a future - calculated-channels or exports adapter on the same store from - colliding on raw resource ids. - * **Typing.** ``put`` only accepts :class:`ChannelCacheEntry`; - ``get`` ``isinstance``-checks the raw value before handing it back, - so a corrupt or cross-adapter row reads as a miss instead of - blowing up downstream pandas code. - * **Size measurement.** The store stays value-agnostic; the adapter - already computes ``size_bytes`` on the entry via - :func:`_new_cache_entry` (``DataFrame.memory_usage(deep=True)``) so - it just forwards that to the store's oversize guard. - * **Resource-side state.** :attr:`name_id_map` lives here because - it's channel-specific bookkeeping needed to wire raw fetch - responses (keyed by channel *name*) back to the cache (keyed by - channel *id*). - - The :class:`DiskCacheAdapter` ``Protocol`` is intentionally not - declared yet — there's only one adapter shape so far. When a second - resource grows its own adapter, extract the Protocol from the two - real shapes rather than guessing from one. + Each ``(channel_id, run_id)`` bucket is split across two key shapes + in the underlying store: + + * One **index** entry (``channel:::idx``) holding a + :class:`SegmentIndex` — a tiny list of :class:`SegmentRef` ptrs + describing every segment that exists for the bucket. + * One **segment** entry per fetch (``channel:::seg:``) + holding the :class:`ChannelCacheEntry` for that fetch's slice. + + Reads stitch the relevant segments together via :meth:`get_range`, + which also reports which sub-ranges of the requested window have no + cached coverage (``gaps``) so the caller can fetch only the holes. + + Eviction tolerance: ``diskcache``'s LRU can drop a segment while + keeping the index. The reader treats a missing segment as a gap, + which forces a wire fetch for that range. The opposite (index + evicted, segments orphaned) is also fine — orphans are unreachable + and will eventually LRU-evict. Attributes: - KEY_PREFIX: Namespace prefix for keys this adapter writes to the - shared :class:`DiskCache`. Picked at class scope so adapters - in other resources can pick distinct prefixes without runtime - negotiation. + KEY_PREFIX: Namespace prefix for every key this adapter writes + to the shared :class:`DiskCache`. Picked at class scope so + adapters in other resources can pick distinct prefixes + without runtime negotiation. + SCHEMA_VERSION: Index entry schema version. Bump when the + shape changes incompatibly so older on-disk indexes are + discarded on read rather than mis-deserialized. + """ KEY_PREFIX: str = "channel:" + SCHEMA_VERSION: int = 1 def __init__(self, store: DiskCache): """Wrap ``store`` with channel-data semantics. @@ -110,73 +123,238 @@ def __init__(self, store: DiskCache): self._store = store self.name_id_map: dict[str, str] = {} - def _key(self, channel_id: str, run_id: str | None) -> str: - # Run id is part of the key because the same channel under - # different runs has different data. Empty run segment is safe - # because real run ids are UUIDs (never empty) so there's no - # collision between the run-scoped and unscoped buckets. + # --- key helpers --- + + def _bucket_prefix(self, channel_id: str, run_id: str | None) -> str: + # Stem all index/segment keys for a bucket share. Empty run + # segment is safe because real run ids are UUIDs (never empty) + # so there's no collision between the run-scoped and unscoped + # buckets. return f"{self.KEY_PREFIX}{run_id or ''}:{channel_id}" + def _index_key(self, channel_id: str, run_id: str | None) -> str: + return f"{self._bucket_prefix(channel_id, run_id)}:idx" + + def _segment_key(self, channel_id: str, run_id: str | None, seg_id: int) -> str: + return f"{self._bucket_prefix(channel_id, run_id)}:seg:{seg_id}" + @property def store(self) -> DiskCache: """The shared underlying store. Tests reach in for store-level state.""" return self._store - def has(self, channel_id: str, run_id: str | None = None) -> bool: - """True if ``(channel_id, run_id)`` is cached. + # --- public API --- + + def has_any(self, channel_id: str, run_id: str | None = None) -> bool: + """True if at least one segment is cached for this bucket. - Replaces an earlier ``__contains__`` that only took a channel id. - ``in`` was dropped on purpose: forcing the call site to name the - run keeps callers from silently sharing cache entries across runs. + Cheap check (one index read) — does NOT touch segment data. + Useful for "should I consult the cache at all?" gates. """ - return self._key(channel_id, run_id) in self._store + idx = self._load_index(channel_id, run_id) + return idx is not None and bool(idx.segments) - def get(self, channel_id: str, run_id: str | None = None) -> ChannelCacheEntry | None: - """Return the entry for ``(channel_id, run_id)`` if cached, otherwise None. + def get_range( + self, + channel_id: str, + run_id: str | None, + start_time: datetime, + end_time: datetime, + ) -> tuple[pd.DataFrame | None, list[TimeRange]]: + """Return cached data covering ``[start_time, end_time]`` plus gaps. - Type-checks the raw value before returning so a row written by a - different adapter (or a corrupt entry that survived) reads as a - miss instead of being handed back as the wrong type. - """ - raw = self._store.get(self._key(channel_id, run_id)) - if not isinstance(raw, ChannelCacheEntry): - return None - return raw + Walks the bucket's segments, slices each one to the query range, + stitches them together, and reports which sub-ranges still need + a wire fetch. - def put( + Returns: + ``(stitched_data, gaps)`` where ``stitched_data`` is the + concat of every overlapping segment sliced to the query + range (or ``None`` if no rows were found), and ``gaps`` is + the list of ``(gap_start, gap_end)`` sub-ranges within the + query window not covered by any present segment. + ``gaps == []`` means the cache fully covers the request. + """ + idx = self._load_index(channel_id, run_id) + if idx is None or not idx.segments: + return None, [(start_time, end_time)] + + # Sort by start_time so the stitch order is deterministic. + sorted_refs = sorted(idx.segments, key=lambda r: r.start_time) + + frames: list[pd.DataFrame] = [] + present_ranges: list[TimeRange] = [] + for ref in sorted_refs: + # Skip non-overlapping segments cheaply (no segment load). + if ref.end_time < start_time or ref.start_time > end_time: + continue + entry = self._load_segment(channel_id, run_id, ref.seg_id) + if entry is None: + # Evicted by diskcache LRU (or index/segment got out of + # sync). Treat the segment's range as uncovered so the + # caller refetches it. + continue + + present_ranges.append((max(ref.start_time, start_time), min(ref.end_time, end_time))) + sliced = entry.data[start_time:end_time] # type: ignore[misc] + if len(sliced) > 0: + frames.append(sliced) + + gaps = self._compute_gaps(start_time, end_time, present_ranges) + if not frames: + return None, gaps + if len(frames) == 1: + return frames[0], gaps + # Multiple segments → stitch. ``groupby.last()`` dedups any + # boundary-overlapping timestamps and keeps the later segment's + # value on conflict (sorted by start_time above). + return pd.concat(frames).groupby(level=0).last(), gaps + + def put_segment( self, channel_id: str, - entry: ChannelCacheEntry, - *, - run_id: str | None = None, + run_id: str | None, + data: pd.DataFrame, + start_time: datetime, + end_time: datetime, ) -> None: - """Insert or replace the entry for ``(channel_id, run_id)``. + """Write a new segment and update the index. - Forwards :attr:`ChannelCacheEntry.size_bytes` to the store so its - oversize guard can decide whether to write or skip+warn. No-op - when the underlying store is disabled. + Args: + channel_id: Parent channel id (bitfield elements all share + the same id — group them upstream and pass one wide + frame here). + run_id: Per-run cache dimension; ``None`` for the unscoped + bucket. + data: This fetch's rows, indexed by tz-aware ``DatetimeIndex``. + start_time: Claimed lower bound of cache coverage for this + segment. Callers can claim more than ``data`` actually + spans to record "we asked the wire about this range" and + avoid re-fetching empty sub-ranges. + end_time: Claimed upper bound; same semantics as start. + + Per-fetch disk write is O(this segment) — no rewrite of any + already-cached segment. + + Write order is segment-first, then index: an interrupted update + leaves an unreachable orphan segment (harmless; LRU-evicts) but + never leaves the index pointing at a missing segment. """ - self._store.put(self._key(channel_id, run_id), entry, size_bytes=entry.size_bytes) + if len(data) == 0: + # Skipping empty puts keeps the segment list shorter at the + # cost of re-fetching no-data ranges. Acceptable trade for + # the draft; see the class-level TODO. + return + + size_bytes = int(data.memory_usage(deep=True).sum()) + idx = self._load_index(channel_id, run_id) or SegmentIndex( + schema_version=self.SCHEMA_VERSION + ) + seg_id = idx.next_seg_id + + entry = ChannelCacheEntry( + data=data, + start_time=start_time, + end_time=end_time, + size_bytes=size_bytes, + ) + + # Segment first. + seg_key = self._segment_key(channel_id, run_id, seg_id) + self._store.put(seg_key, entry, size_bytes=size_bytes) + if seg_key not in self._store: + # The store rejected the segment (oversize, disabled, etc.). + # Skipping the index update keeps us from leaving a + # dangling reference to a never-written segment. + return + + idx.segments.append( + SegmentRef( + seg_id=seg_id, + start_time=start_time, + end_time=end_time, + size_bytes=size_bytes, + ) + ) + idx.next_seg_id += 1 + self._store.put( + self._index_key(channel_id, run_id), + idx, + size_bytes=max(1024, 128 * len(idx.segments)), + ) def invalidate(self, channel_id: str, run_id: str | None = None) -> None: - """Remove ``(channel_id, run_id)`` from the cache. Safe when absent. + """Drop every segment in a bucket plus the index. Safe when absent. - Only invalidates the one run-scoped bucket — entries for the same - ``channel_id`` under other runs are preserved. + Only touches the one ``(channel_id, run_id)`` bucket — segments + under other runs survive. """ - self._store.invalidate(self._key(channel_id, run_id)) + idx = self._load_index(channel_id, run_id) + if idx is None: + return + for ref in idx.segments: + self._store.invalidate(self._segment_key(channel_id, run_id, ref.seg_id)) + self._store.invalidate(self._index_key(channel_id, run_id)) def clear(self) -> None: - """Wipe every channel entry across all runs. Other adapters' entries are preserved. - - Walks the shared store's keyspace once and drops anything under - :attr:`KEY_PREFIX`. ``list(...)`` snapshots the iterator since - we mutate during iteration. - """ + """Wipe every channel entry (all buckets, all runs).""" for key in list(self._store): if key.startswith(self.KEY_PREFIX): self._store.invalidate(key) + # --- internal helpers --- + + def _load_index(self, channel_id: str, run_id: str | None) -> SegmentIndex | None: + raw = self._store.get(self._index_key(channel_id, run_id)) + if not isinstance(raw, SegmentIndex): + return None + if raw.schema_version != self.SCHEMA_VERSION: + # Future migration: discard incompatible on-disk indexes. + return None + return raw + + def _load_segment( + self, channel_id: str, run_id: str | None, seg_id: int + ) -> ChannelCacheEntry | None: + raw = self._store.get(self._segment_key(channel_id, run_id, seg_id)) + if not isinstance(raw, ChannelCacheEntry): + return None + return raw + + @staticmethod + def _compute_gaps( + query_start: datetime, + query_end: datetime, + covered: list[TimeRange], + ) -> list[TimeRange]: + """Sub-ranges of ``[query_start, query_end]`` not in ``covered``. + + ``covered`` is the list of segment ranges (already clamped to + the query window). Algorithm: merge overlapping/adjacent + intervals, then sweep emitting gaps between them. + """ + if not covered: + return [(query_start, query_end)] + + sorted_ranges = sorted(covered, key=lambda r: r[0]) + merged: list[TimeRange] = [sorted_ranges[0]] + for seg_start, seg_end in sorted_ranges[1:]: + last_start, last_end = merged[-1] + if seg_start <= last_end: + merged[-1] = (last_start, max(last_end, seg_end)) + else: + merged.append((seg_start, seg_end)) + + gaps: list[TimeRange] = [] + cursor = query_start + for seg_start, seg_end in merged: + if seg_start > cursor: + gaps.append((cursor, seg_start)) + cursor = max(cursor, seg_end) + if cursor < query_end: + gaps.append((cursor, query_end)) + return gaps + class DataLowLevelClient(LowLevelClientBase, WithGrpcClient): """Low-level client for fetching channel data. @@ -250,78 +428,6 @@ async def _get_data_impl( response = cast("GetDataResponse", response) return response.data, response.next_page_token # type: ignore # mypy doesn't know RepeatedCompositeFieldContainer can be treated like a list - def _filter_cached_channels( - self, channel_ids: list[str], run_id: str | None = None - ) -> tuple[list[str], list[str]]: - """Split ``channel_ids`` into (cached-under-``run_id``, not-cached).""" - cached_channels = [] - not_cached_channels = [] - for channel_id in channel_ids: - if self.channel_cache.has(channel_id, run_id): - cached_channels.append(channel_id) - else: - not_cached_channels.append(channel_id) - return cached_channels, not_cached_channels - - def _check_cache( - self, - *, - channel_id: str, - start_time: datetime, - end_time: datetime, - run_id: str | None = None, - ) -> tuple[pd.DataFrame | None, datetime | None, datetime | None]: - """Check if the data for a channel during a run is cached and return how to query remaining data if so. - - There are a variety of requested start/end time vs cached start/end time cases to consider. - Below diagram represents time aligned ranges for each case: - - Cache interval: |-------------------------------| - Case 1: |---------------------------| - Case 2: |--------------------------------| - Case 3: |----------| - Case 4: |--------------------------------| - Case 5: |------| or |-----------------------------------------| - - Returns: - A tuple of (data, start_time, end_time) - where data is a pandas dataframe and start and end times are what should be used for the next call based on what is not covered by the cached data. - """ - cached_data = self.channel_cache.get(channel_id, run_id) - ret_start_time = start_time - ret_end_time = end_time - ret_data = None - if cached_data: - start_time_cached = cached_data.start_time - end_time_cached = cached_data.end_time - ret_data = cached_data.data - # Filter data to desiredtime range - ret_data = ret_data[start_time:end_time] # type: ignore # mypy doesn't understand pandas that well seemingly - - if start_time_cached <= start_time: - if start_time < end_time_cached: - if end_time <= end_time_cached: - # Case 1 - ret_start_time = None # type: ignore - ret_end_time = None # type: ignore - else: - # Case 2 - ret_start_time = end_time_cached - ret_end_time = end_time - else: - # Case 3 - return (None, start_time, end_time) - else: - if start_time_cached < end_time and end_time <= end_time_cached: - # Case 4 - ret_start_time = start_time - ret_end_time = start_time_cached - else: - # Case 5 - return (None, start_time, end_time) - - return (ret_data, ret_start_time, ret_end_time) - def _update_cache( self, *, @@ -330,51 +436,73 @@ def _update_cache( end_time: datetime, run_id: str | None = None, ): - """Update the cache with the new data and start/end times.""" + """Write each channel's fresh data as a new segment. + + Per-fetch disk write is O(this segment) — no merging with prior + segments and no re-pickle of accumulated data, so n sequential + incremental pulls cost O(n) total disk write instead of O(n²). + + Bitfield grouping: ``try_deserialize_channel_data`` returns one + dotted-name DataFrame per bitfield element, all mapping to the + same parent ``channel_id`` via :attr:`name_id_map`. We group + them by parent id and concat into one wide frame so each fetch + produces exactly one segment per channel, regardless of how + many elements that channel exposes. + + Empty data is skipped (no zero-row segments). This means a + no-run query that returns nothing won't record coverage, so the + next identical query will re-fetch — see the ``put_segment`` + TODO for "we tried, no data" segments. + """ assert start_time is not None assert end_time is not None name_id_map = self.channel_cache.name_id_map + # Group dotted-name frames by parent channel id so bitfield + # elements land in one segment. + by_channel_id: dict[str, list[pd.DataFrame]] = {} for channel_name, data in channel_data.items(): channel_id = name_id_map.get(channel_name) if not channel_id: raise ValueError( - f"{channel_name} not found in name_id_map. Not sure got data for this channel without a call that should've updated the map." + f"{channel_name} not found in name_id_map. Not sure got " + f"data for this channel without a call that should've " + f"updated the map." ) + by_channel_id.setdefault(channel_id, []).append(data) - suggested_start_time = start_time + for channel_id, frames in by_channel_id.items(): + if len(frames) == 1: + combined = frames[0] + else: + # Bitfield: per-element single-column frames → one wide + # frame. ``groupby.last`` dedups any boundary overlaps. + combined = pd.concat(frames).groupby(level=0).last() + + if len(combined) == 0: + continue + + # Segment coverage range. For run-scoped queries, claim + # only what the data actually spans (we can't assert + # absence outside the data — the run might not have + # started yet). For unscoped queries, claim the full + # requested range so a follow-up of the same range hits. if run_id: - if len(data) > 0: - suggested_start_time = data.index[0] - else: - # Because we didn't get any data, we can't know what the start time should be. - # And because this was queried w/ a run ID, we can't say there's no data before the run started. - # So we just don't update the cache. - continue - - existing = self.channel_cache.get(channel_id, run_id) - if existing is not None: - # ``groupby(level=0).last()`` is only needed to dedup - # overlapping timestamps. Strictly disjoint ranges - # (append after / prepend before) carry no overlap. - if len(data) > 0 and data.index[0] > existing.end_time: - merged_data = pd.concat([existing.data, data]) - elif len(data) > 0 and data.index[-1] < existing.start_time: - merged_data = pd.concat([data, existing.data]) - else: - merged_data = pd.concat([existing.data, data]).groupby(level=0).last() - entry = _new_cache_entry( - data=merged_data, - start_time=min(suggested_start_time, existing.start_time), - end_time=max(end_time, existing.end_time), - ) + seg_start = combined.index[0] + if not isinstance(seg_start, datetime): + seg_start = seg_start.to_pydatetime() + seg_end = end_time else: - entry = _new_cache_entry( - data=data, - start_time=suggested_start_time, - end_time=end_time, - ) - self.channel_cache.put(channel_id, entry, run_id=run_id) + seg_start = start_time + seg_end = end_time + + self.channel_cache.put_segment( + channel_id=channel_id, + run_id=run_id, + data=combined, + start_time=seg_start, + end_time=seg_end, + ) async def get_channel_data( self, @@ -388,25 +516,50 @@ async def get_channel_data( ignore_cache: bool = False, ) -> dict[str, pd.DataFrame]: """Get the data for a channel during a run.""" - ret_data = {} + ret_data: dict[str, pd.DataFrame] = {} # No data will be returned if end_time is not provided. start_time = start_time or datetime.fromtimestamp(0, tz=timezone.utc) end_time = end_time or datetime.now(timezone.utc) self._update_name_id_map(channels) - channel_ids = [c.id_ for c in channels] - cached_channels, not_cached_channels = ( - ([], channel_ids) - if ignore_cache - else self._filter_cached_channels(channel_ids, run_id=run_id) # type: ignore - ) + + # Two work queues. Fully uncached channels share the full range + # and get batched; partial-hit channels carry per-gap ranges + # and go one fetch at a time. + fully_uncached: list[str] = [] + partial_gaps: list[tuple[str, list[TimeRange]]] = [] + + for channel in channels: + cid = channel.id_ + assert cid is not None + if ignore_cache: + cached_data: pd.DataFrame | None = None + gaps: list[TimeRange] = [(start_time, end_time)] + else: + cached_data, gaps = self.channel_cache.get_range(cid, run_id, start_time, end_time) + + if cached_data is not None: + # Slice per column so each result key carries only its + # own element frame (matches the per-element shape + # ``try_deserialize_channel_data`` produces; without + # this slice, a bitfield's wide cached frame would land + # under every dotted key). + for name in cached_data.columns: + ret_data[name] = cached_data[[name]] + + if not gaps: + continue + if len(gaps) == 1 and gaps[0] == (start_time, end_time): + fully_uncached.append(cid) + else: + partial_gaps.append((cid, gaps)) tasks = [] - # Queue up calls for non-cached channels in batches. + # Batch fully-uncached channels (sharing the full requested + # range) into one wire call each. batch_size = REQUEST_BATCH_SIZE - for i in range(0, len(not_cached_channels), batch_size): # type: ignore - batch = not_cached_channels[i : i + batch_size] # type: ignore - + for i in range(0, len(fully_uncached), batch_size): + batch = fully_uncached[i : i + batch_size] task = asyncio.create_task( self._handle_pagination( self._get_data_impl, @@ -422,44 +575,37 @@ async def get_channel_data( ) tasks.append(task) - # Handling cached channels 1 by 1 instead of in batches to account for channels that may have been cached from calls with different start/end times. - for channel_id in cached_channels: - cached_data, new_start_time, new_end_time = self._check_cache( - channel_id=channel_id, - start_time=start_time, - end_time=end_time, - run_id=run_id, - ) - - if cached_data is not None: - for name in cached_data.columns: - ret_data[name] = cached_data[[name]] - if new_start_time is None: - # Cache fully encompassed the desired time range so don't queue a call. - continue - task = asyncio.create_task( - self._handle_pagination( - self._get_data_impl, - kwargs={ - "channel_ids": [channel_id], - "run_id": run_id, - "start_time": new_start_time, - "end_time": new_end_time or end_time, - }, - page_size=page_size, - max_results=max_results, + # Partial gaps: one fetch per (channel, gap). + for cid, gaps in partial_gaps: + for gap_start, gap_end in gaps: + task = asyncio.create_task( + self._handle_pagination( + self._get_data_impl, + kwargs={ + "channel_ids": [cid], + "run_id": run_id, + "start_time": gap_start, + "end_time": gap_end, + }, + page_size=page_size, + max_results=max_results, + ) ) - ) - tasks.append(task) + tasks.append(task) pages = await asyncio.gather(*tasks) ret_data = self._merge_pages(pages, initial=ret_data) - # Skip the cache update when no fresh pages arrived. + # Skip the cache update when no fresh pages arrived (pure cache + # hit). Avoids the redundant "rewrite same bytes" case that + # Tier 1A also targeted under the old single-entry shape. had_fresh_data = any(pages) if not ignore_cache and had_fresh_data: self._update_cache( - channel_data=ret_data, start_time=start_time, end_time=end_time, run_id=run_id + channel_data=ret_data, + start_time=start_time, + end_time=end_time, + run_id=run_id, ) return ret_data diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py index ad3935211..90b12be32 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py @@ -3,8 +3,8 @@ Five classes, narrowest scope first: * :class:`TestChannelDataCache` — the typed adapter over the shared - :class:`DiskCache`. Covers key namespacing, the isinstance guard on - ``get``, and the prefix-scoped ``clear``. + :class:`DiskCache`. Covers key namespacing, eviction-tolerant + stitching, gap computation, and the prefix-scoped ``clear``. * :class:`TestMergePages` — ``DataLowLevelClient._merge_pages``, the per-channel concat helper. * :class:`TestDataLowLevelClient` — constructor wiring and per-instance @@ -23,6 +23,14 @@ was a class attribute that grew without bound. ``test_per_instance_isolation`` is the canary that catches anyone re-introducing that pattern, even though ownership has since moved to the client. + +The cache stores data as per-fetch *segments* under one index per +``(channel_id, run_id)`` bucket. ``put_segment`` writes a new segment ++ updates the index; ``get_range`` walks the index, slices each +overlapping segment to the query window, and reports any uncovered +sub-ranges as gaps. The segment model is what eliminates the prior +write amplification on incremental pulls (where every fetch re-pickled +the entire accumulated frame). """ from __future__ import annotations @@ -214,198 +222,422 @@ async def fake_impl( # ---------- tests ----------- +def _put( + adapter: ChannelDataCache, + channel_id: str, + *, + data: pd.DataFrame | None = None, + rows: int = 5, + start: datetime = _NOW, + offset: int = 0, + freq: str = "ms", + seg_start: datetime | None = None, + seg_end: datetime | None = None, + run_id: str | None = None, +) -> pd.DataFrame: + """Convenience: build a frame, write it as one segment, return it. + + ``seg_start`` / ``seg_end`` default to the data's actual range so + tests get tightly-bounded segments unless they specifically want to + claim extra coverage. + """ + if data is None: + data = _frame(channel_id, rows=rows, start=start, offset=offset, freq=freq) + if seg_start is None: + seg_start = data.index[0].to_pydatetime() + if seg_end is None: + seg_end = data.index[-1].to_pydatetime() + adapter.put_segment( + channel_id=channel_id, + run_id=run_id, + data=data, + start_time=seg_start, + end_time=seg_end, + ) + return data + + class TestChannelDataCache: """The typed adapter over the shared :class:`DiskCache`. - Four invariants get pinned: + Five invariants get pinned across the per-segment shape: 1. Every operation routes through the namespaced key - (``channel::``), so two adapters sharing one store - don't collide on bare resource ids. + (``channel:::{idx,seg:N}``), so two adapters sharing + one store don't collide on bare resource ids. 2. Run id is part of the cache dimension: the same ``channel_id`` under two different runs is two cache buckets, not one. - 3. :meth:`ChannelDataCache.get` returns ``None`` on a type-mismatch - hit (e.g. a row another adapter wrote) instead of handing - arbitrary objects to downstream pandas code. - 4. :meth:`ChannelDataCache.clear` wipes only the adapter's namespace + 3. :meth:`ChannelDataCache.get_range` stitches multiple segments + and reports uncovered sub-ranges as gaps. Missing (evicted) + segments degrade to gaps, never to errors. + 4. :meth:`ChannelDataCache.invalidate` drops every segment in a + bucket and the index, leaving other buckets untouched. + 5. :meth:`ChannelDataCache.clear` wipes only the adapter's namespace — entries belonging to other adapters survive. Store-level behaviour (oversized guards, cross-session reload, marker-checked clear_disk) is exercised in ``test_disk_cache.py``. """ - def test_get_miss_returns_none(self, tmp_path): + def test_miss_returns_none_and_full_gap(self, tmp_path): adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "miss")) try: - assert not adapter.has("c1") - assert adapter.get("c1") is None + assert not adapter.has_any("c1") + data, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END) + assert data is None + assert gaps == [(_NOW, _WINDOW_END)] finally: adapter.store.close() - def test_round_trip(self, tmp_path): - """Put then get returns an equivalent entry.""" + def test_round_trip_single_segment(self, tmp_path): + """Put one segment, get_range back covers the whole frame and reports no gap.""" adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "rt")) try: - entry = _entry(rows=8) - adapter.put("c1", entry) - assert adapter.has("c1") - got = adapter.get("c1") + df = _put(adapter, "c1", rows=8) + assert adapter.has_any("c1") + got, gaps = adapter.get_range("c1", None, df.index[0], df.index[-1]) assert got is not None - pd.testing.assert_frame_equal(got.data, entry.data) - assert got.start_time == entry.start_time - assert got.end_time == entry.end_time + pd.testing.assert_frame_equal(got, df) + assert gaps == [] finally: adapter.store.close() - def test_writes_use_namespaced_key(self, tmp_path): - """The raw store sees ``channel::``, not the bare id. + def test_writes_use_namespaced_index_and_segment_keys(self, tmp_path): + """The raw store sees ``channel:::idx`` + ``...:seg:0``. - Pins the key-shape contract two adapters share. Without it, a - second adapter that happens to share an id with the channel - adapter would clobber the channel row. Empty run segment marks - the unscoped bucket; UUIDs can never be empty so there's no - collision risk. + Pins the per-segment key shape: one index plus one segment key + per fetch. Without the prefix, a second adapter that happens to + share an id with the channel adapter would clobber the rows. """ store = DiskCache(disk_path=tmp_path / "ns") adapter = ChannelDataCache(store) try: - adapter.put("c1", _entry(rows=4)) - assert "channel::c1" in store + _put(adapter, "c1", rows=4) + assert "channel::c1:idx" in store + assert "channel::c1:seg:0" in store assert "c1" not in store - assert "channel:c1" not in store # old (pre-run-scoping) shape + assert "channel:c1" not in store # never the bare-id shape finally: store.close() def test_run_id_is_part_of_the_key(self, tmp_path): """Same channel under two runs is two cache buckets, not one. - Regression guard for the bug this code was added to fix: a bare - ``channel:`` key conflated runs and served run-A's data to a - query for run-B. ``run_id`` is now a label on the cache - dimension; put under run-A must not be visible to a get under - run-B (or the unscoped bucket). + Regression guard for the run-scoping bug: a bare ``channel:`` + key conflated runs and served run-A's data to a query for run-B. + With segments, that turns into "the two buckets share an index" + — the test below pins them as fully independent. """ adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "runs")) try: - entry_a = _entry(rows=4) - entry_b = _entry(rows=8) - adapter.put("c1", entry_a, run_id="run-A") - adapter.put("c1", entry_b, run_id="run-B") - - assert adapter.has("c1", "run-A") - assert adapter.has("c1", "run-B") - assert not adapter.has("c1") # unscoped bucket stays empty - assert not adapter.has("c1", "run-C") # unknown run still misses - - got_a = adapter.get("c1", "run-A") - got_b = adapter.get("c1", "run-B") + df_a = _put(adapter, "c1", rows=4, run_id="run-A") + df_b = _put(adapter, "c1", rows=8, run_id="run-B") + + assert adapter.has_any("c1", "run-A") + assert adapter.has_any("c1", "run-B") + assert not adapter.has_any("c1") # unscoped bucket stays empty + assert not adapter.has_any("c1", "run-C") # unknown run still misses + + got_a, _ = adapter.get_range("c1", "run-A", df_a.index[0], df_a.index[-1]) + got_b, _ = adapter.get_range("c1", "run-B", df_b.index[0], df_b.index[-1]) assert got_a is not None + assert len(got_a) == 4 assert got_b is not None - assert len(got_a.data) == 4 - assert len(got_b.data) == 8 + assert len(got_b) == 8 finally: adapter.store.close() def test_unscoped_and_scoped_buckets_are_independent(self, tmp_path): - """An unscoped put (``run_id=None``) doesn't satisfy a run-scoped get. - - Pins the other direction of the run-scope contract. - """ + """An unscoped put (``run_id=None``) doesn't satisfy a run-scoped get.""" adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "indep")) try: - adapter.put("c1", _entry(rows=4)) # no run - assert adapter.has("c1") - assert not adapter.has("c1", "run-A") - assert adapter.get("c1", "run-A") is None + _put(adapter, "c1", rows=4) # no run + assert adapter.has_any("c1") + assert not adapter.has_any("c1", "run-A") + data, gaps = adapter.get_range("c1", "run-A", _NOW, _WINDOW_END) + assert data is None + assert gaps == [(_NOW, _WINDOW_END)] finally: adapter.store.close() - def test_get_isinstance_check_filters_foreign_rows(self, tmp_path): - """A row whose value isn't a ChannelCacheEntry reads as a miss. + def test_get_range_isinstance_filters_foreign_segments(self, tmp_path): + """A segment row with the wrong shape reads as a miss → gap. Models a corrupt entry or a key collision from another writer. - ``ChannelDataCache.get`` must isinstance-check the raw value so - callers downstream never receive the wrong shape. + ``_load_segment`` isinstance-checks before handing back; the + evicted-segment-as-gap fallback covers this too. """ store = DiskCache(disk_path=tmp_path / "foreign") adapter = ChannelDataCache(store) try: - store.put("channel::c1", {"not": "an entry"}, size_bytes=64) - assert adapter.get("c1") is None + _put(adapter, "c1", rows=4) + # Overwrite the segment's payload with foreign data; the + # index still claims it exists, so the read should treat + # the segment range as a gap. + store.put("channel::c1:seg:0", {"not": "an entry"}, size_bytes=64) + data, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END) + assert data is None + # The whole query range is uncovered (one merged gap). + assert gaps == [(_NOW, _WINDOW_END)] finally: store.close() def test_invalidate_is_run_scoped(self, tmp_path): - """``invalidate`` only drops the named ``(channel, run)`` bucket. - - Entries for the same channel under other runs survive — runs - are independent cache dimensions, so dropping one shouldn't - cascade. - """ + """``invalidate`` only drops the named ``(channel, run)`` bucket.""" adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "inval")) try: adapter.invalidate("never_added") # safe before any puts - adapter.put("c1", _entry(rows=4), run_id="run-A") - adapter.put("c1", _entry(rows=8), run_id="run-B") + _put(adapter, "c1", rows=4, run_id="run-A") + _put(adapter, "c1", rows=8, run_id="run-B") adapter.invalidate("c1", "run-A") - assert not adapter.has("c1", "run-A") - assert adapter.get("c1", "run-A") is None - assert adapter.has("c1", "run-B") # run-B survives + assert not adapter.has_any("c1", "run-A") + assert adapter.has_any("c1", "run-B") # run-B survives finally: adapter.store.close() def test_clear_is_prefix_scoped(self, tmp_path): - """``clear`` drops channel rows across all runs, leaves other adapters alone. - - Simulates a second resource writing to the same store with a - different prefix; the channel adapter's clear must not be a - whole-store wipe, but it must reach every run-scoped bucket. - """ + """``clear`` drops channel rows across all runs, leaves other adapters alone.""" store = DiskCache(disk_path=tmp_path / "scoped") adapter = ChannelDataCache(store) try: - adapter.put("c1", _entry(rows=4)) # unscoped - adapter.put("c2", _entry(rows=4), run_id="run-A") - # Simulate a row written by a different adapter. + _put(adapter, "c1", rows=4) # unscoped + _put(adapter, "c2", rows=4, run_id="run-A") store.put("other:1", "foreign-value", size_bytes=64) adapter.clear() - assert not adapter.has("c1") - assert not adapter.has("c2", "run-A") + assert not adapter.has_any("c1") + assert not adapter.has_any("c2", "run-A") assert "other:1" in store finally: store.close() def test_size_bytes_propagates_to_store(self, tmp_path): - """The adapter forwards the entry's ``size_bytes`` to the store guard. + """Oversized segments are skipped by the store; index/segment-write order matters. - Sized below the entry's actual ``size_bytes`` so the store's - oversize guard kicks in. The adapter never measures size itself; - it relies on ``_new_cache_entry`` having stamped the value. + The segment is written first, then the index. When the store + refuses the segment (oversize), the index stays empty and + ``has_any`` reports false. Without this ordering you'd get an + index entry pointing at a missing segment. """ - entry = _entry(rows=10_000) - store = DiskCache(disk_path=tmp_path / "size", disk_max_bytes=entry.size_bytes // 2) + big = _frame("c1", rows=10_000) + size_bytes = int(big.memory_usage(deep=True).sum()) + store = DiskCache(disk_path=tmp_path / "size", disk_max_bytes=size_bytes // 2) adapter = ChannelDataCache(store) try: - adapter.put("c1", entry) - assert not adapter.has("c1") # oversize skipped by the store + adapter.put_segment( + "c1", None, big, big.index[0].to_pydatetime(), big.index[-1].to_pydatetime() + ) + assert not adapter.has_any("c1") finally: store.close() def test_no_op_store_keeps_adapter_silent(self): - """An adapter on a disabled store behaves like a cold cache. - - Disabling the store is the path ``client.cache.disable()`` - exercises; resources can keep their adapter reference and every - operation just no-ops. - """ + """An adapter on a disabled store behaves like a cold cache.""" adapter = ChannelDataCache(DiskCache()) assert not adapter.store.disk_enabled - adapter.put("c1", _entry(rows=4)) - assert not adapter.has("c1") - assert adapter.get("c1") is None + _put(adapter, "c1", rows=4) + assert not adapter.has_any("c1") + data, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END) + assert data is None + assert gaps == [(_NOW, _WINDOW_END)] adapter.invalidate("c1") adapter.clear() + # --- stitching + gap behaviour (new with the per-segment shape) --- + + def test_get_range_stitches_multiple_segments(self, tmp_path): + """Two segments whose claimed ranges together cover the query → one stitched frame. + + Segments claim ``[seg_start, seg_end]`` boundaries that abut at + the query midpoint so gap math reports zero gaps. The stitch + path concats and dedups via ``groupby.last``. + """ + adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "stitch")) + try: + df1 = _frame("c1", rows=5, start=_NOW, freq="ms", offset=0) + df1_end = df1.index[-1].to_pydatetime() + df2 = _frame( + "c1", + rows=5, + start=df1_end + timedelta(milliseconds=1), + freq="ms", + offset=100, + ) + query_start = df1.index[0].to_pydatetime() + query_end = df2.index[-1].to_pydatetime() + # Claim seg2 starts right where seg1 ends so the abutting + # ranges leave no gap. Real callers do this via + # ``_update_cache`` claiming the requested window. + adapter.put_segment("c1", None, df1, query_start, df1_end) + adapter.put_segment("c1", None, df2, df1_end, query_end) + got, gaps = adapter.get_range("c1", None, query_start, query_end) + assert got is not None + assert len(got) == 10 + expected = pd.concat([df1, df2]).groupby(level=0).last() + pd.testing.assert_frame_equal(got.sort_index(), expected.sort_index(), check_freq=False) + assert gaps == [] # abutting claimed ranges → no gap + finally: + adapter.store.close() + + def test_get_range_reports_internal_gap_between_segments(self, tmp_path): + """Query window wider than the two segments → one gap between them.""" + adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "midgap")) + try: + seg_a_start = _NOW + seg_a_end = _NOW + timedelta(seconds=5) + seg_b_start = _NOW + timedelta(seconds=10) + seg_b_end = _NOW + timedelta(seconds=15) + adapter.put_segment( + "c1", + None, + _frame("c1", rows=2, start=seg_a_start, freq="s"), + seg_a_start, + seg_a_end, + ) + adapter.put_segment( + "c1", + None, + _frame("c1", rows=2, start=seg_b_start, freq="s"), + seg_b_start, + seg_b_end, + ) + _, gaps = adapter.get_range("c1", None, seg_a_start, seg_b_end) + assert gaps == [(seg_a_end, seg_b_start)] + finally: + adapter.store.close() + + def test_get_range_reports_outer_gaps(self, tmp_path): + """Query wider than the cached segment on both sides → two gaps.""" + adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "outer")) + try: + seg_start = _NOW + timedelta(seconds=5) + seg_end = _NOW + timedelta(seconds=10) + adapter.put_segment( + "c1", + None, + _frame("c1", rows=2, start=seg_start, freq="s"), + seg_start, + seg_end, + ) + query_end = _NOW + timedelta(seconds=15) + _, gaps = adapter.get_range("c1", None, _NOW, query_end) + assert gaps == [(_NOW, seg_start), (seg_end, query_end)] + finally: + adapter.store.close() + + def test_evicted_segment_degrades_to_gap(self, tmp_path): + """If the store loses a segment (LRU), the reader treats its range as a gap. + + Pins the eviction-tolerance contract. The index can outlive + its segments under memory pressure; reads must not error and + must surface the uncovered range so the caller refetches. + """ + store = DiskCache(disk_path=tmp_path / "evict") + adapter = ChannelDataCache(store) + try: + df = _put(adapter, "c1", rows=5) + store.invalidate("channel::c1:seg:0") # simulate eviction + data, gaps = adapter.get_range("c1", None, df.index[0], df.index[-1]) + assert data is None + assert gaps == [(df.index[0], df.index[-1])] + finally: + store.close() + + def test_segment_count_grows_with_each_put(self, tmp_path): + """``put_segment`` writes a new key per call — no merging on the write path. + + Pins the "no rewrite of accumulated bytes per fetch" contract. + ``next_seg_id`` advances; raw store key count grows by one + index update + one new segment per put. + """ + store = DiskCache(disk_path=tmp_path / "grow") + adapter = ChannelDataCache(store) + try: + for i in range(3): + _put( + adapter, + "c1", + rows=2, + start=_NOW + timedelta(seconds=i * 10), + freq="s", + ) + # Expect: 1 index + 3 segments. + channel_keys = sorted(k for k in store if k.startswith("channel:")) + assert channel_keys == [ + "channel::c1:idx", + "channel::c1:seg:0", + "channel::c1:seg:1", + "channel::c1:seg:2", + ] + finally: + store.close() + + def test_empty_put_is_skipped(self, tmp_path): + """``put_segment`` with an empty frame is a no-op (no index update). + + Documented trade-off: no "we tried, no data" segments. A + follow-up identical query will re-fetch. See class TODO. + """ + adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "empty")) + try: + empty = pd.DataFrame( + {"c1": []}, + index=pd.DatetimeIndex([], tz=timezone.utc), + ) + adapter.put_segment("c1", None, empty, _NOW, _WINDOW_END) + assert not adapter.has_any("c1") + finally: + adapter.store.close() + + # --- gap-math unit tests --- + + @pytest.mark.parametrize( + ("covered", "expected"), + [ + # Fully uncovered. + ([], [(_NOW, _NOW + timedelta(seconds=10))]), + # Fully covered. + ([(_NOW, _NOW + timedelta(seconds=10))], []), + # Left-only gap. + ( + [(_NOW + timedelta(seconds=5), _NOW + timedelta(seconds=10))], + [(_NOW, _NOW + timedelta(seconds=5))], + ), + # Right-only gap. + ( + [(_NOW, _NOW + timedelta(seconds=5))], + [(_NOW + timedelta(seconds=5), _NOW + timedelta(seconds=10))], + ), + # Two segments, one internal gap. + ( + [ + (_NOW, _NOW + timedelta(seconds=2)), + (_NOW + timedelta(seconds=7), _NOW + timedelta(seconds=10)), + ], + [(_NOW + timedelta(seconds=2), _NOW + timedelta(seconds=7))], + ), + # Overlapping covered ranges merge before gap math. + ( + [ + (_NOW, _NOW + timedelta(seconds=6)), + (_NOW + timedelta(seconds=4), _NOW + timedelta(seconds=8)), + ], + [(_NOW + timedelta(seconds=8), _NOW + timedelta(seconds=10))], + ), + ], + ids=[ + "fully_uncovered", + "fully_covered", + "left_gap", + "right_gap", + "internal_gap", + "overlapping_covered_merges", + ], + ) + def test_compute_gaps(self, covered, expected): + query_start = _NOW + query_end = _NOW + timedelta(seconds=10) + assert ChannelDataCache._compute_gaps(query_start, query_end, covered) == expected + class TestMergePages: """Behaviour of :meth:`DataLowLevelClient._merge_pages`. @@ -417,8 +649,8 @@ class TestMergePages: * Single-frame channels skip the concat entirely (cheap identity path). * Multi-frame channels concat in collected order; ``groupby.last`` makes the latest frame win on overlapping timestamps. - * Cached slices from ``_check_cache`` are folded in as the *first* - frame so fresh pages still win on overlap. + * Cached slices from the segment-stitched read are folded in as + the *first* frame so fresh pages still win on overlap. """ @pytest.mark.parametrize("pages", [[], [[]]], ids=["no_tasks_queued", "task_returned_empty"]) @@ -466,7 +698,7 @@ def test_overlapping_timestamps_later_page_wins(self) -> None: assert (result["chan"]["chan"] == 99).all() def test_cached_slice_folded_in_first_and_loses_on_overlap(self) -> None: - """Cached slice from ``_check_cache`` is the first frame in the merge. + """Cached slice from the segment read is the first frame in the merge. Fresh pages must overwrite cached values on duplicate timestamps, matching the pre-existing "latest fetch wins" semantic. @@ -532,9 +764,9 @@ class attribute, so every ``SiftClient`` in the process appended client_a = _client_with_cache(tmp_path, "a") client_b = _client_with_cache(tmp_path, "b") try: - client_a.channel_cache.put("c1", _entry(rows=10)) - assert client_a.channel_cache.has("c1") - assert not client_b.channel_cache.has("c1") + _put(client_a.channel_cache, "c1", rows=10) + assert client_a.channel_cache.has_any("c1") + assert not client_b.channel_cache.has_any("c1") finally: client_a.channel_cache.store.close() client_b.channel_cache.store.close() @@ -702,9 +934,9 @@ async def test_run_id_keeps_cache_buckets_separate(self, tmp_path) -> None: pd.testing.assert_frame_equal(second["c1"].sort_index(), df_b.sort_index()) # Both runs end up cached independently; the unscoped bucket # stays empty because every query named a run. - assert client.channel_cache.has("c1", "run-A") - assert client.channel_cache.has("c1", "run-B") - assert not client.channel_cache.has("c1") + assert client.channel_cache.has_any("c1", "run-A") + assert client.channel_cache.has_any("c1", "run-B") + assert not client.channel_cache.has_any("c1") finally: client.channel_cache.store.close() @@ -729,13 +961,13 @@ async def test_pure_cache_hit_does_not_rewrite_disk(self, tmp_path) -> None: ) put_calls: list[Any] = [] - original_put = client.channel_cache.put + original_put = client.channel_cache.put_segment - def spy_put(*args: Any, **kwargs: Any) -> None: + def spy_put_segment(*args: Any, **kwargs: Any) -> None: put_calls.append((args, kwargs)) original_put(*args, **kwargs) - client.channel_cache.put = spy_put # type: ignore[method-assign] + client.channel_cache.put_segment = spy_put_segment # type: ignore[method-assign] with _fake_grpc(client, {"c1": []}): await client.get_channel_data( channels=[_channel("c1")], @@ -747,15 +979,12 @@ def spy_put(*args: Any, **kwargs: Any) -> None: client.channel_cache.store.close() @pytest.mark.asyncio - async def test_disjoint_forward_paging_preserves_all_rows(self, tmp_path) -> None: - """Strictly disjoint append skips ``groupby.last`` and keeps every row. - - Pins the Tier-1C optimization: when the second query's range - starts strictly after the cached range, ``_update_cache`` uses - plain ``pd.concat`` instead of ``concat + groupby.last``. The - groupby would be correct but wasted (no overlap to dedup); this - test guards against a future regression that drops rows on the - fast path. + async def test_disjoint_forward_paging_lands_in_separate_segments(self, tmp_path) -> None: + """Two disjoint forward pulls write two segments and stitch on read. + + Under the segment shape, each fetch lands in its own cache key + — no merging on the write path. A subsequent read stitches the + segments back together via ``get_range``. """ client = _client_with_cache(tmp_path) df1 = _frame("c1", rows=5, start=_NOW, offset=0) @@ -775,10 +1004,20 @@ async def test_disjoint_forward_paging_preserves_all_rows(self, tmp_path) -> Non end_time=_WINDOW_END, ) - entry = client.channel_cache.get("c1") - assert entry is not None - assert len(entry.data) == 10 - assert set(entry.data.columns) == {"c1"} + # Two segments under the parent id (no rewrite on the second + # fetch — that's the whole point of the per-segment shape). + idx = client.channel_cache._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 2 + + stitched, gaps = client.channel_cache.get_range("c1", None, _NOW, _WINDOW_END) + assert stitched is not None + assert len(stitched) == 10 + assert set(stitched.columns) == {"c1"} + # Reads still report the truly-uncovered window between the + # two segments (segments only claim what their data spans + # when run-scoped, but here we used the requested ranges). + assert gaps == [] finally: client.channel_cache.store.close() @@ -801,7 +1040,7 @@ async def test_ignore_cache_true_returns_fresh_and_skips_write(self, tmp_path) - ignore_cache=True, ) pd.testing.assert_frame_equal(result["c1"], df) - assert not client.channel_cache.has("c1") + assert not client.channel_cache.has_any("c1") finally: client.channel_cache.store.close() @@ -821,9 +1060,9 @@ class TestBitFieldChannels: every element, all pointing at the parent id — so when ``_update_cache`` looks up a dotted name, it lands on the parent's cache row. - 2. Element frames merge into a single wide-DataFrame cache entry - under the parent id, and ``_filter_cached_channels`` correctly - treats the bitfield as cached as a unit. + 2. Element frames merge into a single wide-DataFrame segment + under the parent id, and ``get_range`` correctly treats the + bitfield as cached as a unit (zero gaps for the warmed window). 3. Fresh-fetch and cache-hit return shapes match — both produce one single-column frame per dotted element name. The cache-hit branch slices the wide cached frame per column to keep this @@ -847,13 +1086,12 @@ def test_update_name_id_map_populates_dotted_and_parent_names(self) -> None: "ch1.hi": "abc", } - def test_update_cache_merges_elements_under_parent_id(self, tmp_path) -> None: - """All element frames land in one cache entry keyed by the parent id. + def test_update_cache_merges_elements_into_one_segment(self, tmp_path) -> None: + """All element frames land in one segment keyed by the parent id. - ``_update_cache`` iterates element frames; the first put writes - a single-column entry, the second sees ``existing`` for the - same parent id and merges via - ``pd.concat([...]).groupby(level=0).last()`` into a wide frame. + ``_update_cache`` groups frames by their parent id before + writing, so bitfield elements that came in one wire fetch + produce exactly one wide segment — not one segment per element. Nothing should land under a dotted-name key. """ client = _client_with_cache(tmp_path) @@ -867,25 +1105,31 @@ def test_update_cache_merges_elements_under_parent_id(self, tmp_path) -> None: start_time=_NOW, end_time=_WINDOW_END, ) - assert client.channel_cache.has("abc") - assert not client.channel_cache.has("ch1.lo") - assert not client.channel_cache.has("ch1.hi") - entry = client.channel_cache.get("abc") - assert entry is not None - assert set(entry.data.columns) == {"ch1.lo", "ch1.hi"} - # Overlapping indices collapse via groupby.last; 5 rows in, - # 5 rows out (with both element columns populated). - assert len(entry.data) == 5 + assert client.channel_cache.has_any("abc") + assert not client.channel_cache.has_any("ch1.lo") + assert not client.channel_cache.has_any("ch1.hi") + + # Exactly one segment under the parent id holding both + # element columns. ``get_range`` returns the wide frame; + # downstream slicing happens in ``get_channel_data``. + idx = client.channel_cache._load_index("abc", None) + assert idx is not None + assert len(idx.segments) == 1 + + data, _ = client.channel_cache.get_range("abc", None, _NOW, _WINDOW_END) + assert data is not None + assert set(data.columns) == {"ch1.lo", "ch1.hi"} + assert len(data) == 5 finally: client.channel_cache.store.close() - def test_filter_cached_channels_treats_bitfield_as_one(self, tmp_path) -> None: - """``_filter_cached_channels`` consults the parent id, not element names. + def test_get_range_treats_bitfield_as_one_bucket(self, tmp_path) -> None: + """A bitfield warmed by a prior fetch reports zero gaps for the same range. Pins the read side of the id/name asymmetry: a bitfield warmed - by a prior fetch is "cached" as a unit. Without this, a second - call would re-issue the wire fetch for the parent id even - though the elements are in the cache. + by a prior fetch is "cached" as a unit under the parent id. + Without this, a second call would re-fetch the whole window + even though the elements are in the cache. """ client = _client_with_cache(tmp_path) ch = _bitfield_channel(cid="abc", name="ch1", elements=["lo", "hi"]) @@ -899,9 +1143,9 @@ def test_filter_cached_channels_treats_bitfield_as_one(self, tmp_path) -> None: start_time=_NOW, end_time=_WINDOW_END, ) - cached, not_cached = client._filter_cached_channels(["abc"]) - assert cached == ["abc"] - assert not_cached == [] + data, gaps = client.channel_cache.get_range("abc", None, _NOW, _WINDOW_END) + assert data is not None + assert gaps == [] # parent id covers the whole window finally: client.channel_cache.store.close() @@ -932,10 +1176,10 @@ async def test_get_channel_data_fresh_bitfield_returns_per_element_frames( assert list(result["ch1.hi"].columns) == ["ch1.hi"] pd.testing.assert_frame_equal(result["ch1.lo"].sort_index(), df_lo.sort_index()) pd.testing.assert_frame_equal(result["ch1.hi"].sort_index(), df_hi.sort_index()) - # Cache holds one wide entry under the parent id. - entry = client.channel_cache.get("abc") - assert entry is not None - assert set(entry.data.columns) == {"ch1.lo", "ch1.hi"} + # Cache holds one wide segment under the parent id. + data, _ = client.channel_cache.get_range("abc", None, _NOW, _WINDOW_END) + assert data is not None + assert set(data.columns) == {"ch1.lo", "ch1.hi"} finally: client.channel_cache.store.close() diff --git a/python/lib/sift_client/_tests/test_client_cache.py b/python/lib/sift_client/_tests/test_client_cache.py index 414b44b00..d1de3f866 100644 --- a/python/lib/sift_client/_tests/test_client_cache.py +++ b/python/lib/sift_client/_tests/test_client_cache.py @@ -234,9 +234,9 @@ class TestStats: 2. **Enabled, empty** — ``enabled=True``, ``path`` populated, sizes/counts at zero. Distinguishes "nothing cached yet" from "no cache configured". - 3. **Enabled, populated** — channel entries count matches what the - adapter wrote, and foreign-prefix rows don't bleed into the - channel counter. + 3. **Enabled, populated** — bucket and segment counters reflect + what the adapter wrote, and foreign-prefix rows don't bleed + into the channel counters. """ def test_stats_when_disabled(self): @@ -249,7 +249,8 @@ def test_stats_when_disabled(self): assert stats.max_bytes is None assert stats.size_bytes == 0 assert stats.entry_count == 0 - assert stats.channel_entries == 0 + assert stats.channel_buckets == 0 + assert stats.channel_segments == 0 assert "disabled" in str(stats).lower() def test_stats_when_enabled_empty(self, tmp_path): @@ -263,55 +264,68 @@ def test_stats_when_enabled_empty(self, tmp_path): assert stats.path == path assert stats.max_bytes == 8 * 1024 * 1024 assert stats.entry_count == 0 - assert stats.channel_entries == 0 + assert stats.channel_buckets == 0 + assert stats.channel_segments == 0 # path appears in the friendly print assert path in str(stats) finally: client._disk_cache.close() # type: ignore[union-attr] - def test_stats_counts_channel_entries(self, tmp_path): - """Channel writes increment ``channel_entries`` and ``entry_count``. + def test_stats_counts_buckets_and_segments(self, tmp_path): + """Writes increment ``channel_buckets`` and ``channel_segments``. Uses ``ChannelDataCache`` directly (rather than the ``DataLowLevelClient`` end-to-end path) so the test stays - focused on the stats accounting. + focused on the stats accounting. Two of the three channels + get a second segment so the bucket-vs-segment split is + observable, not just N == N. """ + from datetime import datetime, timedelta, timezone + import pandas as pd - from sift_client._internal.low_level_wrappers.data import ( - ChannelDataCache, - _new_cache_entry, - ) + from sift_client._internal.low_level_wrappers.data import ChannelDataCache client = _make_client() client.cache.enable(path=str(tmp_path / "stats")) try: store = client._get_disk_cache() adapter = ChannelDataCache(store) + base = datetime(2025, 1, 1, tzinfo=timezone.utc) for i, cid in enumerate(("c1", "c2", "c3")): df = pd.DataFrame( {cid: [float(i)]}, - index=pd.date_range("2025-01-01", periods=1, freq="s", tz="UTC"), + index=pd.date_range(base, periods=1, freq="s", tz="UTC"), ) - adapter.put( + adapter.put_segment( cid, - _new_cache_entry( - data=df, - start_time=df.index[0].to_pydatetime(), - end_time=df.index[-1].to_pydatetime(), - ), + None, + df, + start_time=base, + end_time=base, + ) + # Append a second, disjoint segment to two of the buckets + # so segments > buckets. + for cid in ("c1", "c2"): + later = base + timedelta(minutes=5) + df = pd.DataFrame( + {cid: [9.0]}, + index=pd.date_range(later, periods=1, freq="s", tz="UTC"), ) + adapter.put_segment(cid, None, df, start_time=later, end_time=later) stats = client.cache.stats() assert stats.enabled is True - assert stats.channel_entries == 3 - assert stats.entry_count == 3 + assert stats.channel_buckets == 3 + assert stats.channel_segments == 5 + # 3 index rows + 5 segment rows. + assert stats.entry_count == 8 assert stats.size_bytes > 0 finally: client._disk_cache.close() # type: ignore[union-attr] def test_stats_ignores_foreign_adapter_keys_in_channel_count(self, tmp_path): - """Keys outside the channel namespace don't bump ``channel_entries``. + """Keys outside the channel namespace don't bump channel counters. Pins the prefix-scoping so a future second adapter doesn't double-count here. @@ -325,7 +339,8 @@ def test_stats_ignores_foreign_adapter_keys_in_channel_count(self, tmp_path): stats = client.cache.stats() assert stats.entry_count == 2 - assert stats.channel_entries == 0 + assert stats.channel_buckets == 0 + assert stats.channel_segments == 0 finally: client._disk_cache.close() # type: ignore[union-attr] diff --git a/python/lib/sift_client/util/cache.py b/python/lib/sift_client/util/cache.py index 92d417473..265e3e863 100644 --- a/python/lib/sift_client/util/cache.py +++ b/python/lib/sift_client/util/cache.py @@ -67,11 +67,17 @@ class CacheStats: Tends to trend slightly higher than the sum of per-entry ``size_bytes`` the resources hand to the store. * **entry_count** — total cache keys across all adapter prefixes - (channel entries + any future foreign-adapter rows). - * **channel_entries** — channel cache entries (one per - ``(channel_id, run_id)`` bucket under the current single-entry - shape). Counted by walking the channel adapter's namespace - prefix. + (channel index + segment rows, plus any future foreign-adapter + rows). + * **channel_buckets** — number of distinct + ``(channel_id, run_id)`` channel buckets currently cached. + Counted by walking the channel adapter's index keys; one index + per bucket regardless of how many segments live under it. + * **channel_segments** — total cached segment frames across every + bucket. Each ``put_segment`` call adds one; ``diskcache`` LRU + evictions can drop them independently of their parent index, so + this can be lower than the sum of segments the adapter has + written over the session. ``str(stats)`` prints a multi-line summary suitable for notebook/REPL display; the structured fields are available for @@ -83,7 +89,8 @@ class CacheStats: max_bytes: int | None size_bytes: int entry_count: int - channel_entries: int + channel_buckets: int + channel_segments: int def __str__(self) -> str: if not self.enabled: @@ -94,7 +101,9 @@ def __str__(self) -> str: "Sift cache:\n" f" path: {self.path}\n" f" used: {_format_bytes(self.size_bytes)} / {cap}{pct}\n" - f" entries: {self.entry_count} ({self.channel_entries} channel)" + f" channels: {self.channel_buckets} buckets, " + f"{self.channel_segments} segments\n" + f" entries: {self.entry_count} total" ) @@ -191,7 +200,8 @@ def stats(self) -> CacheStats: Sift cache: path: /tmp/sift-data-cache used: 142.3 MiB / 4.0 GiB (3.5%) - entries: 487 (487 channel) + channels: 12 buckets, 487 segments + entries: 499 total """ # Importing here keeps this module light at import time — pulling # the adapter pulls pandas, which is the whole reason @@ -211,19 +221,33 @@ def stats(self) -> CacheStats: max_bytes=None, size_bytes=0, entry_count=0, - channel_entries=0, + channel_buckets=0, + channel_segments=0, ) # One pass over the keyspace. Cheap — diskcache keys are SQLite # rows, and we only touch metadata (no value loads). - channel_entries = sum(1 for key in store if key.startswith(ChannelDataCache.KEY_PREFIX)) + # Bucket index keys end in ``:idx`` and segment keys carry + # ``:seg:`` inside the suffix; those two shapes partition the + # channel adapter's namespace cleanly. + channel_prefix = ChannelDataCache.KEY_PREFIX + channel_buckets = 0 + channel_segments = 0 + for key in store: + if not key.startswith(channel_prefix): + continue + if key.endswith(":idx"): + channel_buckets += 1 + elif ":seg:" in key: + channel_segments += 1 return CacheStats( enabled=True, path=store.disk_path, max_bytes=store.disk_max_bytes, size_bytes=store.volume(), entry_count=len(store), - channel_entries=channel_entries, + channel_buckets=channel_buckets, + channel_segments=channel_segments, ) def clear(self, path: str | os.PathLike[str] | None = None) -> None: From f7aede89a36f49c3ebfd4291449505f983f2ef23 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 30 Jun 2026 11:28:18 -0700 Subject: [PATCH 2/7] add compaction and clean up cache implementaiton a bit. --- .../_internal/low_level_wrappers/data.py | 412 +++++++++++++----- .../_internal/low_level_wrappers/test_data.py | 370 ++++++++++++++-- 2 files changed, 654 insertions(+), 128 deletions(-) diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data.py b/python/lib/sift_client/_internal/low_level_wrappers/data.py index 309eec3cc..5a4919f10 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data.py @@ -36,43 +36,33 @@ REQUEST_BATCH_SIZE = 1 -class ChannelCacheEntry(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True) - data: pd.DataFrame - start_time: datetime - end_time: datetime - size_bytes: int - - -def _new_cache_entry( - data: pd.DataFrame, start_time: datetime, end_time: datetime -) -> ChannelCacheEntry: - return ChannelCacheEntry( - data=data, - start_time=start_time, - end_time=end_time, - size_bytes=int(data.memory_usage(deep=True).sum()), - ) - - TimeRange = Tuple[datetime, datetime] class SegmentRef(BaseModel): - """Pointer to one cached segment, stored on the per-bucket index.""" + """Pointer to one cached segment, stored on the per-bucket index. + + ``seg_id`` is ``None`` for an **empty ref** — a record of "we + queried this range and the wire returned no data". Empty refs + contribute to coverage (so a repeat of a known-empty range + doesn't hit the wire). + """ model_config = ConfigDict(frozen=True) - seg_id: int + seg_id: int | None start_time: datetime end_time: datetime - size_bytes: int + + @property + def is_empty(self) -> bool: + """True for an empty ref (no backing segment body).""" + return self.seg_id is None class SegmentIndex(BaseModel): """Per-(channel, run) index of cached segments.""" model_config = ConfigDict(arbitrary_types_allowed=True) - schema_version: int = 1 next_seg_id: int = 0 segments: list[SegmentRef] = [] @@ -83,11 +73,17 @@ class ChannelDataCache: Each ``(channel_id, run_id)`` bucket is split across two key shapes in the underlying store: - * One **index** entry (``channel:::idx``) holding a + * One **index** entry (``channel:v2:::idx``) holding a :class:`SegmentIndex` — a tiny list of :class:`SegmentRef` ptrs - describing every segment that exists for the bucket. - * One **segment** entry per fetch (``channel:::seg:``) - holding the :class:`ChannelCacheEntry` for that fetch's slice. + describing every segment that exists for the bucket. Some refs + are *empty* (``seg_id is None``) — they record "we queried this + range and the wire returned no data" without a backing segment. + * One **segment** entry per fetch with data + (``channel:v2:::seg:``) holding the + :class:`pandas.DataFrame` for that fetch's slice. The + :class:`SegmentRef` on the index already carries the claimed + time range and is the source of truth for coverage, so segment + bodies are stored as raw frames. Reads stitch the relevant segments together via :meth:`get_range`, which also reports which sub-ranges of the requested window have no @@ -100,18 +96,26 @@ class ChannelDataCache: and will eventually LRU-evict. Attributes: - KEY_PREFIX: Namespace prefix for every key this adapter writes - to the shared :class:`DiskCache`. Picked at class scope so - adapters in other resources can pick distinct prefixes - without runtime negotiation. - SCHEMA_VERSION: Index entry schema version. Bump when the - shape changes incompatibly so older on-disk indexes are - discarded on read rather than mis-deserialized. + KEY_PREFIX: Versioned namespace prefix for every key this + adapter writes to the shared :class:`DiskCache`. Picked at + class scope so adapters in other resources can pick + distinct prefixes without runtime negotiation. The ``v1`` + suffix is the schema version: bump it (e.g. to + ``"channel:v2:"``) when either entry shape changes + incompatibly so old keys are silently unreachable rather + than mis-deserialized. + MAX_SEGMENTS_PER_BUCKET: Per-bucket cap on segment count + before :meth:`_compact_bucket` folds them into one. + Without compaction a long incrementally-pulled run grows + ``len(idx.segments)`` without bound, so every full-range + ``get_range`` would load and concat all segments — + swapping the old O(n^2) write for unbounded read fan-out. + The cap bounds that fan-out at a known constant. """ - KEY_PREFIX: str = "channel:" - SCHEMA_VERSION: int = 1 + KEY_PREFIX: str = "channel:v2:" + MAX_SEGMENTS_PER_BUCKET: int = 16 def __init__(self, store: DiskCache): """Wrap ``store`` with channel-data semantics. @@ -188,15 +192,25 @@ def get_range( # Skip non-overlapping segments cheaply (no segment load). if ref.end_time < start_time or ref.start_time > end_time: continue - entry = self._load_segment(channel_id, run_id, ref.seg_id) - if entry is None: + + clamped = (max(ref.start_time, start_time), min(ref.end_time, end_time)) + + if ref.is_empty: + # Empty ref: the wire was asked about this range and + # returned no data. Count it toward coverage so the + # caller doesn't refetch. + present_ranges.append(clamped) + continue + + frame = self._load_segment(channel_id, run_id, ref.seg_id) + if frame is None: # Evicted by diskcache LRU (or index/segment got out of # sync). Treat the segment's range as uncovered so the # caller refetches it. continue - present_ranges.append((max(ref.start_time, start_time), min(ref.end_time, end_time))) - sliced = entry.data[start_time:end_time] # type: ignore[misc] + present_ranges.append(clamped) + sliced = frame[start_time:end_time] # type: ignore[misc] if len(sliced) > 0: frames.append(sliced) @@ -226,7 +240,8 @@ def put_segment( frame here). run_id: Per-run cache dimension; ``None`` for the unscoped bucket. - data: This fetch's rows, indexed by tz-aware ``DatetimeIndex``. + data: This fetch's rows, indexed by tz-aware + ``DatetimeIndex``. start_time: Claimed lower bound of cache coverage for this segment. Callers can claim more than ``data`` actually spans to record "we asked the wire about this range" and @@ -238,61 +253,53 @@ def put_segment( Write order is segment-first, then index: an interrupted update leaves an unreachable orphan segment (harmless; LRU-evicts) but - never leaves the index pointing at a missing segment. + never leaves the index pointing at a missing segment. Empty + puts skip the segment write entirely and update the index + directly. """ - if len(data) == 0: - # Skipping empty puts keeps the segment list shorter at the - # cost of re-fetching no-data ranges. Acceptable trade for - # the draft; see the class-level TODO. - return + idx = self._load_index(channel_id, run_id) or SegmentIndex() - size_bytes = int(data.memory_usage(deep=True).sum()) - idx = self._load_index(channel_id, run_id) or SegmentIndex( - schema_version=self.SCHEMA_VERSION - ) - seg_id = idx.next_seg_id - - entry = ChannelCacheEntry( - data=data, - start_time=start_time, - end_time=end_time, - size_bytes=size_bytes, - ) - - # Segment first. - seg_key = self._segment_key(channel_id, run_id, seg_id) - self._store.put(seg_key, entry, size_bytes=size_bytes) - if seg_key not in self._store: - # The store rejected the segment (oversize, disabled, etc.). - # Skipping the index update keeps us from leaving a - # dangling reference to a never-written segment. - return - - idx.segments.append( - SegmentRef( - seg_id=seg_id, - start_time=start_time, - end_time=end_time, - size_bytes=size_bytes, - ) - ) - idx.next_seg_id += 1 - self._store.put( - self._index_key(channel_id, run_id), - idx, - size_bytes=max(1024, 128 * len(idx.segments)), - ) + if len(data) == 0: + # Empty ref: coverage-only entry, no segment body. Doesn't + # bump ``next_seg_id`` because there's no segment to key. + idx.segments.append(SegmentRef(seg_id=None, start_time=start_time, end_time=end_time)) + self._write_index(channel_id, run_id, idx) + else: + size_bytes = int(data.memory_usage(deep=True).sum()) + seg_id = idx.next_seg_id + + # Segment first. + seg_key = self._segment_key(channel_id, run_id, seg_id) + self._store.put(seg_key, data, size_bytes=size_bytes) + if seg_key not in self._store: + # The store rejected the segment (oversize, disabled, + # etc.). Skipping the index update keeps us from + # leaving a dangling reference to a never-written + # segment. + return + + idx.segments.append(SegmentRef(seg_id=seg_id, start_time=start_time, end_time=end_time)) + idx.next_seg_id += 1 + self._write_index(channel_id, run_id, idx) + + # Bound read fan-out: collapse all segments into one once the + # bucket would carry more than the cap. + if len(idx.segments) > self.MAX_SEGMENTS_PER_BUCKET: + self._compact_bucket(channel_id, run_id) def invalidate(self, channel_id: str, run_id: str | None = None) -> None: """Drop every segment in a bucket plus the index. Safe when absent. Only touches the one ``(channel_id, run_id)`` bucket — segments - under other runs survive. + under other runs survive. Empty refs carry no segment key, so + only data refs (``seg_id is not None``) hit the store here. """ idx = self._load_index(channel_id, run_id) if idx is None: return for ref in idx.segments: + if ref.is_empty: + continue self._store.invalidate(self._segment_key(channel_id, run_id, ref.seg_id)) self._store.invalidate(self._index_key(channel_id, run_id)) @@ -308,19 +315,176 @@ def _load_index(self, channel_id: str, run_id: str | None) -> SegmentIndex | Non raw = self._store.get(self._index_key(channel_id, run_id)) if not isinstance(raw, SegmentIndex): return None - if raw.schema_version != self.SCHEMA_VERSION: - # Future migration: discard incompatible on-disk indexes. - return None return raw def _load_segment( self, channel_id: str, run_id: str | None, seg_id: int - ) -> ChannelCacheEntry | None: + ) -> pd.DataFrame | None: raw = self._store.get(self._segment_key(channel_id, run_id, seg_id)) - if not isinstance(raw, ChannelCacheEntry): + if not isinstance(raw, pd.DataFrame): return None return raw + def _write_index(self, channel_id: str, run_id: str | None, idx: SegmentIndex) -> None: + # Pydantic indexes serialize small (refs are scalars), but we + # still pass a size hint that scales with segment count so the + # store's per-entry-size eviction sees an honest weight. + self._store.put( + self._index_key(channel_id, run_id), + idx, + size_bytes=max(1024, 128 * len(idx.segments)), + ) + + def _compact_bucket(self, channel_id: str, run_id: str | None) -> None: + """Fold every ref in a bucket into a single merged ref. + + Called by :meth:`put_segment` once the per-bucket ref count + crosses :attr:`MAX_SEGMENTS_PER_BUCKET`. The motivation is read + fan-out: without compaction ``get_range`` on a long + incrementally-pulled run would walk one ref per prior fetch, + so the per-read cost climbs linearly in the number of writes. + After compaction the bucket holds at most one data ref (or one + empty ref, if every prior fetch returned no data) until the + next put. + + Empty refs (``seg_id is None``) carry coverage but no body; + their claimed ``[start, end]`` is folded into the merged ref's + claim along with the data refs. If every loadable body went + away (all-empty or every data ref LRU-evicted), the bucket + compacts to a single empty ref covering the union of all + prior claims so the "no data here" coverage is preserved. + + Write ordering (crash safety): + + 1. Write the merged segment under a fresh ``seg_id`` so a + crash here leaves the old segments + old index intact + (the orphan merged seg LRU-evicts). + 2. Rewrite the index to point at *only* the merged seg. A + crash here leaves the old segs as orphans (LRU-evicts). + 3. Delete the old segment keys. Any crash here is harmless + since the index already points elsewhere. + + Cost note: under a flat cap M, total work across N fetches is + O(N^2 / M) — quadratic in fetches with the constant reduced by + the cap. That's the documented first-cut; if compaction shows + up as a hot spot in stats, the next step is LSM-style + geometric levels rather than tuning M. + """ + idx = self._load_index(channel_id, run_id) + if idx is None or len(idx.segments) <= self.MAX_SEGMENTS_PER_BUCKET: + # Bucket already cleaned up (concurrent invalidate, LRU, + # or a parallel compactor) — nothing to do. + return + + refs = sorted(idx.segments, key=lambda r: r.start_time) + + # Empty refs have no body to load — their contribution is + # purely coverage, which we fold into the merged claim below. + # LRU may also have dropped a data ref's body; those refs + # degrade to gaps (we don't merge missing data). + frames: list[pd.DataFrame] = [] + for ref in refs: + if ref.is_empty: + continue + frame = self._load_segment(channel_id, run_id, ref.seg_id) + if frame is not None: + frames.append(frame) + + old_seg_ids = [r.seg_id for r in refs if not r.is_empty] + + # Claimed coverage of the merged ref spans *every* ref's claim + # (data + empty + evicted) — the index already represents "we + # asked the wire about this range" and compaction must preserve + # that, even if the underlying data went away. + merged_start = min(r.start_time for r in refs) + merged_end = max(r.end_time for r in refs) + + if not frames: + # No usable data bodies left — bucket collapses to a single + # empty ref covering the union of all prior claims. Future + # reads in that range are cache hits returning no rows. + self._replace_index_and_drop_segments( + channel_id, + run_id, + new_segments=[ + SegmentRef(seg_id=None, start_time=merged_start, end_time=merged_end) + ], + next_seg_id=idx.next_seg_id, + drop_seg_ids=old_seg_ids, + ) + logger.debug( + "compacted bucket %r:%r from %d refs to 1 empty ref", + run_id or "", + channel_id, + len(refs), + ) + return + + # Mirror :meth:`get_range`'s dedup: later segments win on + # boundary-overlapping timestamps (refs are already sorted by + # start_time above). + merged_data = frames[0] if len(frames) == 1 else pd.concat(frames).groupby(level=0).last() + merged_size = int(merged_data.memory_usage(deep=True).sum()) + new_seg_id = idx.next_seg_id + + # Step 1: write the merged segment first. + new_seg_key = self._segment_key(channel_id, run_id, new_seg_id) + self._store.put(new_seg_key, merged_data, size_bytes=merged_size) + if new_seg_key not in self._store: + # Store rejected (typically oversize). Leave the bucket + # alone so the per-fetch segments remain readable; the + # next ``put_segment`` will retry the cap check. + logger.debug( + "compaction skipped for bucket %r:%r (merged size %d exceeded store cap)", + run_id or "", + channel_id, + merged_size, + ) + return + + # Steps 2 + 3: rewrite the index to point at only the merged + # seg, then drop superseded segment keys. + self._replace_index_and_drop_segments( + channel_id, + run_id, + new_segments=[ + SegmentRef(seg_id=new_seg_id, start_time=merged_start, end_time=merged_end) + ], + next_seg_id=new_seg_id + 1, + drop_seg_ids=old_seg_ids, + ) + logger.debug( + "compacted bucket %r:%r from %d refs to 1 (merged size %d)", + run_id or "", + channel_id, + len(refs), + merged_size, + ) + + def _replace_index_and_drop_segments( + self, + channel_id: str, + run_id: str | None, + *, + new_segments: list[SegmentRef], + next_seg_id: int, + drop_seg_ids: list[int], + ) -> None: + """Rewrite the index, then invalidate superseded segment keys. + + Order matters for crash safety: the new index goes down first + so that an interrupt before the deletes leaves the old segs as + unreachable orphans (which LRU-evict) rather than leaving the + index pointing at deleted keys. + """ + self._write_index( + channel_id, + run_id, + SegmentIndex(next_seg_id=next_seg_id, segments=new_segments), + ) + for seg_id in drop_seg_ids: + self._store.invalidate(self._segment_key(channel_id, run_id, seg_id)) + @staticmethod def _compute_gaps( query_start: datetime, @@ -432,11 +596,12 @@ def _update_cache( self, *, channel_data: dict[str, pd.DataFrame], + fetched_ranges_per_channel: dict[str, list[TimeRange]], start_time: datetime, end_time: datetime, run_id: str | None = None, ): - """Write each channel's fresh data as a new segment. + """Write each channel's fresh data or empty-ref as a new segment. Per-fetch disk write is O(this segment) — no merging with prior segments and no re-pickle of accumulated data, so n sequential @@ -449,10 +614,26 @@ def _update_cache( produces exactly one segment per channel, regardless of how many elements that channel exposes. - Empty data is skipped (no zero-row segments). This means a - no-run query that returns nothing won't record coverage, so the - next identical query will re-fetch — see the ``put_segment`` - TODO for "we tried, no data" segments. + Empty results are recorded as empty refs (no segment body) — + one ref per fetched range — so a repeat of a known-empty range + is a cache hit returning no rows instead of another wire call. + For run-scoped queries we claim only the exact fetched window: + absence outside it isn't assertable (the run may not have + started yet, may have ended early, etc.). + + Args: + channel_data: Merged per-name frames coming out of + :meth:`_merge_pages`. A channel absent from this dict + returned zero rows across every page; a present channel + with a zero-row frame had every page return zero rows + (bitfield elements that all came back empty). + fetched_ranges_per_channel: Per-channel-id list of the + ``[start, end]`` windows we actually asked the wire + about this call. Drives empty-ref recording: any cid + in this dict whose data dropped out for the empty case + gets one empty ref per range. Pure cache hits don't + reach this method (the caller skips when nothing was + fetched). """ assert start_time is not None assert end_time is not None @@ -471,6 +652,7 @@ def _update_cache( ) by_channel_id.setdefault(channel_id, []).append(data) + ids_with_data: set[str] = set() for channel_id, frames in by_channel_id.items(): if len(frames) == 1: combined = frames[0] @@ -480,8 +662,12 @@ def _update_cache( combined = pd.concat(frames).groupby(level=0).last() if len(combined) == 0: + # Bitfield with every element empty, or other edge + # cases — handled by the empty-ref loop below. continue + ids_with_data.add(channel_id) + # Segment coverage range. For run-scoped queries, claim # only what the data actually spans (we can't assert # absence outside the data — the run might not have @@ -504,6 +690,24 @@ def _update_cache( end_time=seg_end, ) + # Empty refs for channels that were queried this call but came + # back with zero rows. One ref per fetched range so the + # coverage claim matches what we actually asked the wire about + # — overclaiming would let a future query for an adjacent + # range hit the cache and miss data that actually exists. + empty_df = pd.DataFrame() + for cid, ranges in fetched_ranges_per_channel.items(): + if cid in ids_with_data: + continue + for fetched_start, fetched_end in ranges: + self.channel_cache.put_segment( + channel_id=cid, + run_id=run_id, + data=empty_df, + start_time=fetched_start, + end_time=fetched_end, + ) + async def get_channel_data( self, *, @@ -525,9 +729,14 @@ async def get_channel_data( # Two work queues. Fully uncached channels share the full range # and get batched; partial-hit channels carry per-gap ranges - # and go one fetch at a time. + # and go one fetch at a time. ``fetched_ranges_per_channel`` + # records the exact ``[s, e]`` windows we asked the wire about, + # per channel — used downstream to record empty refs (a "no + # data here" coverage claim) so a repeat of a known-empty + # query is a cache hit instead of another wire call. fully_uncached: list[str] = [] partial_gaps: list[tuple[str, list[TimeRange]]] = [] + fetched_ranges_per_channel: dict[str, list[TimeRange]] = {} for channel in channels: cid = channel.id_ @@ -549,6 +758,7 @@ async def get_channel_data( if not gaps: continue + fetched_ranges_per_channel.setdefault(cid, []).extend(gaps) if len(gaps) == 1 and gaps[0] == (start_time, end_time): fully_uncached.append(cid) else: @@ -596,13 +806,15 @@ async def get_channel_data( pages = await asyncio.gather(*tasks) ret_data = self._merge_pages(pages, initial=ret_data) - # Skip the cache update when no fresh pages arrived (pure cache - # hit). Avoids the redundant "rewrite same bytes" case that - # Tier 1A also targeted under the old single-entry shape. - had_fresh_data = any(pages) - if not ignore_cache and had_fresh_data: + # Pure cache hits never reach ``_update_cache`` because nothing + # was fetched (``fetched_ranges_per_channel`` is empty). When + # we did fetch, ``_update_cache`` writes both data segments and + # empty refs — the latter covers the "asked the wire, got + # nothing" case so a repeat doesn't refetch. + if not ignore_cache and fetched_ranges_per_channel: self._update_cache( channel_data=ret_data, + fetched_ranges_per_channel=fetched_ranges_per_channel, start_time=start_time, end_time=end_time, run_id=run_id, diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py index 90b12be32..22386944f 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py @@ -45,10 +45,8 @@ from sift_client._internal.disk_cache import DiskCache from sift_client._internal.low_level_wrappers.data import ( - ChannelCacheEntry, ChannelDataCache, DataLowLevelClient, - _new_cache_entry, ) from sift_client.sift_types.channel import ( Channel, @@ -80,16 +78,6 @@ def _frame( ).astype({cid: value_dtype}) -def _entry(*, rows: int = 5, value_dtype: str = "float64") -> ChannelCacheEntry: - """``ChannelCacheEntry`` wrapping a small generated DataFrame.""" - data = _frame(rows=rows, value_dtype=value_dtype) - return _new_cache_entry( - data=data, - start_time=data.index[0].to_pydatetime(), - end_time=data.index[-1].to_pydatetime(), - ) - - def _channel(cid: str) -> Channel: """Minimal ``Channel`` with required fields populated.""" return Channel( @@ -263,7 +251,7 @@ class TestChannelDataCache: Five invariants get pinned across the per-segment shape: 1. Every operation routes through the namespaced key - (``channel:::{idx,seg:N}``), so two adapters sharing + (``channel:v2:::{idx,seg:N}``), so two adapters sharing one store don't collide on bare resource ids. 2. Run id is part of the cache dimension: the same ``channel_id`` under two different runs is two cache buckets, not one. @@ -303,20 +291,23 @@ def test_round_trip_single_segment(self, tmp_path): adapter.store.close() def test_writes_use_namespaced_index_and_segment_keys(self, tmp_path): - """The raw store sees ``channel:::idx`` + ``...:seg:0``. + """The raw store sees ``channel:v2:::idx`` + ``...:seg:0``. Pins the per-segment key shape: one index plus one segment key per fetch. Without the prefix, a second adapter that happens to share an id with the channel adapter would clobber the rows. + The ``v1`` is the schema version baked into the prefix so a + bump silently retires the entire old keyspace. """ store = DiskCache(disk_path=tmp_path / "ns") adapter = ChannelDataCache(store) try: _put(adapter, "c1", rows=4) - assert "channel::c1:idx" in store - assert "channel::c1:seg:0" in store + assert "channel:v2::c1:idx" in store + assert "channel:v2::c1:seg:0" in store assert "c1" not in store assert "channel:c1" not in store # never the bare-id shape + assert "channel::c1:idx" not in store # never the unversioned shape finally: store.close() @@ -374,7 +365,7 @@ def test_get_range_isinstance_filters_foreign_segments(self, tmp_path): # Overwrite the segment's payload with foreign data; the # index still claims it exists, so the read should treat # the segment range as a gap. - store.put("channel::c1:seg:0", {"not": "an entry"}, size_bytes=64) + store.put("channel:v2::c1:seg:0", {"not": "an entry"}, size_bytes=64) data, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END) assert data is None # The whole query range is uncovered (one merged gap). @@ -535,7 +526,7 @@ def test_evicted_segment_degrades_to_gap(self, tmp_path): adapter = ChannelDataCache(store) try: df = _put(adapter, "c1", rows=5) - store.invalidate("channel::c1:seg:0") # simulate eviction + store.invalidate("channel:v2::c1:seg:0") # simulate eviction data, gaps = adapter.get_range("c1", None, df.index[0], df.index[-1]) assert data is None assert gaps == [(df.index[0], df.index[-1])] @@ -563,28 +554,85 @@ def test_segment_count_grows_with_each_put(self, tmp_path): # Expect: 1 index + 3 segments. channel_keys = sorted(k for k in store if k.startswith("channel:")) assert channel_keys == [ - "channel::c1:idx", - "channel::c1:seg:0", - "channel::c1:seg:1", - "channel::c1:seg:2", + "channel:v2::c1:idx", + "channel:v2::c1:seg:0", + "channel:v2::c1:seg:1", + "channel:v2::c1:seg:2", ] finally: store.close() - def test_empty_put_is_skipped(self, tmp_path): - """``put_segment`` with an empty frame is a no-op (no index update). + def test_empty_put_records_empty_ref_and_skips_refetch(self, tmp_path): + """``put_segment`` with empty data records a ``seg_id=None`` ref. + + Empty ref behavior: - Documented trade-off: no "we tried, no data" segments. A - follow-up identical query will re-fetch. See class TODO. + * ``has_any`` returns True (the bucket has a coverage claim, + just no body). + * ``get_range`` over the claimed range returns ``(None, [])`` + — no data, no gaps, so the caller doesn't refetch. + * No segment key lands in the store (only the index gets a + row); the empty ref lives entirely on the index. """ - adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "empty")) + store = DiskCache(disk_path=tmp_path / "empty") + adapter = ChannelDataCache(store) try: empty = pd.DataFrame( {"c1": []}, index=pd.DatetimeIndex([], tz=timezone.utc), ) adapter.put_segment("c1", None, empty, _NOW, _WINDOW_END) - assert not adapter.has_any("c1") + + assert adapter.has_any("c1") + data, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END) + assert data is None + assert gaps == [] + + channel_keys = sorted(k for k in store if k.startswith("channel:")) + assert channel_keys == ["channel:v2::c1:idx"] + finally: + store.close() + + def test_empty_ref_only_covers_its_claimed_range(self, tmp_path): + """Reads outside the empty ref's claimed range still report gaps. + + Pins the "claim only what you queried" contract: an empty ref + covering ``[T0, T5]`` doesn't suppress a refetch for + ``[T5, T10]``. + """ + adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "narrow")) + try: + claim_end = _NOW + timedelta(seconds=5) + adapter.put_segment("c1", None, pd.DataFrame(), _NOW, claim_end) + + _, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END) + # Empty ref covers [_NOW, claim_end]; remainder is a gap. + assert gaps == [(claim_end, _WINDOW_END)] + finally: + adapter.store.close() + + def test_empty_ref_alongside_data_segment(self, tmp_path): + """Stitching data + empty ref returns the data and reports no gap. + + Mirrors a real partial-hit scenario: cache already has data + for ``[T0, T5]``, the caller queries ``[T0, T10]``, the gap + ``[T5, T10]`` fetch returns no rows and lands as an empty + ref. Subsequent ``get_range(T0, T10)`` returns the cached + data with ``gaps == []``. + """ + adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "mix")) + try: + data = _frame("c1", rows=5, freq="s") + mid = data.index[-1].to_pydatetime() + tail = mid + timedelta(seconds=5) + + adapter.put_segment("c1", None, data, data.index[0].to_pydatetime(), mid) + adapter.put_segment("c1", None, pd.DataFrame(), mid, tail) + + got, gaps = adapter.get_range("c1", None, data.index[0].to_pydatetime(), tail) + assert got is not None + pd.testing.assert_frame_equal(got, data) + assert gaps == [] finally: adapter.store.close() @@ -639,6 +687,226 @@ def test_compute_gaps(self, covered, expected): assert ChannelDataCache._compute_gaps(query_start, query_end, covered) == expected +class TestCompaction: + """``ChannelDataCache.MAX_SEGMENTS_PER_BUCKET`` triggers a merge. + + The cap bounds read fan-out: without it, every incremental + ``put_segment`` would add another segment that ``get_range`` would + have to load. Compaction folds the bucket into a single segment + once the count crosses the cap, so the next ``get_range`` only + walks one segment. + + The interesting invariants: + + 1. **Below cap, no merge.** Compaction is a write-time side effect, + not a constant background tax; under the cap the bucket layout + is unchanged. + 2. **At cap+1, fires inline.** ``put_segment`` returns with the + bucket already collapsed — the next reader doesn't pay for a + fragmented bucket. + 3. **Data preserved through merge.** Concrete row count and column + contents survive; LRU-evicted bodies degrade to absence (not + errors) but their refs are still dropped from the index so the + index never points at nothing. + 4. **Old segment keys are deleted.** Otherwise the store carries + the merged seg + every superseded seg until LRU catches up, + which would defeat the bucket's byte-budget accounting. + + A ``monkeypatch`` lowers the cap so tests stay fast — the prod + constant is large enough that a literal 17-fetch test would dwarf + the rest of the suite. + """ + + def test_below_cap_no_compaction(self, tmp_path, monkeypatch): + """Three segments under a cap of four stay separate.""" + monkeypatch.setattr(ChannelDataCache, "MAX_SEGMENTS_PER_BUCKET", 4) + store = DiskCache(disk_path=tmp_path / "below") + adapter = ChannelDataCache(store) + try: + for i in range(3): + _put(adapter, "c1", rows=2, start=_NOW + timedelta(seconds=i * 10)) + + channel_keys = sorted(k for k in store if k.startswith("channel:")) + assert channel_keys == [ + "channel:v2::c1:idx", + "channel:v2::c1:seg:0", + "channel:v2::c1:seg:1", + "channel:v2::c1:seg:2", + ] + finally: + store.close() + + def test_crossing_cap_collapses_to_single_segment(self, tmp_path, monkeypatch): + """One write past the cap triggers an inline compaction. + + Pins all three observable effects in one place: + * Index ends up with exactly one ref. + * The merged segment's ``seg_id`` is fresh (no overlap with + any pre-compaction id). + * Old segment keys are gone from the raw store. + """ + monkeypatch.setattr(ChannelDataCache, "MAX_SEGMENTS_PER_BUCKET", 3) + store = DiskCache(disk_path=tmp_path / "cross") + adapter = ChannelDataCache(store) + try: + for i in range(4): + _put(adapter, "c1", rows=2, start=_NOW + timedelta(seconds=i * 10)) + + idx = adapter._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 1 + merged_id = idx.segments[0].seg_id + assert merged_id >= 4 + + channel_keys = sorted(k for k in store if k.startswith("channel:")) + assert channel_keys == [ + "channel:v2::c1:idx", + f"channel:v2::c1:seg:{merged_id}", + ] + finally: + store.close() + + def test_compaction_preserves_data(self, tmp_path, monkeypatch): + """``get_range`` after compaction returns the same rows as before. + + Builds disjoint, time-sorted frames so the dedup path is a + no-op; correctness of the dedup itself is covered by the + overlapping-timestamps test below. + """ + monkeypatch.setattr(ChannelDataCache, "MAX_SEGMENTS_PER_BUCKET", 2) + store = DiskCache(disk_path=tmp_path / "preserve") + adapter = ChannelDataCache(store) + try: + frames = [] + for i in range(3): + frame = _put(adapter, "c1", rows=2, start=_NOW + timedelta(seconds=i * 10)) + frames.append(frame) + + idx = adapter._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 1 + + expected = pd.concat(frames) + got, gaps = adapter.get_range( + "c1", + None, + frames[0].index[0].to_pydatetime(), + frames[-1].index[-1].to_pydatetime(), + ) + assert gaps == [] + assert got is not None + pd.testing.assert_frame_equal(got, expected) + finally: + store.close() + + def test_compaction_dedups_overlapping_timestamps(self, tmp_path, monkeypatch): + """When two segments share a timestamp, the later put wins. + + Matches :meth:`get_range`'s ``groupby(level=0).last()`` dedup, + so a refetch that overwrites a stale cached value still ends + up with the fresh value after compaction. + """ + monkeypatch.setattr(ChannelDataCache, "MAX_SEGMENTS_PER_BUCKET", 1) + store = DiskCache(disk_path=tmp_path / "dedup") + adapter = ChannelDataCache(store) + try: + ts = _NOW + stale = pd.DataFrame( + {"c1": [1.0]}, + index=pd.DatetimeIndex([ts], tz=timezone.utc), + ) + fresh = pd.DataFrame( + {"c1": [2.0]}, + index=pd.DatetimeIndex([ts], tz=timezone.utc), + ) + adapter.put_segment("c1", None, stale, ts, ts) + adapter.put_segment("c1", None, fresh, ts, ts) + + idx = adapter._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 1 + + got, _ = adapter.get_range("c1", None, ts, ts) + assert got is not None + assert got.iloc[0, 0] == 2.0 + finally: + store.close() + + def test_empty_refs_fold_into_merged_claim(self, tmp_path, monkeypatch): + """Empty refs alongside data refs collapse into one data ref. + + After compaction the bucket carries a single data segment + whose claimed range spans all prior refs (data + empty). The + empty refs are dropped from the index, their coverage now + represented by the merged ref's wider claim. + """ + monkeypatch.setattr(ChannelDataCache, "MAX_SEGMENTS_PER_BUCKET", 2) + store = DiskCache(disk_path=tmp_path / "fold_empty") + adapter = ChannelDataCache(store) + try: + data = _put(adapter, "c1", rows=4, start=_NOW, freq="s") + data_end = data.index[-1].to_pydatetime() + empty_end = data_end + timedelta(seconds=10) + # Second put (empty) crosses the cap of 2 → compacts. + adapter.put_segment("c1", None, pd.DataFrame(), data_end, empty_end) + adapter.put_segment( + "c1", None, pd.DataFrame(), empty_end, empty_end + timedelta(seconds=10) + ) + + idx = adapter._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 1 + sole = idx.segments[0] + assert sole.seg_id is not None # data ref survives + assert sole.start_time == data.index[0].to_pydatetime() + assert sole.end_time == empty_end + timedelta(seconds=10) + + # Query the wider claim — data slice returns the actual + # rows, and the empty tail counts as covered (no gap). + got, gaps = adapter.get_range("c1", None, data.index[0].to_pydatetime(), sole.end_time) + assert got is not None + pd.testing.assert_frame_equal(got, data) + assert gaps == [] + finally: + store.close() + + def test_all_empty_bucket_collapses_to_single_empty_ref(self, tmp_path, monkeypatch): + """Compacting a bucket with only empty refs leaves one empty ref. + + The merged claim is the union of every prior empty ref's + claim, so the "no data anywhere in this window" coverage is + preserved while the per-fetch refs collapse. + """ + monkeypatch.setattr(ChannelDataCache, "MAX_SEGMENTS_PER_BUCKET", 2) + store = DiskCache(disk_path=tmp_path / "all_empty") + adapter = ChannelDataCache(store) + try: + t0 = _NOW + t1 = _NOW + timedelta(seconds=10) + t2 = _NOW + timedelta(seconds=20) + t3 = _NOW + timedelta(seconds=30) + for s, e in [(t0, t1), (t1, t2), (t2, t3)]: + adapter.put_segment("c1", None, pd.DataFrame(), s, e) + + idx = adapter._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 1 + sole = idx.segments[0] + assert sole.seg_id is None + assert sole.start_time == t0 + assert sole.end_time == t3 + + data, gaps = adapter.get_range("c1", None, t0, t3) + assert data is None + assert gaps == [] + + # No segment keys at all — index-only bucket. + seg_keys = [k for k in store if k.startswith("channel:") and ":seg:" in k] + assert seg_keys == [] + finally: + store.close() + + class TestMergePages: """Behaviour of :meth:`DataLowLevelClient._merge_pages`. @@ -1021,6 +1289,50 @@ async def test_disjoint_forward_paging_lands_in_separate_segments(self, tmp_path finally: client.channel_cache.store.close() + @pytest.mark.asyncio + async def test_empty_wire_response_records_empty_ref_and_skips_refetch(self, tmp_path) -> None: + """A query that returns no rows caches an empty ref; the repeat hits cache. + + End-to-end version of the empty-ref contract: a wire call that + returns zero rows lands an empty :class:`SegmentRef` in the + bucket so a repeat of the same query is a full cache hit (no + wire call, no rows). Pins the "known-empty range doesn't + refetch" behavior the segment-cache write amplification fix + depended on for correctness. + """ + client = _client_with_cache(tmp_path) + try: + # First call: zero pages → empty wire response. + with _fake_grpc(client, {"c1": []}) as log1: + result1 = await client.get_channel_data( + channels=[_channel("c1")], + start_time=_NOW, + end_time=_WINDOW_END, + ) + assert log1, "first call should hit the wire" + assert "c1" not in result1 + + # Bucket carries an empty ref for the queried window. + assert client.channel_cache.has_any("c1") + idx = client.channel_cache._load_index("c1", None) + assert idx is not None + assert len(idx.segments) == 1 + assert idx.segments[0].seg_id is None + assert idx.segments[0].start_time == _NOW + assert idx.segments[0].end_time == _WINDOW_END + + # Second identical call: cache hit, no wire traffic. + with _fake_grpc(client, {"c1": []}) as log2: + result2 = await client.get_channel_data( + channels=[_channel("c1")], + start_time=_NOW, + end_time=_WINDOW_END, + ) + assert log2 == [], "second call should be a full cache hit" + assert "c1" not in result2 + finally: + client.channel_cache.store.close() + @pytest.mark.asyncio async def test_ignore_cache_true_returns_fresh_and_skips_write(self, tmp_path) -> None: """``ignore_cache=True`` returns mock data and leaves the cache empty. @@ -1102,6 +1414,7 @@ def test_update_cache_merges_elements_into_one_segment(self, tmp_path) -> None: try: client._update_cache( channel_data={"ch1.lo": df_lo, "ch1.hi": df_hi}, + fetched_ranges_per_channel={"abc": [(_NOW, _WINDOW_END)]}, start_time=_NOW, end_time=_WINDOW_END, ) @@ -1140,6 +1453,7 @@ def test_get_range_treats_bitfield_as_one_bucket(self, tmp_path) -> None: "ch1.lo": _frame("ch1.lo", rows=5), "ch1.hi": _frame("ch1.hi", rows=5), }, + fetched_ranges_per_channel={"abc": [(_NOW, _WINDOW_END)]}, start_time=_NOW, end_time=_WINDOW_END, ) From aa2cdcd701b4e354609cc3d50f2c9cd0b20c0558 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 30 Jun 2026 13:38:02 -0700 Subject: [PATCH 3/7] commit moved file --- python/lib/sift_client/cache.py | 224 ++++++++++++++++++++++++++++++++ 1 file changed, 224 insertions(+) create mode 100644 python/lib/sift_client/cache.py diff --git a/python/lib/sift_client/cache.py b/python/lib/sift_client/cache.py new file mode 100644 index 000000000..abab302c3 --- /dev/null +++ b/python/lib/sift_client/cache.py @@ -0,0 +1,224 @@ +"""User-facing surface for the shared on-disk cache. + +This module hosts the small bag of methods exposed as ``client.cache``, +plus the :class:`CacheStats` snapshot type returned by +:meth:`CacheNamespace.stats`. Both classes are public — they're imported +directly off ``sift_client`` and surfaced in the generated API docs. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from sift_client._internal.disk_cache import DiskCache + +if TYPE_CHECKING: + import os + + from sift_client.client import SiftClient + +logger = logging.getLogger(__name__) + + +_BYTE_UNITS = ( + (1024**4, "TiB"), + (1024**3, "GiB"), + (1024**2, "MiB"), + (1024, "KiB"), +) + + +def _format_bytes(n: int) -> str: + """Render ``n`` bytes in the largest unit that doesn't underflow to zero.""" + for threshold, suffix in _BYTE_UNITS: + if n >= threshold: + return f"{n / threshold:.1f} {suffix}" + return f"{n} B" + + +@dataclass(frozen=True) +class CacheStats: + """Snapshot of the shared on-disk cache at call time. + + Returned by :meth:`CacheNamespace.stats`. Frozen dataclass so it + plays well with logging, snapshot tests, and "compare two readings" + diagnostics without surprise mutation. + + Field semantics: + + * **enabled** — whether the disk handle is open. When ``False``, all + the size/count fields are zero regardless of on-disk state. + * **path** — directory the cache is open against, or ``None`` when + disabled. Useful for "where does this cache actually live?". + * **max_bytes** — configured byte cap on disk usage, or ``None`` + when disabled. ``diskcache``'s LRU evicts once usage approaches + this. + * **size_bytes** — current on-disk usage including SQLite overhead. + Tends to trend slightly higher than the sum of per-entry + ``size_bytes`` the resources hand to the store. + * **entry_count** — total cache keys across all adapter prefixes + (channel entries + any future foreign-adapter rows). + * **channel_entries** — channel cache entries (one per + ``(channel_id, run_id)`` bucket under the current single-entry + shape). Counted by walking the channel adapter's namespace + prefix. + + ``str(stats)`` prints a multi-line summary suitable for + notebook/REPL display; the structured fields are available for + programmatic checks. + """ + + enabled: bool + path: str | None + max_bytes: int | None + size_bytes: int + entry_count: int + channel_entries: int + + def __str__(self) -> str: + if not self.enabled: + return "Sift cache: disabled" + cap = _format_bytes(self.max_bytes) if self.max_bytes is not None else "no cap" + pct = f" ({self.size_bytes / self.max_bytes * 100:.1f}%)" if self.max_bytes else "" + return ( + "Sift cache:\n" + f" path: {self.path}\n" + f" used: {_format_bytes(self.size_bytes)} / {cap}{pct}\n" + f" entries: {self.entry_count} ({self.channel_entries} channel)" + ) + + +class CacheNamespace: + """Resource-agnostic surface for the on-disk cache shared by all resources. + + Exposed as ``client.cache``. The actual handle + (:class:`~sift_client._internal.disk_cache.DiskCache`) is constructed + lazily on first use so importing :mod:`sift_client` doesn't pay the + diskcache cost up front. Configuration changes made before that + first use are recorded against the client's + :class:`~sift_client._internal.disk_cache_config.DiskCacheConfig` + and applied when the store opens; changes after first use are + routed directly to the live store. + + Default policy: disk caching is **opt-out** (the config is + constructed with ``enabled=True``). Users who don't want any state + on disk call :meth:`disable` to silence it; users who want a custom + location or byte cap call :meth:`enable` with arguments. + """ + + def __init__(self, client: SiftClient): + """Bind this namespace to ``client``. Constructed by :class:`SiftClient`.""" + self._client = client + + def enable( + self, + *, + path: str | os.PathLike[str] | None = None, + max_bytes: int | None = None, + ) -> None: + """Enable (or reconfigure) on-disk caching. + + Disk caching is **on by default** at + :attr:`~sift_client._internal.disk_cache.DiskCache.DEFAULT_DISK_PATH`; + use this method to override the path or size, or to turn the + cache back on after a prior :meth:`disable` call. + + Reconfiguring a live cache (different ``path`` or ``max_bytes``) + closes the previous handle and opens a new one. Existing entries + at the new path become available as cache hits. + + An explicit ``path`` that can't be opened (permission denied, + read-only filesystem, ...) raises so the caller knows their + request didn't take. The default-path open does *not* raise — + see :meth:`SiftClient._get_disk_cache` for the silent fall-back. + + Args: + path: Directory to persist to. ``None`` (the default) uses + the cache's :attr:`DEFAULT_DISK_PATH`. + max_bytes: Byte cap on disk usage. ``None`` uses the cache's + :attr:`DEFAULT_DISK_MAX_BYTES` (4 GiB). When the bound + is reached, ``diskcache``'s LRU eviction takes over. + + Example: + client.cache.enable(path="/data/sift-cache") + client.cache.enable(max_bytes=1024 ** 3) # 1 GiB + """ + client = self._client + client._disk_cache_config.enable(path=path, max_bytes=max_bytes) + if client._disk_cache is not None: + client._disk_cache.enable(path=path, max_bytes=max_bytes) + + def disable(self) -> None: + """Opt out of on-disk caching (no reads or writes). + + Caching is on by default; call this when you don't want any + cached data written to or read from disk. Closes any open cache + file handle. The on-disk directory is NOT deleted — use + :meth:`clear` to wipe it. + """ + client = self._client + client._disk_cache_config.disable() + if client._disk_cache is not None: + client._disk_cache.disable() + + def stats(self) -> CacheStats: + """Return a snapshot of the current cache state. + + Example: + >>> print(client.cache.stats()) + Sift cache: + path: /tmp/sift-data-cache + used: 142.3 MiB / 4.0 GiB (3.5%) + entries: 487 (487 channel) + """ + # Importing here keeps this module light at import time — pulling + # the adapter pulls pandas, which is the whole reason + # ``_ensure_data_low_level_client`` is lazy too. + from sift_client._internal.low_level_wrappers.data import ChannelDataCache + + store = self._client._get_disk_cache() + if not store.disk_enabled: + return CacheStats( + enabled=False, + path=None, + max_bytes=None, + size_bytes=0, + entry_count=0, + channel_entries=0, + ) + + # One pass over the keyspace. Cheap — diskcache keys are SQLite + # rows, and we only touch metadata (no value loads). + channel_entries = sum(1 for key in store if key.startswith(ChannelDataCache.KEY_PREFIX)) + return CacheStats( + enabled=True, + path=store.disk_path, + max_bytes=store.disk_max_bytes, + size_bytes=store.volume(), + entry_count=len(store), + channel_entries=channel_entries, + ) + + def clear(self, path: str | os.PathLike[str] | None = None) -> None: + """Delete a previously-persisted on-disk cache directory. + + Drops stale caches from previous sessions, recovers from a + corrupt cache, or reclaims disk space. Removes the directory + entirely; if disk caching is on, the next access re-opens an + empty cache at the same path. + + Args: + path: Directory of the cache to clear. ``None`` (the default) + targets the cache's + :attr:`~sift_client._internal.disk_cache.DiskCache.DEFAULT_DISK_PATH`. + + Raises: + ValueError: If ``path`` exists but does not look like a sift + data cache directory. + """ + DiskCache.clear_disk(path) + + +__all__ = ["CacheNamespace", "CacheStats"] From 888f636162682eaf9929bdb9a8524d25fa3ba0e3 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 30 Jun 2026 13:40:17 -0700 Subject: [PATCH 4/7] rm old cache --- python/lib/sift_client/cache.py | 224 -------------------------------- 1 file changed, 224 deletions(-) delete mode 100644 python/lib/sift_client/cache.py diff --git a/python/lib/sift_client/cache.py b/python/lib/sift_client/cache.py deleted file mode 100644 index abab302c3..000000000 --- a/python/lib/sift_client/cache.py +++ /dev/null @@ -1,224 +0,0 @@ -"""User-facing surface for the shared on-disk cache. - -This module hosts the small bag of methods exposed as ``client.cache``, -plus the :class:`CacheStats` snapshot type returned by -:meth:`CacheNamespace.stats`. Both classes are public — they're imported -directly off ``sift_client`` and surfaced in the generated API docs. -""" - -from __future__ import annotations - -import logging -from dataclasses import dataclass -from typing import TYPE_CHECKING - -from sift_client._internal.disk_cache import DiskCache - -if TYPE_CHECKING: - import os - - from sift_client.client import SiftClient - -logger = logging.getLogger(__name__) - - -_BYTE_UNITS = ( - (1024**4, "TiB"), - (1024**3, "GiB"), - (1024**2, "MiB"), - (1024, "KiB"), -) - - -def _format_bytes(n: int) -> str: - """Render ``n`` bytes in the largest unit that doesn't underflow to zero.""" - for threshold, suffix in _BYTE_UNITS: - if n >= threshold: - return f"{n / threshold:.1f} {suffix}" - return f"{n} B" - - -@dataclass(frozen=True) -class CacheStats: - """Snapshot of the shared on-disk cache at call time. - - Returned by :meth:`CacheNamespace.stats`. Frozen dataclass so it - plays well with logging, snapshot tests, and "compare two readings" - diagnostics without surprise mutation. - - Field semantics: - - * **enabled** — whether the disk handle is open. When ``False``, all - the size/count fields are zero regardless of on-disk state. - * **path** — directory the cache is open against, or ``None`` when - disabled. Useful for "where does this cache actually live?". - * **max_bytes** — configured byte cap on disk usage, or ``None`` - when disabled. ``diskcache``'s LRU evicts once usage approaches - this. - * **size_bytes** — current on-disk usage including SQLite overhead. - Tends to trend slightly higher than the sum of per-entry - ``size_bytes`` the resources hand to the store. - * **entry_count** — total cache keys across all adapter prefixes - (channel entries + any future foreign-adapter rows). - * **channel_entries** — channel cache entries (one per - ``(channel_id, run_id)`` bucket under the current single-entry - shape). Counted by walking the channel adapter's namespace - prefix. - - ``str(stats)`` prints a multi-line summary suitable for - notebook/REPL display; the structured fields are available for - programmatic checks. - """ - - enabled: bool - path: str | None - max_bytes: int | None - size_bytes: int - entry_count: int - channel_entries: int - - def __str__(self) -> str: - if not self.enabled: - return "Sift cache: disabled" - cap = _format_bytes(self.max_bytes) if self.max_bytes is not None else "no cap" - pct = f" ({self.size_bytes / self.max_bytes * 100:.1f}%)" if self.max_bytes else "" - return ( - "Sift cache:\n" - f" path: {self.path}\n" - f" used: {_format_bytes(self.size_bytes)} / {cap}{pct}\n" - f" entries: {self.entry_count} ({self.channel_entries} channel)" - ) - - -class CacheNamespace: - """Resource-agnostic surface for the on-disk cache shared by all resources. - - Exposed as ``client.cache``. The actual handle - (:class:`~sift_client._internal.disk_cache.DiskCache`) is constructed - lazily on first use so importing :mod:`sift_client` doesn't pay the - diskcache cost up front. Configuration changes made before that - first use are recorded against the client's - :class:`~sift_client._internal.disk_cache_config.DiskCacheConfig` - and applied when the store opens; changes after first use are - routed directly to the live store. - - Default policy: disk caching is **opt-out** (the config is - constructed with ``enabled=True``). Users who don't want any state - on disk call :meth:`disable` to silence it; users who want a custom - location or byte cap call :meth:`enable` with arguments. - """ - - def __init__(self, client: SiftClient): - """Bind this namespace to ``client``. Constructed by :class:`SiftClient`.""" - self._client = client - - def enable( - self, - *, - path: str | os.PathLike[str] | None = None, - max_bytes: int | None = None, - ) -> None: - """Enable (or reconfigure) on-disk caching. - - Disk caching is **on by default** at - :attr:`~sift_client._internal.disk_cache.DiskCache.DEFAULT_DISK_PATH`; - use this method to override the path or size, or to turn the - cache back on after a prior :meth:`disable` call. - - Reconfiguring a live cache (different ``path`` or ``max_bytes``) - closes the previous handle and opens a new one. Existing entries - at the new path become available as cache hits. - - An explicit ``path`` that can't be opened (permission denied, - read-only filesystem, ...) raises so the caller knows their - request didn't take. The default-path open does *not* raise — - see :meth:`SiftClient._get_disk_cache` for the silent fall-back. - - Args: - path: Directory to persist to. ``None`` (the default) uses - the cache's :attr:`DEFAULT_DISK_PATH`. - max_bytes: Byte cap on disk usage. ``None`` uses the cache's - :attr:`DEFAULT_DISK_MAX_BYTES` (4 GiB). When the bound - is reached, ``diskcache``'s LRU eviction takes over. - - Example: - client.cache.enable(path="/data/sift-cache") - client.cache.enable(max_bytes=1024 ** 3) # 1 GiB - """ - client = self._client - client._disk_cache_config.enable(path=path, max_bytes=max_bytes) - if client._disk_cache is not None: - client._disk_cache.enable(path=path, max_bytes=max_bytes) - - def disable(self) -> None: - """Opt out of on-disk caching (no reads or writes). - - Caching is on by default; call this when you don't want any - cached data written to or read from disk. Closes any open cache - file handle. The on-disk directory is NOT deleted — use - :meth:`clear` to wipe it. - """ - client = self._client - client._disk_cache_config.disable() - if client._disk_cache is not None: - client._disk_cache.disable() - - def stats(self) -> CacheStats: - """Return a snapshot of the current cache state. - - Example: - >>> print(client.cache.stats()) - Sift cache: - path: /tmp/sift-data-cache - used: 142.3 MiB / 4.0 GiB (3.5%) - entries: 487 (487 channel) - """ - # Importing here keeps this module light at import time — pulling - # the adapter pulls pandas, which is the whole reason - # ``_ensure_data_low_level_client`` is lazy too. - from sift_client._internal.low_level_wrappers.data import ChannelDataCache - - store = self._client._get_disk_cache() - if not store.disk_enabled: - return CacheStats( - enabled=False, - path=None, - max_bytes=None, - size_bytes=0, - entry_count=0, - channel_entries=0, - ) - - # One pass over the keyspace. Cheap — diskcache keys are SQLite - # rows, and we only touch metadata (no value loads). - channel_entries = sum(1 for key in store if key.startswith(ChannelDataCache.KEY_PREFIX)) - return CacheStats( - enabled=True, - path=store.disk_path, - max_bytes=store.disk_max_bytes, - size_bytes=store.volume(), - entry_count=len(store), - channel_entries=channel_entries, - ) - - def clear(self, path: str | os.PathLike[str] | None = None) -> None: - """Delete a previously-persisted on-disk cache directory. - - Drops stale caches from previous sessions, recovers from a - corrupt cache, or reclaims disk space. Removes the directory - entirely; if disk caching is on, the next access re-opens an - empty cache at the same path. - - Args: - path: Directory of the cache to clear. ``None`` (the default) - targets the cache's - :attr:`~sift_client._internal.disk_cache.DiskCache.DEFAULT_DISK_PATH`. - - Raises: - ValueError: If ``path`` exists but does not look like a sift - data cache directory. - """ - DiskCache.clear_disk(path) - - -__all__ = ["CacheNamespace", "CacheStats"] From b4917b6e690189fb10938fa9228ec848865e9f19 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 30 Jun 2026 14:10:04 -0700 Subject: [PATCH 5/7] fix(python): satisfy mypy on segment-cache empty-ref narrowing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dropped the `SegmentRef.is_empty` property in favor of direct `seg_id is None` checks at call sites — the property returned `bool` but didn't narrow `seg_id` from `int | None` to `int`, leaving `_load_segment`, `_segment_key`, and `_replace_index_and_drop_segments` arg-type errors. Same fix for the compaction list comprehension via explicit `list[int]` annotation. Also cast `combined.index[0]` to `pd.Timestamp` in `_update_cache` (and in the test helper) so pandas-stubs' wider `Scalar` union doesn't trip the `.to_pydatetime()` call — the index is always a `DatetimeIndex` at runtime. Co-authored-by: Cursor --- .../_internal/low_level_wrappers/data.py | 24 ++++++++----------- .../_internal/low_level_wrappers/test_data.py | 6 ++--- 2 files changed, 13 insertions(+), 17 deletions(-) diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data.py b/python/lib/sift_client/_internal/low_level_wrappers/data.py index 5a4919f10..8e4f28008 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data.py @@ -53,11 +53,6 @@ class SegmentRef(BaseModel): start_time: datetime end_time: datetime - @property - def is_empty(self) -> bool: - """True for an empty ref (no backing segment body).""" - return self.seg_id is None - class SegmentIndex(BaseModel): """Per-(channel, run) index of cached segments.""" @@ -195,7 +190,7 @@ def get_range( clamped = (max(ref.start_time, start_time), min(ref.end_time, end_time)) - if ref.is_empty: + if ref.seg_id is None: # Empty ref: the wire was asked about this range and # returned no data. Count it toward coverage so the # caller doesn't refetch. @@ -298,7 +293,7 @@ def invalidate(self, channel_id: str, run_id: str | None = None) -> None: if idx is None: return for ref in idx.segments: - if ref.is_empty: + if ref.seg_id is None: continue self._store.invalidate(self._segment_key(channel_id, run_id, ref.seg_id)) self._store.invalidate(self._index_key(channel_id, run_id)) @@ -384,13 +379,13 @@ def _compact_bucket(self, channel_id: str, run_id: str | None) -> None: # degrade to gaps (we don't merge missing data). frames: list[pd.DataFrame] = [] for ref in refs: - if ref.is_empty: + if ref.seg_id is None: continue frame = self._load_segment(channel_id, run_id, ref.seg_id) if frame is not None: frames.append(frame) - old_seg_ids = [r.seg_id for r in refs if not r.is_empty] + old_seg_ids: list[int] = [r.seg_id for r in refs if r.seg_id is not None] # Claimed coverage of the merged ref spans *every* ref's claim # (data + empty + evicted) — the index already represents "we @@ -673,14 +668,15 @@ def _update_cache( # absence outside the data — the run might not have # started yet). For unscoped queries, claim the full # requested range so a follow-up of the same range hits. + seg_end = end_time if run_id: - seg_start = combined.index[0] - if not isinstance(seg_start, datetime): - seg_start = seg_start.to_pydatetime() - seg_end = end_time + # ``combined.index`` is a ``DatetimeIndex`` (built from + # the wire's nanosecond timestamps), so ``index[0]`` is + # always ``pd.Timestamp`` at runtime; pandas-stubs types + # it as the wider ``Scalar`` union. + seg_start = cast(pd.Timestamp, combined.index[0]).to_pydatetime() else: seg_start = start_time - seg_end = end_time self.channel_cache.put_segment( channel_id=channel_id, diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py index 22386944f..0490aed58 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py @@ -37,7 +37,7 @@ from contextlib import contextmanager from datetime import datetime, timedelta, timezone -from typing import Any, Iterator +from typing import Any, Iterator, cast from unittest.mock import MagicMock, patch import pandas as pd @@ -232,9 +232,9 @@ def _put( if data is None: data = _frame(channel_id, rows=rows, start=start, offset=offset, freq=freq) if seg_start is None: - seg_start = data.index[0].to_pydatetime() + seg_start = cast(pd.Timestamp, data.index[0]).to_pydatetime() if seg_end is None: - seg_end = data.index[-1].to_pydatetime() + seg_end = cast(pd.Timestamp, data.index[-1]).to_pydatetime() adapter.put_segment( channel_id=channel_id, run_id=run_id, From 96038d4fbb7a306c3f328caa31b0799de25d90ea Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 30 Jun 2026 14:20:34 -0700 Subject: [PATCH 6/7] lint --- python/lib/sift_client/_internal/low_level_wrappers/data.py | 2 +- .../_tests/_internal/low_level_wrappers/test_data.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data.py b/python/lib/sift_client/_internal/low_level_wrappers/data.py index 8e4f28008..7b5faaf38 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data.py @@ -674,7 +674,7 @@ def _update_cache( # the wire's nanosecond timestamps), so ``index[0]`` is # always ``pd.Timestamp`` at runtime; pandas-stubs types # it as the wider ``Scalar`` union. - seg_start = cast(pd.Timestamp, combined.index[0]).to_pydatetime() + seg_start = cast("pd.Timestamp", combined.index[0]).to_pydatetime() else: seg_start = start_time diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py index 0490aed58..2aeb956ec 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py @@ -232,9 +232,9 @@ def _put( if data is None: data = _frame(channel_id, rows=rows, start=start, offset=offset, freq=freq) if seg_start is None: - seg_start = cast(pd.Timestamp, data.index[0]).to_pydatetime() + seg_start = cast("pd.Timestamp", data.index[0]).to_pydatetime() if seg_end is None: - seg_end = cast(pd.Timestamp, data.index[-1]).to_pydatetime() + seg_end = cast("pd.Timestamp", data.index[-1]).to_pydatetime() adapter.put_segment( channel_id=channel_id, run_id=run_id, From 72dfc3ade13c8a0c34dadacef2ada99fa5d1f511 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 30 Jun 2026 14:26:17 -0700 Subject: [PATCH 7/7] fix(python): narrow merged_id before >= comparison in compaction test Pyright flagged the post-compaction `merged_id >= 4` check because `SegmentRef.seg_id` is now `int | None` for empty refs. The merged ref is always backed by data here, so add an explicit `is not None` guard before the ordering assertion. Co-authored-by: Cursor --- .../sift_client/_tests/_internal/low_level_wrappers/test_data.py | 1 + 1 file changed, 1 insertion(+) diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py index 2aeb956ec..8b5c98370 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py @@ -756,6 +756,7 @@ def test_crossing_cap_collapses_to_single_segment(self, tmp_path, monkeypatch): assert idx is not None assert len(idx.segments) == 1 merged_id = idx.segments[0].seg_id + assert merged_id is not None assert merged_id >= 4 channel_keys = sorted(k for k in store if k.startswith("channel:"))