perf: tail-only f32 re-encode on streaming append

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

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,

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,13 +558,17 @@ 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,
                     "kind": c.kind,
                     "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,
         }
 

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. |
 

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

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.
 

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