[iris] Push IrisClient.list_jobs filters into server-side JobQuery

Previously IrisClient.list_jobs() fetched every job in the cluster (paging
offset=0, 500, 1000, ... through the full table) and filtered client-side,
so `iris job list --state running` scanned the entire jobs table on every
invocation. Translate `state` and `prefix` into JobQuery.state_filter and
JobQuery.name_filter so the controller prunes in SQL. Prefix is re-checked
client-side because name_filter is a substring, not an anchored prefix.
Collapses `states: list` to `state:` since no caller passed more than one.

Author: rjpower

lib/iris/src/iris/cli/job.py

@@ -953,16 +953,16 @@ def list_jobs(ctx, state: str | None, prefix: str | None, json_output: bool) ->
     controller_url = require_controller_url(ctx)
     client = IrisClient.remote(controller_url, workspace=Path.cwd(), token_provider=ctx.obj.get("token_provider"))
 
-    states: list[job_pb2.JobState] | None = None
+    state_value: job_pb2.JobState | None = None
     if state is not None:
         state_lower = state.lower()
         if state_lower not in _STATE_MAP:
             valid = ", ".join(sorted(_STATE_MAP.keys()))
             raise click.UsageError(f"Unknown state '{state}'. Valid states: {valid}")
-        states = [_STATE_MAP[state_lower]]
+        state_value = _STATE_MAP[state_lower]
 
     prefix_name = JobName.from_wire(prefix) if prefix else None
-    jobs = client.list_jobs(states=states, prefix=prefix_name)
+    jobs = client.list_jobs(state=state_value, prefix=prefix_name)
 
     # Sort by submitted_at descending (most recent first)
     jobs.sort(key=lambda j: j.submitted_at.epoch_ms, reverse=True)

lib/iris/src/iris/client/client.py

@@ -51,7 +51,8 @@
     TaskAttempt,
     adjust_tpu_replicas,
 )
-from iris.rpc import job_pb2
+from iris.rpc import controller_pb2, job_pb2
+from iris.rpc.proto_utils import job_state_friendly
 from iris.time_proto import timestamp_from_proto
 from rigging.timing import Duration, Timestamp
 
@@ -724,28 +725,36 @@ def terminate(self, job_id: JobName) -> None:
     def list_jobs(
         self,
         *,
-        states: list[job_pb2.JobState] | None = None,
+        state: job_pb2.JobState | None = None,
         prefix: JobName | None = None,
     ) -> list[job_pb2.JobStatus]:
         """List jobs with optional filtering.
 
+        Filters are pushed down to the server via ``JobQuery`` so the
+        controller does not page-walk its entire jobs table: ``state`` becomes
+        ``state_filter`` and ``prefix`` becomes a ``name_filter`` substring
+        match. The prefix is re-validated client-side because ``name_filter``
+        is a substring, not an anchored prefix.
+
         Args:
-            states: If provided, only return jobs in these states
+            state: If provided, only return jobs in this state
             prefix: If provided, only return jobs whose JobName starts with this prefix
 
         Returns:
             List of JobStatus matching the filters
         """
-        all_jobs = self._cluster_client.list_jobs()
-        result = []
-        for job in all_jobs:
-            if states is not None and job.state not in states:
-                continue
-            job_name = JobName.from_wire(job.job_id)
-            if prefix is not None and not job_name.to_wire().startswith(prefix.to_wire()):
-                continue
-            result.append(job)
-        return result
+        query = controller_pb2.Controller.JobQuery()
+        if state is not None:
+            query.state_filter = job_state_friendly(state)
+        if prefix is not None:
+            query.name_filter = prefix.to_wire()
+
+        all_jobs = self._cluster_client.list_jobs(query=query)
+        if prefix is None:
+            return list(all_jobs)
+
+        prefix_wire = prefix.to_wire()
+        return [job for job in all_jobs if JobName.from_wire(job.job_id).to_wire().startswith(prefix_wire)]
 
     def terminate_prefix(
         self,

lib/iris/src/iris/cluster/client/protocol.py

@@ -80,7 +80,12 @@ def list_endpoints(self, prefix: str, *, exact: bool = False) -> list[controller
 
     def list_workers(self) -> list[controller_pb2.Controller.WorkerHealthStatus]: ...
 
-    def list_jobs(self) -> list[job_pb2.JobStatus]: ...
+    def list_jobs(
+        self,
+        *,
+        query: controller_pb2.Controller.JobQuery | None = None,
+        page_size: int = 500,
+    ) -> list[job_pb2.JobStatus]: ...
 
     def get_task_status(self, task_name: JobName) -> job_pb2.TaskStatus: ...
 

lib/iris/tests/client/test_list_jobs_pushdown.py

@@ -0,0 +1,113 @@
+# Copyright The Marin Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Verify IrisClient.list_jobs pushes filters into the server JobQuery.
+
+Server-side filtering matters: without it, a simple
+``iris job list --state running`` triggers a full-table paginated scan
+(offset=0, 500, 1000, ...) which is expensive on large deployments.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+
+from iris.client import IrisClient
+from iris.cluster.types import JobName
+from iris.rpc import controller_pb2, job_pb2
+
+
+@dataclass
+class _RecordingClusterClient:
+    """Captures the JobQuery passed into list_jobs; returns canned jobs."""
+
+    jobs: list[job_pb2.JobStatus] = field(default_factory=list)
+    captured_queries: list[controller_pb2.Controller.JobQuery] = field(default_factory=list)
+
+    def list_jobs(
+        self,
+        *,
+        query: controller_pb2.Controller.JobQuery | None = None,
+        page_size: int = 500,
+    ) -> list[job_pb2.JobStatus]:
+        del page_size
+        captured = controller_pb2.Controller.JobQuery()
+        if query is not None:
+            captured.CopyFrom(query)
+        self.captured_queries.append(captured)
+        return list(self.jobs)
+
+    def shutdown(self, wait: bool = True) -> None:
+        del wait
+
+    def __getattr__(self, name: str) -> Any:
+        raise AttributeError(name)
+
+
+def _make_status(job_id: str, state: job_pb2.JobState) -> job_pb2.JobStatus:
+    status = job_pb2.JobStatus()
+    status.job_id = job_id
+    status.state = state
+    return status
+
+
+@pytest.fixture
+def stub_client() -> tuple[IrisClient, _RecordingClusterClient]:
+    stub = _RecordingClusterClient()
+    client = IrisClient(cluster=stub)
+    return client, stub
+
+
+def test_list_jobs_no_filter_sends_empty_query(stub_client):
+    client, stub = stub_client
+    stub.jobs = [_make_status("/u/a", job_pb2.JOB_STATE_RUNNING)]
+
+    client.list_jobs()
+
+    assert len(stub.captured_queries) == 1
+    q = stub.captured_queries[0]
+    assert q.state_filter == ""
+    assert q.name_filter == ""
+
+
+def test_list_jobs_state_is_pushed_down(stub_client):
+    client, stub = stub_client
+    stub.jobs = [_make_status("/u/a", job_pb2.JOB_STATE_RUNNING)]
+
+    client.list_jobs(state=job_pb2.JOB_STATE_RUNNING)
+
+    assert stub.captured_queries[0].state_filter == "running"
+
+
+def test_list_jobs_prefix_is_pushed_down(stub_client):
+    client, stub = stub_client
+    prefix = JobName.root("alice", "exp")
+    stub.jobs = [_make_status(prefix.to_wire() + "-1", job_pb2.JOB_STATE_PENDING)]
+
+    client.list_jobs(prefix=prefix)
+
+    assert stub.captured_queries[0].name_filter == prefix.to_wire()
+
+
+def test_list_jobs_reanchors_prefix_client_side(stub_client):
+    """name_filter is a substring; client must still enforce startswith.
+
+    `/bob/alice/exp-oops` contains `/alice/exp` as a substring so it
+    slips through the server-side LIKE '%...%' filter. The client must
+    drop it because it is not a true prefix of `/alice/exp`.
+    """
+    client, stub = stub_client
+    prefix = JobName.root("alice", "exp")
+    wire = prefix.to_wire()
+    stub.jobs = [
+        _make_status(wire, job_pb2.JOB_STATE_RUNNING),
+        _make_status(wire + "-child", job_pb2.JOB_STATE_RUNNING),
+        _make_status("/bob/alice/exp-oops", job_pb2.JOB_STATE_RUNNING),
+    ]
+
+    result = client.list_jobs(prefix=prefix)
+
+    assert {j.job_id for j in result} == {wire, wire + "-child"}