Struct PeekClient

pub struct PeekClient {
    coordinator_client: Client,
    compute_instances: BTreeMap<ComputeInstanceId, Client<Timestamp>>,
    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,
}

Clients needed for peek sequencing in the Adapter Frontend.

Fields

coordinator_client: Client

compute_instances: BTreeMap<ComputeInstanceId, Client<Timestamp>>

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

Implementations

impl PeekClient

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

pub(crate) async fn frontend_determine_timestamp( &mut self, catalog_state: &CatalogState, 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<Timestamp>, ReadHolds<Timestamp>), 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.

impl PeekClient

pub fn new( coordinator_client: Client, storage_collections: StorageCollectionsHandle, transient_id_gen: Arc<TransientIdGen>, optimizer_metrics: OptimizerMetrics, persist_client: PersistClient, ) -> Self

Creates a PeekClient.

pub async fn ensure_compute_instance_client( &mut self, compute_instance: ComputeInstanceId, ) -> Result<&mut Client<Timestamp>, InstanceMissing>

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

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

Fetch a snapshot of the catalog for use in frontend peek sequencing. Records the time taken in the adapter metrics, labeled by context.

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

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

pub async fn acquire_read_holds_and_least_valid_write( &mut self, id_bundle: &CollectionIdBundle, ) -> Result<(ReadHolds<Timestamp>, 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 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<Timestamp>, peek_stash_read_batch_size_bytes: usize, peek_stash_read_memory_budget_bytes: usize, ) -> 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.

TODO(peek-seq): add statement logging TODO(peek-seq): cancellation (see pending_peeks/client_pending_peeks wiring in the old sequencing)

Trait Implementations

impl Debug for PeekClient

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

Formats the value using the given formatter.