Struct mz_compute_client::as_of_selection::Context

struct Context<'a, T> {
    collections: BTreeMap<GlobalId, Collection<'a, T>>,
    storage_collections: &'a dyn StorageCollections<Timestamp = T>,
    current_time: T,
}

The as-of selection context.

Fields

collections: BTreeMap<GlobalId, Collection<'a, T>>

storage_collections: &'a dyn StorageCollections<Timestamp = T>

current_time: T

Implementations

impl<'a, T: TimestampManipulation> Context<'a, T>

fn new( dataflows: &[DataflowDescription<Plan<T>, (), T>], storage_collections: &'a dyn StorageCollections<Timestamp = T>, read_policies: &'a BTreeMap<GlobalId, ReadPolicy<T>>, current_time: T, ) -> Self

Initializes an as-of selection context for the given dataflows.

fn expect_collection(&self, id: GlobalId) -> &Collection<'_, T>

Returns the state of the identified collection.

Panics

Panics if the identified collection doesn’t exist.

fn apply_constraint(&self, id: GlobalId, constraint: Constraint<'_, T>) -> bool

Applies the given as-of constraint to the identified collection.

Returns whether the collection’s as-of bounds where changed as a result.

fn apply_upstream_storage_constraints( &self, storage_read_holds: &BTreeMap<GlobalId, ReadHold<T>>, )

Apply as-of constraints imposed by the frontiers of upstream storage collections.

A collection’s as-of must be >= the read frontier of each of its (transitive) storage inputs.

Failing to apply this constraint to a collection is an error. The affected dataflow will not be able to hydrate successfully.

fn apply_downstream_storage_constraints(&self)

Apply as-of constraints imposed by the frontiers of downstream storage collections.

A collection’s as-of must be <= the join of the read frontier and the write frontier of the storage collection it exports to, if any.

We need to consider the read frontier in addition to the write frontier because storage collections are commonly initialyzed with a write frontier of [0], even when they start producing output at some later time. The read frontier is always initialized to the first time the collection will produce valid output for, so it constrains the times that need to be produced by dependencies of newly initialized storage collections.

Failing to apply this constraint to a collection is an error. The storage collection it exports to will have times visible to readers skipped in its output, violating correctness.

fn apply_warmup_constraints(&self)

Apply as-of constraints to ensure collections can hydrate immediately.

A collection’s as-of should be < the write frontier of each of its (transitive) storage inputs.

Failing to apply this constraint is not an error. The affected dataflow will not be able to hydrate immediately, but it will be able to hydrate once its inputs have sufficiently advanced.

fn apply_index_read_policy_constraints(&self)

Apply as-of constraints to ensure indexes contain historical data as requested by their associated read policies.

An index’s as-of should be <= the frontier determined by its read policy applied to its write frontier.

Failing to apply this constraint is not an error. The affected index will not contain historical times for its entire compaction window initially, but will do so once sufficient time has passed.

fn apply_index_current_time_constraints(&self)

Apply as-of constraints to ensure indexes are immediately readable.

An index’s as-of should be <= the current time.

Failing to apply this constraint is not an error. The affected index will not be readable immediately, but will be readable once sufficient time has passed.

fn propagate_bounds_downstream(&self, bound_type: BoundType)

Propagate as-of bounds through the dependency graph, in downstream direction.

fn propagate_bounds_downstream_inner( &self, bound_type: BoundType, constraint_type: ConstraintType, changed: &mut bool, )

fn propagate_bounds_upstream(&self, bound_type: BoundType)

Propagate as-of bounds through the dependency graph, in upstream direction.

fn propagate_bounds_upstream_inner( &self, bound_type: BoundType, constraint_type: ConstraintType, changed: &mut bool, )

fn best_as_of(&self, id: GlobalId) -> Antichain<T>

Selects the “best” as-of for the identified collection, based on its currently known bounds.

We simply use the upper bound here, to maximize the chances of compute reconciliation succeeding. Choosing the latest possible as-of also minimizes the amount of work the dataflow has to spend processing historical data from its sources.

fn prune_sealed_persist_sinks(&mut self)

Removes collections that sink into sealed persist shards from the context.

The dataflows of these collections will get an empty default as-of assigned at the end of the as-of selection process, ensuring that they won’t get installed unnecessarily.

Note that it is valid to remove these collections from consideration because they don’t impose as-of constraints on other compute collections.