Remove the replica columnar path

should_encode_columnar returns False for every ReplicaMetricReport by design --
columnar array merge over many small per-replica sources is slower than the object
path in the controllers decision loop -- so no producer ever emits a columnar replica
frame and the entire receiving side was unreachable.

Drop the replica encoder and decoders, the controllers replica columnar branch, the
per-replica array stores and their dedup-at-write, and the two test files that
covered only that path. Handle reports are now the only columnar producers, so
encode() takes a HandleMetricReport.

This also removes the columnar arms of _get_aggregated_custom_metrics and
_get_raw_custom_metrics, which diverged from the object path on a present-but-empty
series: the object branch reported 0.0 while the columnar branch omitted the replica
entirely, so a custom policy indexing by replica id saw a KeyError on one path only.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: John Taylor <john.taylor@anyscale.com>

Author: johntaylor-cell

python/ray/serve/_private/autoscaling_metrics_codec.py

@@ -22,7 +22,7 @@
 import logging
 import struct
 import zlib
-from typing import Any, Dict, Union
+from typing import Any, Dict
 
 try:
     import numpy as np
@@ -33,8 +33,6 @@
     RUNNING_REQUESTS_KEY,
     DeploymentID,
     HandleMetricReport,
-    ReplicaID,
-    ReplicaMetricReport,
 )
 from ray.serve._private.constants import (
     RAY_SERVE_COLUMNAR_METRICS_MIN_REPLICAS,
@@ -103,15 +101,11 @@ def _flatten_series(series_list):
 # ---------------------------------------------------------------------------
 # encode
 # ---------------------------------------------------------------------------
-def encode(report: Union[HandleMetricReport, ReplicaMetricReport]) -> bytes:
-    if isinstance(report, HandleMetricReport):
-        return _encode_handle(report)
-    return _encode_replica(report)
+def encode(report: HandleMetricReport) -> bytes:
+    return _encode_handle(report)
 
 
-def should_encode_columnar(
-    report: Union[HandleMetricReport, ReplicaMetricReport]
-) -> bool:
+def should_encode_columnar(report: HandleMetricReport) -> bool:
     """Whether a producer should serialize this report columnar (vs Python objects).
 
     Format is chosen by report TYPE and self-identifies on the wire (see is_columnar),
@@ -230,41 +224,6 @@ def _encode_handle(rep: HandleMetricReport) -> bytes:
     return _frame(header, blob)
 
 
-def _encode_replica(rep: ReplicaMetricReport) -> bytes:
-    metric_names = list(rep.metrics.keys())
-    mi = {m: i for i, m in enumerate(metric_names)}
-    entries, series_list = [], []
-    for m in metric_names:
-        series = rep.metrics[m]
-        entries.append((mi[m], 0, len(series)))
-        series_list.append(series)
-    ts, val, _ = _flatten_series(series_list)
-    off = 0
-    for i, (a, _o, n) in enumerate(entries):
-        entries[i] = (a, off, n)
-        off += n
-    arrays = {
-        "entries": (
-            np.array(entries, dtype="<i8") if entries else np.zeros((0, 3), "<i8")
-        ).reshape(-1, 3),
-        "ts": np.array(ts, dtype="<f8"),
-        "val": np.array(val, dtype="<f8"),
-    }
-    descriptors, blob = _pack(arrays)
-    header = {
-        "type": "replica",
-        "replica_unique_id": rep.replica_id.unique_id,
-        "deployment": [
-            rep.replica_id.deployment_id.name,
-            rep.replica_id.deployment_id.app_name,
-        ],
-        "timestamp": rep.timestamp,
-        "metric_names": metric_names,
-        "arrays": descriptors,
-    }
-    return _frame(header, blob)
-
-
 def _frame(header: dict, blob: bytes) -> bytes:
     hb = json.dumps(header).encode()
     # level=1: metric reports are serialized on the producer hot path (per
@@ -319,46 +278,6 @@ def decode(buf: bytes) -> Dict[str, Any]:
 # ---------------------------------------------------------------------------
 # round-trip self-test (TimeStampedValue.value is compare=False, so compare deeply)
 # ---------------------------------------------------------------------------
-def _series_eq(a, b):
-    return len(a) == len(b) and all(
-        x.timestamp == y.timestamp and x.value == y.value for x, y in zip(a, b)
-    )
-
-
-def decode_replica_running_requests(payload, metric_name=RUNNING_REQUESTS_KEY):
-    """For a REPLICA columnar payload, return (replica_id, ts_arr, val_arr, timestamp)
-    for the given metric -- zero-copy arrays, no per-point objects."""
-    view = decode(payload)
-    h = view["header"]
-    dep = DeploymentID(h["deployment"][0], h["deployment"][1])
-    replica_id = ReplicaID(h["replica_unique_id"], dep)
-    ts_arr, val_arr = view["ts"][:0], view["val"][:0]
-    names = h["metric_names"]
-    if metric_name in names:
-        mi = names.index(metric_name)
-        for row in view["entries"]:
-            if int(row[0]) == mi:
-                off, n = int(row[1]), int(row[2])
-                ts_arr, val_arr = view["ts"][off : off + n], view["val"][off : off + n]
-                break
-    return replica_id, ts_arr, val_arr, h["timestamp"]
-
-
-def decode_replica_all_metrics(payload):
-    """For a REPLICA columnar payload, return (replica_id, {metric_name: (ts, val)},
-    timestamp) for ALL metrics -- zero-copy arrays, no per-point objects. Carries
-    custom autoscaling metrics through the columnar path, not just running_requests."""
-    view = decode(payload)
-    h = view["header"]
-    dep = DeploymentID(h["deployment"][0], h["deployment"][1])
-    replica_id = ReplicaID(h["replica_unique_id"], dep)
-    names = h["metric_names"]
-    ts_all, val_all = view["ts"], view["val"]
-    metric_arrays = {}
-    for row in view["entries"]:
-        mi, off, n = int(row[0]), int(row[1]), int(row[2])
-        metric_arrays[names[mi]] = (ts_all[off : off + n], val_all[off : off + n])
-    return replica_id, metric_arrays, h["timestamp"]
 
 
 def decode_handle_flat(payload):

python/ray/serve/_private/autoscaling_state.py

@@ -25,7 +25,6 @@
     ReplicaMetricReport,
     TargetCapacityDirection,
     TimeSeries,
-    TimeStampedValue,
 )
 from ray.serve._private.constants import (
     RAY_SERVE_MIN_HANDLE_METRICS_TIMEOUT_S,
@@ -88,10 +87,8 @@ def __init__(self, deployment_id: DeploymentID):
         self._replica_metrics: Dict[ReplicaID, ReplicaMetricReport] = dict()
         # Columnar per-replica running-requests arrays (wire-detected; producers
         # choose the format via should_encode_columnar).
-        self._replica_running_arrays: Dict[ReplicaID, tuple] = dict()
         # Non-running columnar metrics per replica (custom autoscaling metrics):
         # replica_id -> {metric_name: (ts_arr, val_arr)}.
-        self._replica_custom_arrays: Dict[ReplicaID, Dict[str, tuple]] = dict()
         # Unified per-replica "last accepted report timestamp" across BOTH wire formats.
         # Gates the object AND columnar ingest paths so a delayed report in either format
         # can't overwrite fresher data the other wrote. Cleared only on replica stop --
@@ -214,8 +211,6 @@ def register(self, info: DeploymentInfo, curr_target_num_replicas: int) -> int:
     def on_replica_stopped(self, replica_id: ReplicaID):
         if replica_id in self._replica_metrics:
             del self._replica_metrics[replica_id]
-        self._replica_running_arrays.pop(replica_id, None)
-        self._replica_custom_arrays.pop(replica_id, None)
         self._replica_report_ts.pop(replica_id, None)
 
     def get_num_replicas_lower_bound(self) -> int:
@@ -293,68 +288,18 @@ def record_request_metrics_for_replica(
             self._replica_report_ts[replica_id] = send_timestamp
             # dedup-at-write: this source now reports via cloudpickle; drop any
             # columnar entries so the stores never double-count it.
-            self._replica_running_arrays.pop(replica_id, None)
-            self._replica_custom_arrays.pop(replica_id, None)
-
-    def record_columnar_metrics_for_replica(
-        self, replica_id, metric_arrays, timestamp
-    ) -> None:
-        """Store columnar per-metric arrays for a replica (no per-point objects).
-        running_requests feeds the hot-path store; any other metrics feed the custom
-        store used by custom autoscaling policies (the columnar decode is lossless)."""
-        prev_ts = self._replica_report_ts.get(replica_id)
-        if prev_ts is not None and timestamp <= prev_ts:
-            return
-        self._replica_report_ts[replica_id] = timestamp
-        running = metric_arrays.get(RUNNING_REQUESTS_KEY)
-        if running is not None:
-            self._replica_running_arrays[replica_id] = (
-                running[0],
-                running[1],
-                timestamp,
-            )
-        else:
-            # A newer report that omits running_requests must drop the stale running
-            # timeseries -- the object path replaces the whole report, so missing
-            # running stops contributing there too.
-            self._replica_running_arrays.pop(replica_id, None)
-        custom = {m: a for m, a in metric_arrays.items() if m != RUNNING_REQUESTS_KEY}
-        if custom:
-            self._replica_custom_arrays[replica_id] = custom
-        else:
-            self._replica_custom_arrays.pop(replica_id, None)
-        # dedup-at-write: drop any cloudpickle entry for this source.
-        self._replica_metrics.pop(replica_id, None)
 
     def _columnar_aggregate_total_requests(self) -> float:
         """Aggregate-mode total over pure-columnar stores: replica (direct-ingress)
         running arrays when a RUNNING replica reported, else handle running arrays,
         plus queued -- one fused numpy merge (no per-replica Python objects)."""
-        # Gate on whether a RUNNING replica actually reported, NOT on the store being
-        # non-empty: a lingering stopped-replica array (before on_replica_stopped
-        # clears it) must fall through to handle-running exactly like the object
-        # path, else handle-collected running is dropped (total reads queued-only).
-        replica_segments = self._replica_columnar_segments()
-        if replica_segments:
-            return self._aggregate_segments(
-                replica_segments + self._queued_columnar_segments()
-            )
         if not self._handle_arrays:
             return 0.0
         return self._aggregate_segments(
             self._handle_running_columnar_segments(self._cached_running_replica_strs)
             + self._queued_columnar_segments()
         )
 
-    def _replica_columnar_segments(self):
-        """RUNNING replicas' columnar running-request arrays as (ts, val) segments."""
-        segs = []
-        for replica_id in self._running_replicas:
-            a = self._replica_running_arrays.get(replica_id)
-            if a is not None and a[0].size:
-                segs.append((a[0], a[1]))
-        return segs
-
     def _queued_columnar_segments(self):
         """Columnar per-handle queued arrays as (ts, val) segments."""
         return [
@@ -811,7 +756,7 @@ def _calculate_total_requests_aggregate_mode(self) -> float:
             Total number of requests (average running + queued) calculated from
             timeseries data aggregation.
         """
-        has_columnar = bool(self._replica_running_arrays or self._handle_arrays)
+        has_columnar = bool(self._handle_arrays)
         has_object = bool(self._replica_metrics or self._handle_requests)
         # Homogeneous fleets keep their native fast path. Columnar arrays are used
         # whenever present -- the controller wire-detects the format from the frame
@@ -862,8 +807,7 @@ def _mixed_aggregate_total_requests(self) -> float:
         are converted to small arrays. Empty object series are dropped so they
         cannot flip metrics_collected_on_replicas and suppress handle-side running
         (mirrors the columnar empty-skip). Disjoint by dedup-at-write."""
-        segments = self._replica_columnar_segments()
-        segments += self._series_segments(self._collect_replica_running_requests())
+        segments = self._series_segments(self._collect_replica_running_requests())
         metrics_collected_on_replicas = bool(segments)
         if not metrics_collected_on_replicas:
             segments += self._handle_running_columnar_segments(
@@ -931,8 +875,6 @@ def _get_aggregated_custom_metrics(self) -> Dict[str, Dict[ReplicaID, float]]:
             Dict mapping metric name to dict of replica ID to aggregated metric value.
         """
         aggregated_metrics: Dict[str, Dict[ReplicaID, float]] = defaultdict(dict)
-        now = time.time()
-        agg = self._config.aggregation_function
         for replica_id in self._running_replicas:
             # A replica is in the object store OR the columnar stores (dedup-at-write).
             replica_metric_report = self._replica_metrics.get(replica_id)
@@ -942,18 +884,6 @@ def _get_aggregated_custom_metrics(self) -> Dict[str, Dict[ReplicaID, float]]:
                         replica_id
                     ] = self._merge_and_aggregate_timeseries([timeseries])
                 continue
-            running = self._replica_running_arrays.get(replica_id)
-            if running is not None and running[0].size:
-                aggregated_metrics[RUNNING_REQUESTS_KEY][
-                    replica_id
-                ] = self._aggregate_single_array(running[0], running[1], now, agg)
-            custom = self._replica_custom_arrays.get(replica_id)
-            if custom:
-                for metric_name, (ts, val) in custom.items():
-                    if ts.size:
-                        aggregated_metrics[metric_name][
-                            replica_id
-                        ] = self._aggregate_single_array(ts, val, now, agg)
         return dict(aggregated_metrics)
 
     def _get_raw_custom_metrics(
@@ -971,19 +901,6 @@ def _get_raw_custom_metrics(
                 for metric_name, timeseries in replica_metric_report.metrics.items():
                     raw_metrics[metric_name][replica_id] = timeseries
                 continue
-            running = self._replica_running_arrays.get(replica_id)
-            if running is not None and running[0].size:
-                raw_metrics[RUNNING_REQUESTS_KEY][replica_id] = [
-                    TimeStampedValue(float(running[0][k]), float(running[1][k]))
-                    for k in range(running[0].size)
-                ]
-            custom = self._replica_custom_arrays.get(replica_id)
-            if custom:
-                for metric_name, (ts, val) in custom.items():
-                    raw_metrics[metric_name][replica_id] = [
-                        TimeStampedValue(float(ts[k]), float(val[k]))
-                        for k in range(ts.size)
-                    ]
         return dict(raw_metrics)
 
 
@@ -1236,15 +1153,6 @@ def record_request_metrics_for_replica(
                 dep_id
             ].record_request_metrics_for_replica(replica_metric_report)
 
-    def record_columnar_metrics_for_replica(
-        self, replica_id, metric_arrays, timestamp
-    ) -> None:
-        dep_id = replica_id.deployment_id
-        if dep_id in self._deployment_autoscaling_states:
-            self._deployment_autoscaling_states[
-                dep_id
-            ].record_columnar_metrics_for_replica(replica_id, metric_arrays, timestamp)
-
     def record_request_metrics_for_handle(
         self, handle_metric_report: HandleMetricReport
     ):
@@ -1441,15 +1349,6 @@ def record_request_metrics_for_replica(
         if app_state:
             app_state.record_request_metrics_for_replica(replica_metric_report)
 
-    def record_columnar_metrics_for_replica(
-        self, replica_id, metric_arrays, timestamp
-    ) -> None:
-        app_state = self._app_autoscaling_states.get(replica_id.deployment_id.app_name)
-        if app_state:
-            app_state.record_columnar_metrics_for_replica(
-                replica_id, metric_arrays, timestamp
-            )
-
     def record_request_metrics_for_handle(
         self,
         handle_metric_report: HandleMetricReport,

python/ray/serve/_private/controller.py

@@ -411,35 +411,6 @@ def record_autoscaling_metrics_from_replica(
     ):
         _ingest_start = time.time()
         if isinstance(replica_metric_report, bytes):
-            if autoscaling_metrics_codec.is_columnar(replica_metric_report):
-                if not autoscaling_metrics_codec.can_decode_columnar():
-                    # Wire-detect works without numpy but decoding does not.
-                    autoscaling_metrics_codec.warn_columnar_undecodable_once()
-                    return
-                _decode_start = time.time()
-                (
-                    replica_id,
-                    metric_arrays,
-                    report_ts,
-                ) = autoscaling_metrics_codec.decode_replica_all_metrics(
-                    replica_metric_report
-                )
-                self._health_metrics_tracker.record_decompress(
-                    (time.time() - _decode_start) * 1000
-                )
-                self._record_metrics_delay(
-                    report_ts,
-                    replica_id.deployment_id,
-                    self.replica_metrics_delay_histogram,
-                    self._health_metrics_tracker.record_replica_metrics_delay,
-                )
-                self.autoscaling_state_manager.record_columnar_metrics_for_replica(
-                    replica_id, metric_arrays, report_ts
-                )
-                self._health_metrics_tracker.record_replica_ingest(
-                    (time.time() - _ingest_start) * 1000
-                )
-                return
             _decompress_start = time.time()
             replica_metric_report = decompress_metric_report(replica_metric_report)
             self._health_metrics_tracker.record_decompress(

python/ray/serve/_private/replica.py

@@ -53,7 +53,6 @@
 from ray.dag.py_obj_scanner import _PyObjScanner
 from ray.remote_function import RemoteFunction
 from ray.serve import metrics
-from ray.serve._private import autoscaling_metrics_codec
 from ray.serve._private.common import (
     RUNNING_REQUESTS_KEY,
     DeploymentID,
@@ -978,11 +977,7 @@ def _push_autoscaling_metrics(self) -> None:
             self._pending_metrics_push_ref = (
                 # Actor methods are resolved dynamically on the actor handle.
                 self._controller_handle.record_autoscaling_metrics_from_replica.remote(  # type: ignore[attr-defined]
-                    autoscaling_metrics_codec.encode(replica_metric_report)
-                    if autoscaling_metrics_codec.should_encode_columnar(
-                        replica_metric_report
-                    )
-                    else compress_metric_report(replica_metric_report)
+                    compress_metric_report(replica_metric_report)
                 )
             )