cluster-controller: batch ON REFRESH oracle reads

[aljoscha]

Motivation

The cluster controller fetched refresh-window inputs one scheduled cluster at a time. Each fetch issued and awaited a timestamp-oracle read. With a slow oracle, every ON REFRESH cluster added another round trip before the controller could reconcile unrelated clusters.

Description

Gather each scheduled cluster's catalog and storage inputs in a separate coordinator turn, then perform one shared timestamp-oracle read for the completed reconciliation-phase batch. This keeps oracle latency constant in scheduled-cluster count without creating one long coordinator-loop scan across every bound materialized view.

The batch contract carries one top-level timestamp and timestamp-free per-cluster inputs. A cluster whose required inputs are unavailable is skipped for that phase instead of being interpreted as outside its refresh window.

The adapter guide now records that timestamp-oracle coalescing is opportunistic, serial awaits cannot coalesce, and one explicit shared read is valid only when it satisfies every caller's real-time and contract requirements.

Verification

Adds controller seam coverage for one refresh batch per phase and for partial and wholly unavailable batches that leave affected clusters untouched while other clusters continue reconciling. Adds an mzcompose regression that binds a periodic-refresh materialized view to each of eight scheduled clusters, waits for every cluster to have exactly one replica, injects oracle latency, and compares an unrelated probe cluster's convergence with zero and eight scheduled clusters.

Fixes SQL-569.

[def-]

Thanks for adding the test!

[mtabebe]

Thanks for the fix. I'm curious was this always planned or discovered later through performance testing. Should we update our adapter guide to hint at using some batching for timestamp oracle calls for future claude 🤖 ?

[def-]

Actually the test needs some improvement so it actually fails when refresh is skipped wrongly:

diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py
index 6cecb9a934..26af770fe8 100644
--- a/test/cluster/mzcompose.py
+++ b/test/cluster/mzcompose.py
@@ -7823,6 +7823,27 @@ def workflow_test_controller_oracle_stall(
                 replication_factor = 1 - replication_factor
             return samples

+        def await_scheduled_replicas() -> None:
+            start = time.monotonic()
+            while True:
+                mz.execute(
+                    "SELECT count(DISTINCT c.id), count(r.id) "
+                    "FROM mz_clusters c JOIN mz_cluster_replicas r "
+                    "ON c.id = r.cluster_id WHERE c.name LIKE 'cc_sched%'"
+                )
+                clusters_with_replicas, replica_count = mz.fetchall()[0]
+                if (
+                    clusters_with_replicas == args.scheduled_clusters
+                    and replica_count == args.scheduled_clusters
+                ):
+                    return
+                assert time.monotonic() - start < 120, (
+                    "refresh windows never created exactly one replica for every "
+                    f"scheduled cluster: {clusters_with_replicas} clusters have "
+                    f"{replica_count} replicas, expected {args.scheduled_clusters}"
+                )
+                time.sleep(0.05)
+
         set_latency(toxi, 0)
         mz.execute(
             "CREATE CLUSTER cc_probe "
@@ -7846,14 +7867,27 @@ def workflow_test_controller_oracle_stall(
         )

         converge_ms(0)
+        mz.execute("CREATE TABLE cc_sched_t (x int)")
         for i in range(args.scheduled_clusters):
+            cluster_name = sql.Identifier(f"cc_sched{i}")
             mz.execute(
                 sql.SQL(
                     "CREATE CLUSTER {} (SIZE 'scale=1,workers=1', "
                     "SCHEDULE = ON REFRESH "
                     "(HYDRATION TIME ESTIMATE = '60 seconds'))"
-                ).format(sql.Identifier(f"cc_sched{i}"))
+                ).format(cluster_name)
+            )
+            mz.execute(
+                sql.SQL(
+                    "CREATE MATERIALIZED VIEW {} IN CLUSTER {} "
+                    "WITH (REFRESH = EVERY '1 second') "
+                    "AS SELECT count(*) FROM cc_sched_t"
+                ).format(
+                    sql.Identifier(f"cc_sched{i}_mv"),
+                    cluster_name,
+                )
             )
+        await_scheduled_replicas()

         set_latency(toxi, args.latency_ms)
         stalled_samples = convergence_samples()

(Still green on this PR)

[aljoscha]

Thanks for the fix. I'm curious was this always planned or discovered later through performance testing. Should we update our adapter guide to hint at using some batching for timestamp oracle calls for future claude 🤖 ?

it's something Dennis found in testing! so I might add to the guide in this PR

[aljoscha]

AJ (AI coding agent): Addressed both follow-ups.

@def- Good catch. The timing test now binds a REFRESH EVERY '1 second' materialized view to every scheduled cluster and waits until each cluster has exactly one replica before enabling oracle latency. Skipped refresh processing now fails during setup instead of leaving the latency assertion green.

@mtabebe This was known debt rather than something first discovered by the SQL-569 performance test. During review of #37767, the sequential per-cluster reads were identified as impossible for BatchingTimestampOracle to coalesce, and the implementation left a TODO to hoist the read. SQL-569 implements that work. I also updated the adapter guide to explain that coalescing is opportunistic, serial awaits cannot coalesce, and an explicit shared read is valid only when every caller's real-time bounds and contract allow it.

I rebased the change over current main, including the removal of the legacy scheduler and controller gate.