Struct PeekClient 

pub struct PeekClient {

    coordinator_client: CoordinatorClient,
    catalog_cache: Weak<Catalog>,
    compute_instances: BTreeMap<ComputeInstanceId, InstanceClient>,
    pub storage_collections: StorageCollectionsHandle,
    pub transient_id_gen: Arc<TransientIdGen>,
    pub optimizer_metrics: OptimizerMetrics,
    oracles: BTreeMap<Timeline, Arc<dyn TimestampOracle<Timestamp> + Send + Sync>>,
    persist_client: PersistClient,
    pub statement_logging_frontend: StatementLoggingFrontend,
    pub occ_write_semaphore: Arc<Semaphore>,
    pub frontend_read_then_write_enabled: bool,
    pub(crate) group_commit_notifier: GroupCommitNotifier,
    pub read_only: bool,
}

Clients needed for peek sequencing in the Adapter Frontend.

Fields

coordinator_client: CoordinatorClient

catalog_cache: Weak<Catalog>

Cache of the latest catalog snapshot. Serves PeekClient::catalog_snapshot without a Coordinator round-trip while the catalog’s transient revision is unchanged.

Holds a Weak so that an idle session does not keep a superseded catalog version alive.

compute_instances: BTreeMap<ComputeInstanceId, InstanceClient>

Channels to talk to each compute Instance task directly. Lazily populated. Note that these are never cleaned up. In theory, this could lead to a very slow memory leak if a long-running user session keeps peeking on clusters that are being created and dropped in a hot loop. Hopefully this won’t occur any time soon.

storage_collections: StorageCollectionsHandle

Handle to storage collections for reading frontiers and policies.

transient_id_gen: Arc<TransientIdGen>

A generator for transient GlobalIds, shared with Coordinator.

optimizer_metrics: OptimizerMetrics

oracles: BTreeMap<Timeline, Arc<dyn TimestampOracle<Timestamp> + Send + Sync>>

Per-timeline oracles from the coordinator. Lazily populated.

persist_client: PersistClient

statement_logging_frontend: StatementLoggingFrontend

Statement logging state for frontend peek sequencing.

occ_write_semaphore: Arc<Semaphore>

Semaphore for limiting concurrent OCC (optimistic concurrency control) write operations.

frontend_read_then_write_enabled: bool

Whether frontend OCC read-then-write is enabled (determined once at process startup).

group_commit_notifier: GroupCommitNotifier

Requests a group commit. Used to advance the write timeline when we need the oracle to move but have nothing to write ourselves.

read_only: bool

Whether the coordinator is in read-only mode. Mutations must be rejected.

Implementations

impl PeekClient

pub(crate) async fn try_frontend_peek( &mut self, portal_name: &str, session: &mut Session, logging: &mut ExecutionLogging, ) -> Result<Option<ExecuteResponse>, AdapterError>

Attempt to sequence a peek from the session task.

Returns Ok(Some(response)) if we handled the peek, or Ok(None) to fall back to the Coordinator’s sequencing. If it returns an error, it should be returned to the user.

logging holds the end-of-execution obligation for this statement. The caller retires it, this function only takes the statement over and, at the dispatch sites that hand execution off, defuses the slot.

async fn try_frontend_peek_inner( &mut self, session: &mut Session, catalog: Arc<Catalog>, stmt: Option<Arc<Statement<Raw>>>, params: Params, logging: &mut ExecutionLogging, ) -> Result<Option<ExecuteResponse>, AdapterError>

This is encapsulated in an inner function so that the outer function can still do statement logging after the ? returns of the inner function.

Dispatch sites that hand the statement to the coordinator for asynchronous completion (registered peeks, subscribes) defuse the logging slot at the point where the coordinator takes over. Everywhere else the slot stays armed and SessionClient::execute logs the end from the returned result.

pub(crate) async fn frontend_determine_timestamp( &mut self, session: &Session, id_bundle: &CollectionIdBundle, when: &QueryWhen, compute_instance: ComputeInstanceId, timeline_context: &TimelineContext, oracle_read_ts: Option<Timestamp>, real_time_recency_ts: Option<Timestamp>, ) -> Result<(TimestampDetermination, ReadHolds), AdapterError>

(Similar to Coordinator::determine_timestamp) Determines the timestamp for a query, acquires read holds that ensure the query remains executable at that time, and returns those. The caller is responsible for eventually dropping those read holds.

Note: self is taken &mut because of the lazy fetching in get_compute_instance_client.

fn assert_read_holds_correct( read_holds: &ReadHolds, execution: &Execution, determination: &TimestampDetermination, target_cluster_id: ClusterId, in_immediate_multi_stmt_txn: bool, )

impl PeekClient

pub(crate) async fn frontend_read_then_write( &mut self, session: &mut Session, plan: ReadThenWritePlan, target_cluster: TargetCluster, catalog: &Arc<Catalog>, statement_logging_id: Option<StatementLoggingId>, attempt_state: Arc<FrontendWriteAttemptState>, ) -> Result<ExecuteResponse, AdapterError>

Execute a read-then-write operation using frontend sequencing.

Called by session code when the frontend_read_then_write dyncfg is enabled. The caller owns the end-of-execution logging for statement_logging_id and verified and planned the portal against catalog, which stays in force through optimization and write-target generation capture.

pub(crate) async fn background_read_then_write( &mut self, session: &mut Session, plan: ReadThenWritePlan, cluster_id: ComputeInstanceId, replica_id: ReplicaId, catalog: &Arc<Catalog>, ) -> Result<ExecuteResponse, AdapterError>

Executes a coordinator-owned read-then-write against system relations, pinned to replica_id.

See RtwCaller::Background for what the caller takes on by using this.

async fn read_then_write( &mut self, session: &mut Session, plan: ReadThenWritePlan, target_cluster: TargetCluster, catalog: &Arc<Catalog>, statement_logging_id: Option<StatementLoggingId>, attempt_state: Arc<FrontendWriteAttemptState>, caller: RtwCaller, ) -> Result<ExecuteResponse, AdapterError>

fn optimize_mir_read_then_write( &self, catalog: &Arc<Catalog>, session: &dyn SessionMetadata, plan: &ReadThenWritePlan, cluster_id: ComputeInstanceId, ) -> Result<(Optimizer, GlobalMirPlan<Unresolved>), AdapterError>

Builds the subscribe optimizer and the unresolved global MIR plan for a read-then-write.

The optimized expression is the selection with the mutation already applied, so the subscribe’s sink emits ready-to-write table diffs rather than query results. finishing and returning are deliberately not part of the dataflow, and unmaterializable functions are prepared one-shot.

async fn oracle_read_ts( &mut self, timeline: &TimelineContext, ) -> Result<Option<Timestamp>, AdapterError>

The governing oracle’s read timestamp, used as a lower bound for timestamp selection. None when the selection needs no oracle.

async fn ensure_read_linearized( &mut self, timeline: &TimelineContext, as_of: Timestamp, ) -> Result<(), AdapterError>

Block until the oracle for this query’s timeline has advanced to as_of. Returns immediately if it already has.

This implements the strict-serializable read guarantee for RTW: once this returns, any session observing the oracle sees a read timestamp at least as large as as_of, so reads at as_of (and writes derived from them) cannot appear to “go backwards” relative to subsequent queries.

async fn submit_blind_write( &self, conn_id: ConnectionId, target_id: CatalogItemId, target_global_id: GlobalId, diffs: Vec<(Row, Diff)>, statement_logging_id: Option<StatementLoggingId>, attempt_state: &FrontendWriteAttemptState, ) -> Result<Timestamp, AdapterError>

Submits frontier-independent diffs to group commit, which picks the write timestamp, and returns the timestamp the write committed at.

Only valid for diffs that do not depend on an observed read frontier: the write lands at a timestamp this caller does not choose.

async fn create_internal_subscribe( &self, df_desc: Box<DataflowDescription<LirRelationExpr>>, cluster_id: ComputeInstanceId, replica_id: Option<ReplicaId>, depends_on: BTreeSet<GlobalId>, as_of: Timestamp, arity: usize, sink_id: GlobalId, owner: ActiveSubscribeOwner, start_time: EpochMillis, read_holds: ReadHolds, ) -> Result<SubscribeHandle, AdapterError>

Creates an internal subscribe, meaning one that writes no mz_subscriptions row. Returns a SubscribeHandle that ensures cleanup on drop.

async fn run_occ_loop( &self, subscribe_handle: SubscribeHandle, target_id: CatalogItemId, target_global_id: GlobalId, kind: MutationKind, returning: Vec<MirScalarExpr>, max_result_size: u64, max_query_result_size: u64, row_set_finishing_seconds: Histogram, max_occ_retries: usize, table_desc: RelationDesc, write_conn_id: Option<ConnectionId>, statement_logging_id: Option<StatementLoggingId>, as_of: Timestamp, write_oracle: Option<Arc<dyn TimestampOracle<Timestamp> + Send + Sync>>, attempt_state: &FrontendWriteAttemptState, ) -> (usize, Result<OccOutcome, AdapterError>)

Run the OCC loop: drain the subscribe at as_of, apply the mutation, and submit the resulting diffs as a write.

Semantically a SELECT at target - 1 followed by an INSERT at target. write_oracle chooses target, the subscribe’s frontier certifies the payload is complete below it, and a target the target table has moved past comes back as WriteResult::TimestampPassed, whose next eligible timestamp the loop adopts. At most max_occ_retries attempts.

A subscribe that ends on its own has diffs no frontier can change, and those are returned as OccOutcome::Blind rather than written.

Contract on the caller, both ends of the read: the oracle’s read timestamp must be at or above as_of on entry, and an OccOutcome::NoRowsMatched must be linearized against its empty_as_of before the response goes out.

write_oracle is None only for a timestamp-independent selection. Such a statement reads at Timestamp::maximum(), so it observes no progress past its as_of and always leaves through the blind path.

Returns (retry_count, result) so the caller can record OCC retry metrics regardless of whether the operation succeeded or failed.

impl PeekClient

pub(crate) fn new( coordinator_client: CoordinatorClient, catalog: &Arc<Catalog>, storage_collections: StorageCollectionsHandle, transient_id_gen: Arc<TransientIdGen>, optimizer_metrics: OptimizerMetrics, persist_client: PersistClient, statement_logging_frontend: StatementLoggingFrontend, occ_write_semaphore: Arc<Semaphore>, frontend_read_then_write_enabled: bool, group_commit_notifier: GroupCommitNotifier, read_only: bool, ) -> Self

Creates a PeekClient.

catalog seeds the catalog snapshot cache, so that the session’s first statements don’t need a Command::CatalogSnapshot round-trip.

pub async fn ensure_compute_instance_client( &mut self, compute_instance: ComputeInstanceId, ) -> Result<InstanceClient, CollectionLookupError>

pub async fn ensure_oracle( &mut self, timeline: Timeline, ) -> Result<&mut Arc<dyn TimestampOracle<Timestamp> + Send + Sync>, AdapterError>

pub async fn catalog_snapshot(&mut self, context: &str) -> Arc<Catalog>

Fetch a snapshot of the catalog.

Serves from the session-side cache when the catalog’s transient revision is unchanged since the cached snapshot was taken (see Catalog::transient_revision_is_current). An unchanged revision means the cached snapshot is identical to what a fresh fetch would return. Otherwise falls back to a Command::CatalogSnapshot round-trip and re-populates the cache.

Cache misses record the round-trip time in the adapter metrics, labeled by context. Hits and misses are counted in catalog_snapshot_cache.

pub(crate) async fn call_coordinator<T, F>( &self, f: F, ) -> Result<T, AdapterError>

where F: FnOnce(Sender<T>) -> Command,

Calls the coordinator and returns an error if it drops the response.

pub(crate) fn coordinator_client(&self) -> &CoordinatorClient

The client for sending commands to the coordinator.

pub async fn acquire_read_holds_and_least_valid_write( &mut self, id_bundle: &CollectionIdBundle, ) -> Result<(ReadHolds, Antichain<Timestamp>), CollectionLookupError>

Acquire read holds on the given compute/storage collections, and determine the smallest common valid write frontier among the specified collections.

Similar to Coordinator::acquire_read_holds and TimestampProvider::least_valid_write combined.

Note: Unlike the Coordinator/StorageController’s least_valid_write that treats sinks specially when fetching storage frontiers (see mz_storage_controller::collections_frontiers), we intentionally do not special‑case sinks here because peeks never read from sinks. Therefore, using StorageCollections::collections_frontiers is sufficient.

Note: self is taken &mut because of the lazy fetching in get_compute_instance_client.

pub(crate) async fn implement_fast_path_peek_plan( &mut self, fast_path: FastPathPlan, timestamp: Timestamp, finishing: RowSetFinishing, compute_instance: ComputeInstanceId, target_replica: Option<ReplicaId>, intermediate_result_type: SqlRelationType, max_result_size: u64, max_returned_query_size: Option<u64>, row_set_finishing_seconds: Histogram, input_read_holds: ReadHolds, peek_stash_read_batch_size_bytes: usize, peek_stash_read_memory_budget_bytes: usize, conn_id: ConnectionId, depends_on: BTreeSet<GlobalId>, watch_set: Option<WatchSetCreation>, logging: &mut ExecutionLogging, ) -> Result<ExecuteResponse, AdapterError>

Implement a fast-path peek plan. This is similar to Coordinator::implement_peek_plan, but only for fast path peeks.

Note: self is taken &mut because of the lazy fetching in get_compute_instance_client.

Note: input_read_holds has holds for all inputs. For fast-path peeks, this includes the peek target. For slow-path peeks (to be implemented later), we’ll need to additionally call into the Controller to acquire a hold on the peek target after we create the dataflow.

For a constant peek the logging slot stays armed and the caller logs the end from the returned result. For a PeekExisting/PeekPersist peek, successful registration with the coordinator hands ownership of the end to the coordinator and the slot is defused here. That holds even when the subsequent client.peek() fails to issue.

fn begin_statement_logging( &self, session: &mut Session, params: &Params, logging: &Arc<QCell<PreparedStatementLoggingInfo>>, catalog: &Catalog, lifecycle_timestamps: Option<LifecycleTimestamps>, ) -> StatementLoggingGuard

Begins a new statement execution log entry, sampling permitting.

Only ExecutionLogging::take_over may call this: an entry that exists without the session task owning its end would stay unfinished forever.

pub(crate) fn log_began_execution( &self, record: StatementBeganExecutionRecord, mseh_update: Row, prepared_statement: Option<PreparedStatementEvent>, )

Log the beginning of statement execution.

pub(crate) fn log_set_cluster( &self, id: StatementLoggingId, cluster_id: ClusterId, cluster_name: String, )

Log cluster selection for a statement.

pub(crate) fn log_set_timestamp( &self, id: StatementLoggingId, timestamp: Timestamp, )

Log timestamp determination for a statement.

pub(crate) fn log_set_transient_index_id( &self, id: StatementLoggingId, transient_index_id: GlobalId, )

Log transient index ID for a statement.

pub(crate) fn log_lifecycle_event( &self, id: StatementLoggingId, event: StatementLifecycleEvent, )

Log a statement lifecycle event.

Trait Implementations

impl Debug for PeekClient

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.