Sort line inputs before canonical ingest

CHANGELOG.md

@@ -142,6 +142,11 @@ in the README).
   contract without importing the widget stack.
 
 ### Changed
+- Unsorted line, area, and error-band inputs now canonicalize and stable-sort
+  their parallel columns before the single column-store commit. This removes
+  the unreferenced unsorted copies and duplicate zone-map scans (halving
+  canonical residency for those traces) while retaining datetime/category
+  metadata and stable duplicate-x ordering.
 - **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/test_codspeed_kernels.py

@@ -782,6 +782,19 @@ def build():
     assert buffers
 
 
+def test_unsorted_line_ingest_medium(benchmark):
+    """Canonicalize, stable-sort, and ingest an unsorted line exactly once."""
+    rng = np.random.default_rng(170)
+    order = rng.permutation(MEDIUM_N)
+    x = np.arange(MEDIUM_N, dtype=np.float64)[order]
+    y = np.sin(np.arange(MEDIUM_N, dtype=np.float64) * 0.001)[order]
+
+    fig = benchmark(lambda: xy.chart(xy.line(x=x, y=y)).figure())
+
+    assert len(fig.store) == 2
+    assert fig.store.memory_report()["canonical_bytes"] == x.nbytes + y.nbytes
+
+
 def test_density_view_exact_pan(benchmark):
     """Steady-state exact pan below the pyramid activation threshold."""
     n = 200_000

python/xy/_figure.py

@@ -371,6 +371,38 @@ def _ingest_xy(self, x: Any, y: Any, kind: str) -> tuple[Column, Column]:
             self.store.rollback(checkpoint)
             raise
 
+    def _ingest_sorted_xy(
+        self,
+        x: Any,
+        y: Any,
+        kind: str,
+        *parallel: Any,
+        parallel_labels: tuple[str, ...] = (),
+    ) -> tuple[Column, ...]:
+        """Stable-sort line-like parallel columns before their only ingest."""
+        checkpoint = self.store.checkpoint()
+        try:
+            try:
+                return self.store.ingest_sorted(x, y, *parallel)
+            except columns._ColumnLengthMismatch as error:
+                if error.index == 1:
+                    raise ValueError(
+                        f"{kind} x and y must have equal length, got "
+                        f"{error.expected} and {error.actual}"
+                    ) from error
+                label_index = error.index - 2
+                label = (
+                    parallel_labels[label_index]
+                    if label_index < len(parallel_labels)
+                    else f"{kind} column {error.index}"
+                )
+                raise ValueError(
+                    f"{label} must have length {error.expected}, got {error.actual}"
+                ) from error
+        except Exception:
+            self.store.rollback(checkpoint)
+            raise
+
     def _checkpoint(self) -> _FigureCheckpoint:
         return (
             self.store.checkpoint(),

python/xy/columns.py

@@ -35,6 +35,16 @@
 ZONE_CHUNK = 65_536
 
 
+class _ColumnLengthMismatch(ValueError):
+    """Internal structured error for parallel-column ingest validation."""
+
+    def __init__(self, index: int, expected: int, actual: int) -> None:
+        self.index = index
+        self.expected = expected
+        self.actual = actual
+        super().__init__(f"column {index} must have length {expected}, got {actual}")
+
+
 @dataclass
 class ZoneMaps:
     """Per-chunk column statistics (min/max/counts; design dossier §22)."""
@@ -280,6 +290,19 @@ def ingest_pair(self, x: Any, y: Any) -> tuple[Column, Column]:
         y_arr, y_kind, y_copies = _canonicalize(y)
         if len(x_arr) != len(y_arr):
             raise ValueError(f"x and y must have equal length, got {len(x_arr)} and {len(y_arr)}")
+        return self._ingest_canonical_pair(
+            (x_arr, x_kind, x_copies),
+            (y_arr, y_kind, y_copies),
+        )
+
+    def _ingest_canonical_pair(
+        self,
+        x: tuple[npt.NDArray[np.float64], str, int],
+        y: tuple[npt.NDArray[np.float64], str, int],
+    ) -> tuple[Column, Column]:
+        """Ingest two already-canonical columns, fusing their zone scan."""
+        x_arr, x_kind, x_copies = x
+        y_arr, y_kind, y_copies = y
         x_key = self._array_key(x_arr)
         y_key = self._array_key(y_arr)
         x_hit = self._lookup(x_arr, x_key)
@@ -320,6 +343,67 @@ def ingest_pair(self, x: Any, y: Any) -> tuple[Column, Column]:
         )
         return x_col, y_col
 
+    def ingest_sorted(self, x: Any, *parallel: Any) -> tuple[Column, ...]:
+        """Stable-sort parallel columns by ``x`` before their only ingest.
+
+        Every input is canonicalized and length-validated before the store is
+        mutated.  When ``x`` is unsorted, one stable order is gathered across
+        all columns; the gathered arrays retain their canonical kinds and the
+        required copy is included in ingest accounting.  The first two new
+        columns still share the fused zone-map scan used by ``ingest_pair``.
+
+        ``_ColumnLengthMismatch.index`` identifies the mismatching parallel
+        input (``1`` is the first value after ``x``), allowing trace builders
+        to retain their public, mark-specific validation messages.
+        """
+        canonical = [_canonicalize(values) for values in (x, *parallel)]
+        expected = len(canonical[0][0])
+        for index, (arr, _kind, _copies) in enumerate(canonical[1:], start=1):
+            if len(arr) != expected:
+                raise _ColumnLengthMismatch(index, expected, len(arr))
+
+        x_arr = canonical[0][0]
+        if not kernels.is_sorted(x_arr):
+            order = np.argsort(x_arr, kind="stable")
+            gathered: list[tuple[npt.NDArray[np.float64], str, int]] = []
+            gathered_by_key: dict[
+                tuple[int, int, int], tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]
+            ] = {}
+            for arr, kind, copies in canonical:
+                key = self._array_key(arr)
+                cached = gathered_by_key.get(key)
+                if cached is not None and np.shares_memory(cached[0], arr):
+                    sorted_arr = cached[1]
+                else:
+                    sorted_arr = arr[order]
+                    gathered_by_key[key] = (arr, sorted_arr)
+                gathered.append((sorted_arr, kind, copies + 1))
+            canonical = gathered
+
+        if len(canonical) == 1:
+            arr, kind, copies = canonical[0]
+            return (
+                self._ingest_canonical(
+                    arr,
+                    kind,
+                    copies,
+                    defer_zone_maps=False,
+                ),
+            )
+
+        first, second = self._ingest_canonical_pair(canonical[0], canonical[1])
+        columns = [first, second]
+        for arr, kind, copies in canonical[2:]:
+            columns.append(
+                self._ingest_canonical(
+                    arr,
+                    kind,
+                    copies,
+                    defer_zone_maps=False,
+                )
+            )
+        return tuple(columns)
+
     def memory_report(self) -> dict[str, Any]:
         """Canonical bytes per column — if a number isn't in the report, it
         isn't real (design dossier §27). Derived/GPU classes are added as

python/xy/marks.py

@@ -798,16 +798,10 @@ def line(
     dash_spec = _validate.dash(dash, "line dash")
     checkpoint = self._checkpoint()
     try:
-        xc, yc = self._ingest_xy(x, y, "line")
-        if not kernels.is_sorted(xc.values):
-            # LOD contract (§28): line x must be sorted; the engine sorts once
-            # at ingest, and says so. The predicate is NaN-safe on purpose:
-            # a NaN fails its pairs, so a NaN-carrying x cannot skip the sort
-            # and violate M4's sorted precondition.
-            # argsort places NaNs last, where the m4 window excludes them.
-            order = np.argsort(xc.values, kind="stable")
-            xc = self.store.ingest(xc.values[order])
-            yc = self.store.ingest(yc.values[order])
+        # LOD contract (§28): line x is sorted once at ingest.  The store
+        # canonicalizes before deriving the stable order, so unsorted source
+        # columns never become orphan canonical entries.
+        xc, yc = self._ingest_sorted_xy(x, y, "line")
         style: dict[str, Any] = {"color": color, "width": width, "opacity": opacity}
         style.update(styles._opacity_channels(css))
         if curve != "linear":
@@ -876,19 +870,17 @@ def area(
     dash_spec = _validate.dash(dash, "area dash")
     checkpoint = self._checkpoint()
     try:
-        xc, yc = self._ingest_xy(x, y, "area")
-        bc = (
-            self.store.ingest(np.full(len(xc), self._finite_scalar(base, "area base")))
-            if np.isscalar(base)
-            else self.store.ingest(base)
-        )
-        if len(bc) != len(xc):
-            raise ValueError(f"area base must have length {len(xc)}, got {len(bc)}")
-        if not kernels.is_sorted(xc.values):
-            order = np.argsort(xc.values, kind="stable")
-            xc = self.store.ingest(xc.values[order])
-            yc = self.store.ingest(yc.values[order])
-            bc = self.store.ingest(bc.values[order])
+        if np.isscalar(base):
+            xc, yc = self._ingest_sorted_xy(x, y, "area")
+            bc = self.store.ingest(np.full(len(xc), self._finite_scalar(base, "area base")))
+        else:
+            xc, yc, bc = self._ingest_sorted_xy(
+                x,
+                y,
+                "area",
+                base,
+                parallel_labels=("area base",),
+            )
         style: dict[str, Any] = {
             "color": color,
             "opacity": opacity,
@@ -955,15 +947,13 @@ def error_band(
     fill_spec = _validate.mark_fill(fill, "error_band fill")
     checkpoint = self._checkpoint()
     try:
-        xc, lc = self._ingest_xy(x, lower, "error_band")
-        uc = self.store.ingest(self._as_1d_float(upper, "error_band upper"))
-        if len(uc) != len(xc):
-            raise ValueError(f"error_band upper must have length {len(xc)}, got {len(uc)}")
-        if not kernels.is_sorted(xc.values):
-            order = np.argsort(xc.values, kind="stable")
-            xc = self.store.ingest(xc.values[order])
-            lc = self.store.ingest(lc.values[order])
-            uc = self.store.ingest(uc.values[order])
+        xc, lc, uc = self._ingest_sorted_xy(
+            x,
+            lower,
+            "error_band",
+            self._as_1d_float(upper, "error_band upper"),
+            parallel_labels=("error_band upper",),
+        )
         style: dict[str, Any] = {
             "color": color,
             "opacity": opacity,

spec/design-dossier.md

@@ -190,7 +190,10 @@ The two requirements live primarily in the **data pipeline (§4–§6)**. The re
   *references* (column id + offset + length) into immutable canonical buffers. The
   calc/LOD stages produce *derived* buffers only when they must (e.g. a decimated
   view), never a defensive clone of the raw data. Contrast Plotly's `data` +
-  `_fullData` + `calcdata` triplication.
+  `_fullData` + `calcdata` triplication. Line-like marks canonicalize all parallel
+  columns, derive one stable x-order, and gather that order **before** the store
+  commits them; only the sorted columns become canonical entries, datetime kinds
+  survive the gather, and its unavoidable copy is included in ingest accounting.
 - **Struct-of-Arrays, not Array-of-Structs.** `x[]`, `y[]` as contiguous typed
   arrays — cache-friendly, and each column uploads to the GPU as one vertex buffer
   with no marshalling.

tests/test_components.py

@@ -1705,6 +1705,26 @@ def test_bar_chart_data_keys_and_category_axis():
     assert spec["x_axis"]["categories"] == ["a", "b", "c"]
 
 
+def test_unsorted_line_data_keys_preserve_categorical_order_without_orphans():
+    data = FakeFrame(
+        {
+            "label": np.array(["beta", "alpha", "beta"]),
+            "value": np.array([2.0, 1.0, 3.0]),
+        }
+    )
+
+    fig = xy.line_chart(xy.line(x="label", y="value"), data=data).figure()
+    trace = fig.traces[0]
+    spec, _blob = fig.build_payload()
+
+    assert len(fig.store) == 2
+    assert spec["x_axis"]["kind"] == "category"
+    assert spec["x_axis"]["categories"] == ["beta", "alpha"]
+    np.testing.assert_array_equal(trace.x.values, [0.0, 0.0, 1.0])
+    # Stable sorting keeps both "beta" rows in their source order.
+    np.testing.assert_array_equal(trace.y.values, [2.0, 3.0, 1.0])
+
+
 def test_component_xy_datetime_object_axes_do_not_become_categories():
     x = np.array(
         [