Deduplicate shared payload geometry columns

CHANGELOG.md

@@ -142,6 +142,12 @@ in the README).
   contract without importing the widget stack.
 
 ### Changed
+- First-paint payload writers now offset-encode and ship a shared canonical
+  geometry column once per build. Repeated direct traces (including `x is y`)
+  share one column-table reference in both packed and split layouts, while
+  finite/log-selected and decimated temporaries remain trace-local. This
+  removes `(K - 1) * 4N` redundant bytes for K traces over one N-row shared
+  axis without a client or protocol change.
 - **Responsive, author-defeatable browser chrome.** XY's visual defaults now
   live in a low-priority cascade layer, so Tailwind utilities, ordinary author
   CSS, and slot styles override them without `!important`. Long legends remain

benchmarks/README.md

@@ -157,6 +157,13 @@ alongside the complete categorical first-payload row. Together they distinguish
 native encoding/sampling regressions from payload-policy or transport
 regressions.
 
+`test_first_payload_shared_x_packed` and
+`test_first_payload_shared_x_split` isolate the common multi-series payload:
+16 direct 100k-point traces with one canonical x array and distinct y arrays.
+Both rows hard-assert one shared x table reference and a 6.8 MB payload (17
+distinct f32 columns), so either a redundant encode or duplicate wire buffer is
+visible as both a timing and byte-contract regression.
+
 `test_codspeed_pyplot.py` tracks the `xy.pyplot` shim's overhead against the
 raw declarative API: each workload (line 10k/1M, scatter 100k, histogram,
 categorical bars, a chrome-heavy styled panel, and static PNG export) is built

benchmarks/test_codspeed_kernels.py

@@ -39,6 +39,7 @@
 APPEND_BATCH = 1_000
 STACK_ROWS = 8
 STACK_COLS = 100_000
+SHARED_COLUMN_TRACES = 16
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -96,6 +97,17 @@ def medium_data() -> tuple[np.ndarray, np.ndarray]:
     return x, y
 
 
+@pytest.fixture(scope="module")
+def shared_column_figure() -> Figure:
+    """K direct traces over one canonical x column, the dashboard hot path."""
+    x = 1.6e12 + np.arange(MEDIUM_N, dtype=np.float64)
+    rows = np.arange(MEDIUM_N, dtype=np.float64)
+    marks = [
+        xy.scatter(x=x, y=np.sin(rows * 0.001 + phase)) for phase in range(SHARED_COLUMN_TRACES)
+    ]
+    return xy.chart(*marks).figure()
+
+
 @pytest.fixture(scope="module")
 def export_data() -> tuple[np.ndarray, np.ndarray]:
     rng = np.random.default_rng(23)
@@ -712,6 +724,26 @@ def test_first_payload_scatter_medium(benchmark, medium_data):
     assert payload_bytes > 0
 
 
+def _assert_shared_column_payload(spec: dict, payload_bytes: int) -> None:
+    traces = spec["traces"]
+    assert len(traces) == SHARED_COLUMN_TRACES
+    assert len({trace["x"] for trace in traces}) == 1
+    assert len(spec["columns"]) == SHARED_COLUMN_TRACES + 1
+    assert payload_bytes == (SHARED_COLUMN_TRACES + 1) * MEDIUM_N * np.dtype(np.float32).itemsize
+
+
+def test_first_payload_shared_x_packed(benchmark, shared_column_figure):
+    """Packed K-trace payload encodes and ships its canonical x exactly once."""
+    spec, blob = benchmark(shared_column_figure.build_payload)
+    _assert_shared_column_payload(spec, len(blob))
+
+
+def test_first_payload_shared_x_split(benchmark, shared_column_figure):
+    """Live-host split payload keeps one borrowed buffer for the shared x."""
+    spec, buffers = benchmark(shared_column_figure.build_payload_split)
+    _assert_shared_column_payload(spec, sum(buffer.nbytes for buffer in buffers))
+
+
 def test_first_payload_scatter_categorical_color(benchmark, medium_data):
     """Categorical palette factorization and code shipping for scatter."""
     x, y = medium_data

python/xy/_payload.py

@@ -53,19 +53,56 @@ def __init__(self, *, split: bool = False, borrow_heatmaps: bool = False) -> Non
         self._split = split
         self.borrow_heatmaps = borrow_heatmaps
         self.borrowed: list[np.ndarray] = []
+        # One first-paint build may mention the same canonical Column from
+        # many traces (the common shared-x case). Keep the memo writer-local:
+        # streaming/appends get a fresh writer, and derived row selections
+        # must remain independent columns. Integer identities are paired with
+        # live refs in each value, so object-id reuse cannot create a hit.
+        self._canonical_ships: dict[
+            tuple[int, int, int, float, float, str, object],
+            tuple[Column, np.ndarray, int],
+        ] = {}
 
     def ship(self, values: np.ndarray, col: "Column") -> int:
         """Offset-encoded geometry column: `(v - offset) * scale` as f32
         (§4/§16). Scale is 1.0 except for absurd-magnitude domains, where it
-        normalizes so finite f64 can't overflow to ±inf in f32 (§19)."""
-        encoded = lod.encode_f32_values(
-            values,
-            col.suggest_offset(),
-            col.min,
-            col.max,
-            kind=col.kind,
-        )
-        return self._append(encoded.values, encoded.meta)
+        normalizes so finite f64 can't overflow to ±inf in f32 (§19).
+
+        Only the live canonical array (`values is col.values`) is memoized.
+        Filtered, selected, and decimated arrays are derived payloads whose
+        row semantics belong to one trace, even when their bytes happen to
+        match another trace's temporary.
+        """
+        offset = col.suggest_offset()
+        lo, hi = col.min, col.max
+        encoder = lod.encode_f32_values
+        canonical = values is col.values
+        key = None
+        if canonical:
+            # `offset`, safe scale, `kind`, and encoder identity completely
+            # describe this primitive's current wire encoding. Column object,
+            # store-local id, and array object identity prevent collisions
+            # between independent stores or a Column whose storage was
+            # rebound by append.
+            key = (
+                id(col),
+                col.id,
+                id(values),
+                offset,
+                lod.f32_safe_scale(offset, lo, hi),
+                col.kind,
+                encoder,
+            )
+            memo = self._canonical_ships.get(key)
+            if memo is not None:
+                memo_col, memo_values, index = memo
+                if memo_col is col and memo_values is values:
+                    return index
+        encoded = encoder(values, offset, lo, hi, kind=col.kind)
+        index = self._append(encoded.values, encoded.meta)
+        if key is not None:
+            self._canonical_ships[key] = (col, values, index)
+        return index
 
     def ship_scalar(self, values: np.ndarray) -> int:
         """Raw f32 column already in final units (no offset): channel/grid/heights."""

spec/design/wire-protocol.md

@@ -219,6 +219,18 @@ Column entries otherwise carry `len`, an optional `dtype` (`"u8"` or `"u32"`;
 absent means f32), and, for offset-encoded geometry,
 `offset`/`scale`/`kind`.
 
+Trace fields reference the column table by index, and multiple fields or traces
+may intentionally reference the same entry. During one first-paint build, the
+writer memoizes only an exact live canonical array (`values is col.values`)
+with the same Column identity/store-local id and offset/scale/kind encoder
+semantics. Thus K direct traces over one canonical x column encode and ship x
+once in both packed and split layouts, with no client or protocol change.
+Finite/log-filtered selections, decimated arrays, and other derived temporary
+geometry are never memoized: their row mapping belongs to the individual trace
+even when two temporary arrays happen to contain identical bytes. Keyed
+transition `lo`/`hi` u32 columns also remain trace-local and outside this
+geometry memo.
+
 The client picks the layout from the spec, never from the shape of what
 arrived, and **a disagreement is a fatal error, not a fallback**.
 `payloadBuffers` throws when the spec says `split` and the transport delivered

tests/test_animation.py

@@ -10,10 +10,11 @@
 import xy
 
 
-def _column(blob: bytes, spec: dict, index: int) -> np.ndarray:
+def _column(blob: bytes | list[memoryview], spec: dict, index: int) -> np.ndarray:
     meta = spec["columns"][index]
     dtype = np.uint32 if meta.get("dtype") == "u32" else np.float32
-    return np.frombuffer(blob, dtype=dtype, count=meta["len"], offset=meta["byte_offset"])
+    source = blob if isinstance(blob, bytes) else blob[meta["buf"]]
+    return np.frombuffer(source, dtype=dtype, count=meta["len"], offset=meta["byte_offset"])
 
 
 def test_animation_component_serializes_without_callbacks() -> None:
@@ -154,6 +155,46 @@ def test_keyed_scatter_ships_identity_as_binary_u32_words() -> None:
     assert [column.get("dtype") for column in spec["columns"]].count("u32") == 2
 
 
+@pytest.mark.parametrize("layout", ["packed", "split"])
+def test_shared_geometry_dedup_keeps_animation_keys_trace_local(layout: str) -> None:
+    x = np.arange(6.0)
+    keys = [f"row-{index}" for index in range(len(x))]
+    chart = xy.chart(
+        xy.scatter(x=x, y=x + 10.0, key=keys),
+        xy.scatter(x=x, y=x + 20.0, key=keys),
+        xy.animation(match="key"),
+    )
+    figure = chart.figure()
+    spec, payload = figure.build_payload() if layout == "packed" else figure.build_payload_split()
+    first, second = spec["traces"]
+
+    assert first["x"] == second["x"]
+    key_refs = [
+        first["keys"]["lo"],
+        first["keys"]["hi"],
+        second["keys"]["lo"],
+        second["keys"]["hi"],
+    ]
+    assert len(set(key_refs)) == 4
+    assert len(spec["columns"]) == 7  # shared x + two y + four u32 key words
+    first_pairs = list(
+        zip(
+            _column(payload, spec, first["keys"]["lo"]),
+            _column(payload, spec, first["keys"]["hi"]),
+            strict=True,
+        )
+    )
+    second_pairs = list(
+        zip(
+            _column(payload, spec, second["keys"]["lo"]),
+            _column(payload, spec, second["keys"]["hi"]),
+            strict=True,
+        )
+    )
+    assert first_pairs == second_pairs
+    assert len(set(first_pairs)) == len(keys)
+
+
 def test_stable_keys_are_type_sensitive_and_deterministic() -> None:
     chart = xy.scatter_chart(
         xy.scatter(x=[1.0, 2.0, 3.0], y=[3.0, 4.0, 5.0], key=[1, 1.0, True]),

tests/test_figure.py

@@ -31,6 +31,26 @@ def _decoded_payload_col(spec, blob, ref):
     return vals.astype(np.float64) / meta.get("scale", 1.0) + meta.get("offset", 0.0)
 
 
+def _build_layout(fig, layout):
+    return fig.build_payload() if layout == "packed" else fig.build_payload_split()
+
+
+def _decoded_layout_col(spec, payload, ref):
+    meta = spec["columns"][ref]
+    source = payload if isinstance(payload, bytes) else payload[meta["buf"]]
+    values = np.frombuffer(
+        source,
+        dtype=np.float32,
+        count=meta["len"],
+        offset=meta["byte_offset"],
+    )
+    return values.astype(np.float64) / meta.get("scale", 1.0) + meta.get("offset", 0.0)
+
+
+def _layout_nbytes(payload):
+    return len(payload) if isinstance(payload, bytes) else sum(buf.nbytes for buf in payload)
+
+
 def _bar_payload(spec, blob, tr):
     bar = tr["bar"]
     pos = _decoded_payload_col(spec, blob, bar["pos"])
@@ -135,6 +155,142 @@ def test_build_payload_split_folds_u8_alignment_padding_into_column_buffer():
     assert b"".join(bytes(buf) for buf in bufs) == blob
 
 
+@pytest.mark.parametrize("layout", ["packed", "split"])
+def test_payload_deduplicates_shared_canonical_column(layout, monkeypatch):
+    from xy import lod
+
+    n = 64
+    shared_x = 1.6e12 + np.arange(n, dtype=np.float64)
+    ys = [np.sin(np.arange(n) * 0.1 + phase) for phase in range(3)]
+    fig = Figure()
+    for y in ys:
+        fig.scatter(shared_x, y)
+
+    encode_calls = []
+    encode = lod.encode_f32_values
+
+    def counting_encode(values, offset, lo, hi, *, kind=None):
+        encode_calls.append(values)
+        return encode(values, offset, lo, hi, kind=kind)
+
+    monkeypatch.setattr(lod, "encode_f32_values", counting_encode)
+
+    spec, payload = _build_layout(fig, layout)
+    traces = spec["traces"]
+    assert len({trace["x"] for trace in traces}) == 1
+    assert len(encode_calls) == 1 + len(ys)
+    assert sum(values is fig.traces[0].x.values for values in encode_calls) == 1
+    assert len(spec["columns"]) == 1 + len(ys)
+    assert _layout_nbytes(payload) == (1 + len(ys)) * n * np.dtype(np.float32).itemsize
+    for trace, expected_y in zip(traces, ys, strict=True):
+        np.testing.assert_allclose(_decoded_layout_col(spec, payload, trace["x"]), shared_x)
+        np.testing.assert_allclose(
+            _decoded_layout_col(spec, payload, trace["y"]), expected_y, rtol=1e-6, atol=1e-6
+        )
+
+
+@pytest.mark.parametrize("layout", ["packed", "split"])
+def test_payload_deduplicates_x_is_y_canonical_column(layout):
+    values = np.linspace(-2.0, 3.0, 51, dtype=np.float64)
+    spec, payload = _build_layout(Figure().scatter(values, values), layout)
+    trace = spec["traces"][0]
+
+    assert trace["x"] == trace["y"]
+    assert len(spec["columns"]) == 1
+    assert _layout_nbytes(payload) == values.size * np.dtype(np.float32).itemsize
+    np.testing.assert_allclose(_decoded_layout_col(spec, payload, trace["x"]), values, atol=1e-7)
+
+
+@pytest.mark.parametrize("layout", ["packed", "split"])
+def test_payload_does_not_deduplicate_different_finite_selections(layout):
+    shared_x = np.array([10.0, 20.0, 30.0, 40.0, 50.0])
+    y1 = np.array([1.0, np.nan, 3.0, 4.0, 5.0])
+    y2 = np.array([1.0, 2.0, np.nan, 4.0, 5.0])
+    fig = Figure().scatter(shared_x, y1).scatter(shared_x, y2)
+
+    spec, payload = _build_layout(fig, layout)
+    first, second = spec["traces"]
+    assert first["x"] != second["x"]
+    np.testing.assert_array_equal(
+        _decoded_layout_col(spec, payload, first["x"]), shared_x[[0, 2, 3, 4]]
+    )
+    np.testing.assert_array_equal(
+        _decoded_layout_col(spec, payload, second["x"]), shared_x[[0, 1, 3, 4]]
+    )
+
+
+@pytest.mark.parametrize("layout", ["packed", "split"])
+def test_payload_does_not_deduplicate_different_log_selections(layout):
+    shared_x = np.array([10.0, 20.0, 30.0, 40.0])
+    y1 = np.array([1.0, -2.0, 3.0, 4.0])
+    y2 = np.array([1.0, 2.0, -3.0, 4.0])
+    fig = Figure().scatter(shared_x, y1).scatter(shared_x, y2)
+    fig.set_axis("y", type_="log")
+
+    spec, payload = _build_layout(fig, layout)
+    first, second = spec["traces"]
+    assert first["x"] != second["x"]
+    np.testing.assert_array_equal(
+        _decoded_layout_col(spec, payload, first["x"]), shared_x[[0, 2, 3]]
+    )
+    np.testing.assert_array_equal(
+        _decoded_layout_col(spec, payload, second["x"]), shared_x[[0, 1, 3]]
+    )
+
+
+@pytest.mark.parametrize("layout", ["packed", "split"])
+def test_payload_does_not_deduplicate_decimated_temporaries(layout):
+    n = DECIMATION_THRESHOLD + 1
+    shared_x = np.arange(n, dtype=np.float64)
+    shared_y = np.sin(shared_x * 0.01)
+    fig = Figure().line(shared_x, shared_y).line(shared_x, shared_y)
+
+    spec, _payload = _build_layout(fig, layout)
+    first, second = spec["traces"]
+    assert first["tier"] == second["tier"] == "decimated"
+    assert first["x"] != second["x"]
+    assert first["y"] != second["y"]
+
+
+@pytest.mark.parametrize("split", [False, True])
+def test_payload_writer_memo_uses_live_column_identity_and_encoding_semantics(split, monkeypatch):
+    from xy import lod
+    from xy._payload import _PayloadWriter
+
+    values = np.arange(8.0)
+    first_col = ColumnStore().ingest(values)
+    same_id_col = ColumnStore().ingest(values)
+    assert first_col.id == same_id_col.id == 0
+    assert first_col.values is same_id_col.values
+
+    writer = _PayloadWriter(split=split)
+    first_ref = writer.ship(first_col.values, first_col)
+    assert writer.ship(first_col.values, first_col) == first_ref
+    # Same store-local id and exact array, but a different live Column object.
+    assert writer.ship(same_id_col.values, same_id_col) != first_ref
+
+    derived = first_col.values[:]
+    derived_ref = writer.ship(derived, first_col)
+    assert writer.ship(derived, first_col) != derived_ref
+
+    # Offset/scale/kind wire semantics are part of the key; changing them
+    # cannot reuse the old ref even while Column and values identities remain.
+    monkeypatch.setattr(first_col, "suggest_offset", lambda: 100.0)
+    changed_offset_ref = writer.ship(first_col.values, first_col)
+    assert changed_offset_ref != first_ref
+    assert writer.columns[changed_offset_ref]["offset"] == 100.0
+
+    monkeypatch.setattr(lod, "f32_safe_scale", lambda _offset, _lo, _hi: 0.5)
+    changed_scale_ref = writer.ship(first_col.values, first_col)
+    assert changed_scale_ref not in (first_ref, changed_offset_ref)
+    assert writer.columns[changed_scale_ref]["scale"] == 0.5
+
+    first_col.kind = "time_ms"
+    changed_kind_ref = writer.ship(first_col.values, first_col)
+    assert changed_kind_ref not in (first_ref, changed_offset_ref, changed_scale_ref)
+    assert writer.columns[changed_kind_ref]["kind"] == "time_ms"
+
+
 def test_offset_encoding_roundtrip():
     x = 1.6e12 + np.arange(5000, dtype=np.float64)  # ms timestamps
     y = np.sin(np.arange(5000) * 0.01)