Struct mz_compute::compute_state::ComputeState

pub struct ComputeState {

    pub collections: BTreeMap<GlobalId, CollectionState>,
    pub dropped_collections: Vec<(GlobalId, DroppedCollection)>,
    pub traces: TraceManager,
    pub subscribe_response_buffer: Rc<RefCell<Vec<(GlobalId, SubscribeResponse)>>>,
    pub copy_to_response_buffer: Rc<RefCell<Vec<(GlobalId, CopyToResponse)>>>,
    pub pending_peeks: BTreeMap<Uuid, PendingPeek>,
    pub compute_logger: Option<Logger<ComputeEvent, WorkerIdentifier>>,
    pub persist_clients: Arc<PersistClientCache>,
    pub txns_ctx: TxnsContext,
    pub command_history: ComputeCommandHistory<UIntGauge>,
    max_result_size: u64,
    pub linear_join_spec: LinearJoinSpec,
    pub metrics: ComputeMetrics,
    pub worker_metrics: WorkerMetrics,
    tracing_handle: Arc<TracingHandle>,
    pub context: ComputeInstanceContext,
    pub worker_config: ConfigSet,
    pub hydration_rx: Receiver<HydrationEvent>,
    pub hydration_tx: Sender<HydrationEvent>,
    suspended_collections: BTreeMap<GlobalId, Rc<dyn Any>>,
    pub read_only_rx: Receiver<bool>,
    pub read_only_tx: Sender<bool>,
    pub server_maintenance_interval: Duration,
    pub init_system_time: EpochMillis,
    pub replica_expiration: Antichain<Timestamp>,
}

Worker-local state that is maintained across dataflows.

This state is restricted to the COMPUTE state, the deterministic, idempotent work done between data ingress and egress.

Fields

collections: BTreeMap<GlobalId, CollectionState>

State kept for each installed compute collection.

Each collection has exactly one frontier. How the frontier is communicated depends on the collection type:

• Frontiers of indexes are equal to the frontier of their corresponding traces in the TraceManager.
• Persist sinks store their current frontier in CollectionState::sink_write_frontier.
• Subscribes report their frontiers through the subscribe_response_buffer.

dropped_collections: Vec<(GlobalId, DroppedCollection)>

Collections that were recently dropped and whose removal needs to be reported.

traces: TraceManager

The traces available for sharing across dataflows.

subscribe_response_buffer: Rc<RefCell<Vec<(GlobalId, SubscribeResponse)>>>

Shared buffer with SUBSCRIBE operator instances by which they can respond.

The entries are pairs of sink identifier (to identify the subscribe instance) and the response itself.

copy_to_response_buffer: Rc<RefCell<Vec<(GlobalId, CopyToResponse)>>>

Shared buffer with S3 oneshot operator instances by which they can respond.

The entries are pairs of sink identifier (to identify the s3 oneshot instance) and the response itself.

pending_peeks: BTreeMap<Uuid, PendingPeek>

Peek commands that are awaiting fulfillment.

compute_logger: Option<Logger<ComputeEvent, WorkerIdentifier>>

The logger, from Timely’s logging framework, if logs are enabled.

persist_clients: Arc<PersistClientCache>

A process-global cache of (blob_uri, consensus_uri) -> PersistClient. This is intentionally shared between workers.

txns_ctx: TxnsContext

Context necessary for rendering txn-wal operators.

command_history: ComputeCommandHistory<UIntGauge>

History of commands received by this workers and all its peers.

max_result_size: u64

Max size in bytes of any result.

linear_join_spec: LinearJoinSpec

Specification for rendering linear joins.

metrics: ComputeMetrics

Metrics for this replica.

worker_metrics: WorkerMetrics

Metrics for this replica, specific to a worker.

tracing_handle: Arc<TracingHandle>

A process-global handle to tracing configuration.

context: ComputeInstanceContext

Other configuration for compute

worker_config: ConfigSet

Per-worker dynamic configuration.

This is separate from the process-global ConfigSet and contains config options that need to be applied consistently with compute command order.

For example, for options that influence dataflow rendering it is important that all workers render the same dataflow with the same options. If these options were stored in a global ConfigSet, we couldn’t guarantee that all workers observe changes to them at the same point in the stream of compute commands. Storing per-worker configuration ensures that because each worker’s configuration is only updated once that worker observes the respective UpdateConfiguration command.

hydration_rx: Receiver<HydrationEvent>

Receiver of operator hydration events.

hydration_tx: Sender<HydrationEvent>

Transmitter of operator hydration events.

Copies of this sender are passed to the hydration logging operators.

suspended_collections: BTreeMap<GlobalId, Rc<dyn Any>>

Collections awaiting schedule instruction by the controller.

Each entry stores a reference to a token that can be dropped to unsuspend the collection’s dataflow. Multiple collections can reference the same token if they are exported by the same dataflow.

read_only_rx: Receiver<bool>

When this replica/cluster is in read-only mode it must not affect any changes to external state. This flag can only be changed by a ComputeCommand::AllowWrites.

Everything running on this replica/cluster must obey this flag. At the time of writing the only part that is doing this is persist_sink.

NOTE: In the future, we might want a more complicated flag, for example something that tells us after which timestamp we are allowed to write. In this first version we are keeping things as simple as possible!

read_only_tx: Sender<bool>

Send-side for read-only state.

server_maintenance_interval: Duration

Interval at which to perform server maintenance tasks. Set to a zero interval to perform maintenance with every step_or_park invocation.

init_system_time: EpochMillis

The mz_ore::now::SYSTEM_TIME at which the replica was started.

Used to compute replica_expiration.

replica_expiration: Antichain<Timestamp>

The maximum time for which the replica is expected to live. If not empty, dataflows in the replica can drop diffs associated with timestamps beyond the replica expiration. The replica will panic if such dataflows are not dropped before the replica has expired.

Implementations

impl ComputeState

pub fn new( worker_id: usize, persist_clients: Arc<PersistClientCache>, txns_ctx: TxnsContext, metrics: ComputeMetrics, tracing_handle: Arc<TracingHandle>, context: ComputeInstanceContext, ) -> Self

Construct a new ComputeState.

pub fn collection_exists(&self, id: GlobalId) -> bool

Return whether a collection with the given ID exists.

pub fn expect_collection(&self, id: GlobalId) -> &CollectionState

Return a reference to the identified collection.

Panics if the collection doesn’t exist.

pub fn expect_collection_mut(&mut self, id: GlobalId) -> &mut CollectionState

Return a mutable reference to the identified collection.

Panics if the collection doesn’t exist.

pub fn input_probe_for( &mut self, input_id: GlobalId, collection_ids: impl Iterator<Item = GlobalId>, ) -> Handle<Timestamp>

Construct a new frontier probe for the given input and add it to the state of the given collections.

The caller is responsible for attaching the returned probe handle to the respective dataflow input stream.

fn apply_worker_config(&mut self)

Apply the current worker_config to the compute state.

pub fn apply_expiration_offset(&mut self, offset: Duration)

Apply the provided replica expiration offset by converting it to a frontier relative to the replica’s initialization system time.

Only expected to be called once when creating the instance. Guards against calling it multiple times by checking if the local expiration time is set.

pub fn dataflow_max_inflight_bytes(&self) -> Option<usize>

Returns the cc or non-cc version of “dataflow_max_inflight_bytes”, as appropriate to this replica.