Function object_collection_sql 

fn object_collection_sql(
    cluster_id: ClusterId,
    replica_id: ReplicaId,
    cutoff: &str,
) -> String

The rows this replica has completed that the history table is missing.

Aggregates every worker’s row for an export, and records nothing until all of them have hydrated. One worker is not enough, because a materialized view’s persist sink has a single active worker, hash(sink_id) % workers. Only that worker’s reported output frontier is gated on the shard upper, so only its hydrated_at covers the initial snapshot write. Every other worker clears its sink write frontier and stamps at compute completion, which for a materialized view is before the data is durable. Taking max over a complete set of workers is therefore the only way to get a finish that means the same thing for every object, and it is the rule mz_compute_hydration_times already applies.

Completeness needs no configured worker count. The log carries a row per (export_id, worker_id) from installation with a null hydrated_at, so count(*) = count(hydrated_at) says every row visible at the OCC read timestamp has finished. Per-process logging clocks also determine Differential update timestamps, so a worker whose clock is ahead can be absent at that timestamp. A visible unfinished object is skipped and picked up by a later sweep.

A worker missing at the read timestamp cannot later change the episode key. Its logging clock stamps both the Differential update and installed_at, so late visibility means its installation stamp is later than the visible minimum. The anti-join therefore keeps matching the recorded row. A later hydrated_at can raise the aggregate’s maximum, but history is not repaired after the episode key has been recorded.

The collector deliberately accepts this sampling race rather than depending on ReplicaLocation::workers(). A durable finish can therefore precede the latest worker’s finish. A whole-replica restart resets the collection as a unit.

The interval spans workers, so it carries whatever skew there is between the process clocks that stamped its ends. Each process anchors its logging clock at its own SystemTime. That inflates a duration, and nothing here rejects a row for being inconsistent, which is deliberate: an ordering guard on cross-worker stamps rejects complete episodes permanently, since the log values never change.

Collection has no explicit batch bound. It returns at most one row per not-yet-recorded dataflow, and the OCC path rejects a result that exceeds max_result_size or max_query_result_size. At their 1 GiB defaults that ceiling only matters at millions of dataflows per replica.