diff --git a/benchmarks/test_codspeed_kernels.py b/benchmarks/test_codspeed_kernels.py index 3bd90fab..d4875c92 100644 --- a/benchmarks/test_codspeed_kernels.py +++ b/benchmarks/test_codspeed_kernels.py @@ -1074,6 +1074,24 @@ def append_next(): assert buffers +def test_stream_scatter_append_direct(benchmark, append_data): + """Direct-tier scatter append: the tail-only f32 re-encode keeps the + refresh proportional to the appended batch, not the accumulated rows + (`Column.encoded_f32`; wire-protocol §4). `density=False` pins the + direct tier so accumulation across iterations never flips the path.""" + x, y, tail_x, tail_y = append_data + fig = xy.chart(xy.scatter(x=x, y=y, density=False)).figure() + fig.build_payload(N_BUCKETS) + + def append_next(): + return fig.append(0, tail_x, tail_y) + + update, buffers = benchmark(append_next) + assert update["spec"]["traces"][0]["tier"] == "direct" + assert update["spec"]["traces"][0]["n_points"] >= APPEND_N + APPEND_BATCH + assert buffers + + def test_stream_density_append_incremental_pyramid(benchmark, pyramid_data): """Stable-domain density append with an in-place native pyramid update.""" x, y = pyramid_data diff --git a/python/xy/_payload.py b/python/xy/_payload.py index 98d3dea0..19fa2f79 100644 --- a/python/xy/_payload.py +++ b/python/xy/_payload.py @@ -89,6 +89,15 @@ def ship(self, values: np.ndarray, col: "Column", *, scale: str | None = None) - if lod.pins_offset_to_zero(scale) else col.suggest_offset() ) + if values is col.values and len(values): + # Whole-column identity ship: serve the column's f32 encode cache, + # which re-encodes only the rows appended since the last build — + # the kernel-side twin of the client's tail-only GPU upload. + # Filtered/decimated/masked views take the fresh-encode path below. + offset_f = float(offset) + f32_scale = lod.f32_safe_scale(offset_f, float(col.min), float(col.max)) + enc = col.encoded_f32(offset_f, f32_scale) + return self._append(enc, {"offset": offset_f, "scale": f32_scale, "kind": col.kind}) encoded = lod.encode_f32_values( values, offset, diff --git a/python/xy/columns.py b/python/xy/columns.py index 7d37a745..7850e1af 100644 --- a/python/xy/columns.py +++ b/python/xy/columns.py @@ -205,6 +205,11 @@ class Column: # consecutive append payloads keep byte-identical prefixes — the client's # tail-only GPU upload depends on it (wire-protocol §4). _ship_offset: float | None = field(default=None, init=False, repr=False, compare=False) + # Cached whole-column f32 encoding: (offset, scale, n, base_ptr, buffer). + # A rebuildable derived cache (§27); see `encoded_f32`. + _enc_cache: tuple[float, float, int, int, npt.NDArray[np.float32]] | None = field( + default=None, init=False, repr=False, compare=False + ) def __len__(self) -> int: return len(self.values) @@ -261,6 +266,49 @@ def suggest_offset(self) -> float: self._ship_offset = mid return mid + @property + def encode_cache_bytes(self) -> int: + """Bytes held by the whole-column f32 encode cache (0 when cold).""" + return int(self._enc_cache[4].nbytes) if self._enc_cache is not None else 0 + + def encoded_f32(self, offset: float, scale: float) -> npt.NDArray[np.float32]: + """Whole-column relative-f32 encoding `(v - offset) * scale`, cached. + + The sticky ship offset (`suggest_offset`) keeps the encoding + prefix-stable across appends, so a streaming tick only encodes the + appended tail — O(rows appended), not O(N) — into a capacity-doubling + f32 buffer that mirrors the canonical growth buffer (+4 B/point, + itemized in `memory_report`). Invalidation is conservative: an + offset/scale change, a shrink, or a rebinding/growth-buffer migration + (detected by the values base pointer) re-encodes fully into a *fresh* + buffer — previously shipped split payloads hold zero-copy views of the + old one, so it is never written again past its shipped length. The + cache trusts the canonical contract that column values are only ever + extended via `append`, never mutated in place (§27 rule 1). + """ + n = len(self.values) + ptr = int(self.values.__array_interface__["data"][0]) + cache = self._enc_cache + if cache is not None and cache[0] == offset and cache[1] == scale: + n_old, buf = cache[2], cache[4] + if cache[3] == ptr and n_old <= n: + if buf.shape[0] < n: + grown = np.empty(max(n, buf.shape[0] * 2), dtype=np.float32) + grown[:n_old] = buf[:n_old] + buf = grown + if n > n_old: + buf[n_old:n] = kernels.encode_f32(self.values[n_old:], offset, scale) + self._enc_cache = (offset, scale, n, ptr, buf) + return buf[:n] + # Cold path: adopt the encode output itself as the cache buffer, so a + # first build costs exactly one encode and one exact-size allocation — + # identical to the uncached path (first-paint latency is benchmarked; + # CodSpeed flagged the extra alloc+copy of a slack buffer here). The + # doubling slack starts on the first append instead. + buf = kernels.encode_f32(self.values, offset, scale) + self._enc_cache = (offset, scale, n, ptr, buf) + return buf + def append(self, data: Any) -> None: """Streaming append (design dossier §5, Phase-0 Python-side). @@ -285,6 +333,7 @@ def append(self, data: Any) -> None: return n_old = len(self.values) n_new = n_old + len(arr) + old_ptr = int(self.values.__array_interface__["data"][0]) grow = getattr(self, "_grow", None) if grow is None or grow.shape[0] < n_new: cap = max(n_new, n_old * 2, 1024) @@ -294,6 +343,18 @@ def append(self, data: Any) -> None: self.ingest_copies += 1 # the migration is the O(N) event self._grow[n_old:n_new] = arr self.values = self._grow[:n_new] + # A growth-buffer migration moves the base pointer but preserves the + # prefix byte-for-byte, so the f32 encode cache stays valid — re-key it + # to the new pointer. A cache keyed to some *other* pointer means the + # values were rebound outside this method; drop it (full re-encode is + # always correct). + cache = self._enc_cache + if cache is not None: + if cache[3] == old_ptr and cache[2] <= n_old: + new_ptr = int(self.values.__array_interface__["data"][0]) + self._enc_cache = (cache[0], cache[1], cache[2], new_ptr, cache[4]) + else: + self._enc_cache = None # Recompute only the tail: the last (possibly partial) old chunk plus # everything new. Slicing at a chunk boundary keeps alignment with a # full recompute, so autorange/pruning consumers see identical maps. @@ -476,10 +537,17 @@ def memory_report(self) -> dict[str, Any]: report totals them as ``canonical_capacity_bytes`` — equal to ``canonical_bytes`` for every never-appended figure, and the number the resident total is built from (channels already report their own growth - buffers this way).""" + buffers this way). + + Shipped geometry keeps a whole-column f32 encode cache (tail-only + re-encode on streaming append; `Column.encoded_f32`). That is +4 B per + point of resident RAM per shipped column, itemized per column and + totaled as ``encode_cache_bytes`` — always RAM, even for memmapped + canonical columns.""" resident = 0 resident_capacity = 0 mapped = 0 + encode_cache = 0 columns = [] for c in self._columns: nbytes = int(c.values.nbytes) @@ -490,6 +558,9 @@ def memory_report(self) -> dict[str, Any]: else: resident += nbytes resident_capacity += capacity + # The f32 encode cache is resident RAM regardless of the canonical + # backing (encoding a memmapped column still materializes f32). + encode_cache += c.encode_cache_bytes columns.append( { "id": c.id, @@ -497,6 +568,7 @@ def memory_report(self) -> dict[str, Any]: "len": len(c), "bytes": nbytes, "capacity_bytes": capacity, + "encode_cache_bytes": c.encode_cache_bytes, "backing": "memmap" if memmapped else "ram", "ingest_copies": c.ingest_copies, "null_count": c.zone.null_count, @@ -506,6 +578,7 @@ def memory_report(self) -> dict[str, Any]: "canonical_bytes": resident, "canonical_capacity_bytes": resident_capacity, "canonical_mapped_bytes": mapped, + "encode_cache_bytes": encode_cache, "columns": columns, } diff --git a/spec/benchmarks/results.md b/spec/benchmarks/results.md index 1b92238c..8fd04ab6 100644 --- a/spec/benchmarks/results.md +++ b/spec/benchmarks/results.md @@ -62,7 +62,7 @@ commit so CI artifacts are quick to inspect from logs. | `core_2d_chart_breadth` | Core 2D chart breadth | tracked | The library needs to stay fast beyond the scatter wedge: bars, histograms, areas, and heatmaps are everyday chart workloads. | payload-prep time, payload bytes, standalone HTML bytes, TTFR | `benchmarks/bench_2d_charts.py` vs Plotly/Seaborn; `benchmarks/bench_pyplot_vs_matplotlib.py`; `bench_interaction.py`; CodSpeed core-2D rows | Beat Plotly on user-visible first paint for common 2D charts while tracking Matplotlib/Seaborn raster baselines where applicable. | | — (not in `benchmarks/categories.py`) | Core launch scatter baseline | tracked outside the registry | Launch claims need an immutable, apples-to-apples record of default product behavior from small charts through the 1B-point capacity case. | static PNG time/RSS; interactive TTFR and Python/browser RSS; hardware and SwiftShader kept separate | `benchmarks/bench_launch_scatter.py` vs Plotly and Matplotlib at 10k, 100k, 1M, 10M, and 1B | Preserve the fixed launch contracts and add versioned environment baselines rather than overwriting prior results. | | `input_ingestion` | Input ingestion | tracked | Real applications provide converted, strided, datetime, list, pandas, and Arrow inputs rather than only contiguous f64 arrays. | ingest latency, copies, peak Python memory | `benchmarks/bench_workflows.py` ingestion rows | Keep zero-copy inputs cheap and make unavoidable conversions visible. | -| `streaming_updates` | Streaming updates | tracked | Monitoring and notebook workflows append repeatedly; stable-domain batches should update indexes incrementally while domain growth may rebuild. | append latency, refresh bytes, incremental pyramid update, domain-growth rebuild | `benchmarks/bench_workflows.py` streaming rows; `benchmarks/bench_transport.py` append diagnostics | Keep stable-domain appends proportional to the batch and expose unavoidable rebuild stalls. | +| `streaming_updates` | Streaming updates | tracked | Monitoring and notebook workflows append repeatedly; stable-domain batches should update indexes incrementally while domain growth may rebuild. | append latency, refresh bytes, incremental pyramid update, domain-growth rebuild | `benchmarks/bench_workflows.py` streaming rows; `benchmarks/bench_transport.py` append diagnostics; `test_codspeed_kernels.py::test_stream_line_append`, `::test_stream_scatter_append_direct` (direct-tier tail-only re-encode), `::test_stream_density_append_incremental_pyramid` | Keep stable-domain appends proportional to the batch and expose unavoidable rebuild stalls. | | `log_autorange` | Log autorange | tracked | Large positive/negative and non-finite series are common in monitoring and scientific charts, and log axes must avoid full-data rescans. | range latency, positive-domain correctness, peak Python memory | `benchmarks/bench_workflows.py` log autorange row; `tests/test_figure.py` | Compute correct positive log domains from zone statistics with cost proportional to chunks, not points. | | `static_export` | Static export | tracked | HTML, SVG, and PNG have distinct serialization and browser costs. | export latency, output bytes, peak Python memory | `benchmarks/bench_workflows.py` export rows; `benchmarks/bench_pyplot_vs_matplotlib.py` matched PNG rows | Track each target independently without mixing browser and payload work. | diff --git a/spec/design-dossier.md b/spec/design-dossier.md index 44043aae..f06f4148 100644 --- a/spec/design-dossier.md +++ b/spec/design-dossier.md @@ -994,6 +994,13 @@ Rules that make the mode targets in §2 real: `resident_array_bytes` is built from the capacity total — equal to `canonical_bytes` for any figure that never appended. Continuous channels already reported their own growth buffers this way; columns now match. + Second corollary: every whole-column geometry ship keeps the encoded f32 result as + a per-column cache (`Column.encoded_f32`) so a streaming append re-encodes only the + appended tail (§5) — +4 B/point of resident RAM per shipped column (even when the + canonical column is memmapped), itemized per column and totaled as + `encode_cache_bytes`. It is a rebuildable derived cache under rule 1: any + offset/scale change or values rebinding drops it for a full re-encode into a fresh + buffer, never overwriting bytes a previously shipped split payload still borrows. 5. **Canonical may be out-of-core (native `mmap`).** The "mmap (native)" cell in the table above is realized: a canonical column may be backed by a disk `np.memmap` instead of RAM. Because a memmap is a transparent `ndarray` — same dedup key, same diff --git a/spec/design/wire-protocol.md b/spec/design/wire-protocol.md index 97cc664d..2fa66918 100644 --- a/spec/design/wire-protocol.md +++ b/spec/design/wire-protocol.md @@ -268,7 +268,14 @@ rebuilt otherwise: the kernel makes it *hold* in the common case by keeping each column's encode offset sticky across appends while every value stays within one span of it (`Column.suggest_offset`, ≤1 f32 mantissa bit vs a fresh - midpoint — a right-growing stream never exceeds that bound). + midpoint — a right-growing stream never exceeds that bound). The kernel + exploits the same prefix stability itself: a whole-column geometry ship is + served from a per-column f32 encode cache (`Column.encoded_f32`, a + capacity-doubling buffer mirroring `Column.append`), so each streaming + tick encodes only the appended rows — O(rows appended) CPU per tick, not + O(N) per column. Offset/scale changes or filtered/masked ships fall back + to a full fresh encode, which is always correct; the cache bytes are + itemized in `memory_report()` as `encode_cache_bytes`. - Any precondition failure falls back to destroy + rebuild of that trace, which is always correct. diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 7aa35db7..bdf8dbd6 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -396,6 +396,72 @@ def test_ship_offset_recenters_when_stale_center_leaves_domain(): assert col.suggest_offset() == 4.5 # re-centered to the fresh midpoint +# -------------------------------------------------------------------------- +# Tail-only re-encode — the kernel-side twin of the client's tail upload. +# The sticky offset keeps the encoding prefix-stable, so a streaming tick must +# pay O(appended rows) in encode_f32, not O(N) per column per tick. +# -------------------------------------------------------------------------- + + +def test_append_reencodes_only_the_tail(monkeypatch): + n = 4096 + fig = Figure().scatter(np.arange(float(n)), np.sin(np.arange(float(n)))) + fig.build_payload_split() # first build encodes (and caches) full columns + + calls: list[int] = [] + real = kernels.encode_f32 + + def spy(values, offset, scale=1.0): + calls.append(len(values)) + return real(values, offset, scale) + + monkeypatch.setattr(kernels, "encode_f32", spy) + fig.append(0, [float(n)], [0.5]) + assert calls, "append must encode the appended tail" + # Tail-only: the O(N) full-column re-encode is the regression this pins. + assert max(calls) <= 16, f"append re-encoded a full column: {max(calls)} rows" + + +def test_append_cached_encode_matches_full_reencode(): + fig = Figure().scatter(np.arange(64.0), np.arange(64.0) * 2) + fig.build_payload_split() + msg, buffers = None, None + for i in range(4): + msg, buffers = fig.append(0, [64.0 + i], [128.0 + 2.0 * i]) + spec = msg["spec"] + t = fig.traces[0] + for axis, col in (("x", t.x), ("y", t.y)): + meta, got = _column_bytes(spec, buffers, spec["traces"][0][axis]) + ref = kernels.encode_f32(col.values, meta["offset"], meta["scale"]).tobytes() + assert got == ref, axis + + +def test_shipped_buffers_survive_later_appends_and_recenter(): + # Split-mode buffers are zero-copy views. Later appends write only past + # the shipped prefix, and a cache invalidation (offset recenter) must + # re-encode into a *fresh* buffer — never mutate memory a previously + # shipped payload may still borrow (widget reopen state holds it). + fig = Figure().scatter(np.arange(100.0), np.arange(100.0)) + fig.build_payload_split() + _, bufs1 = fig.append(0, [100.0], [100.0]) + snap = [bytes(b) for b in bufs1] + fig.append(0, [101.0], [101.0]) # tail extension + t = fig.traces[0] + t.x._ship_offset = 1e9 # stale center leaves the domain -> recenter + t.y._ship_offset = 1e9 + fig.append(0, [102.0], [102.0]) # full re-encode under a new offset + assert [bytes(b) for b in bufs1] == snap + + +def test_memory_report_itemizes_encode_cache_bytes(): + n = 256 + fig = Figure().scatter(np.arange(float(n)), np.arange(float(n)) * 3) + report = fig.memory_report() # builds a payload, so the caches are warm + assert report["encode_cache_bytes"] >= 2 * n * 4 # x and y f32 caches + for entry in report["columns"]: + assert entry["encode_cache_bytes"] >= n * 4 + + def test_decimated_entries_record_their_px_width(): from xy.config import DECIMATION_THRESHOLD