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 write frontier of the storage collection it exports to (if any) if it is non-empty, and <= the storage collection’s read frontier otherwise.

Rationale:

• A collection’s as-of must be <= the write frontier of its dependent storage collection, because we need to pick up computing the contents of storage collections where we left off previously, to avoid skipped times observable in the durable output.
• Some dataflows feeding into storage collections (specifically: continual tasks) need to be able to observe input changes at times they write to the output. If we selected the as-of to be equal to the write frontier of the output storage collection, we wouldn’t be able to produce the correct output at that frontier. Thus the selected as-of must be strictly less than the write frontier.
• As an exception to the above, if the output storage collection is empty (i.e. its write frontier is <= its read frontier), we need to allow the as-of to be equal to the read frontier. This is correct in the sense that it mirrors the timestamp selection behavior of the sequencer when it created the collection. Chances are that the sequencer chose the initial as-of (and therefore the initial read frontier of the storage collection) as the smallest possible time that can still be read from the collection inputs, so forcing the upper bound any lower than that read frontier would produce a hard constraint violation.

Failing to apply this constraint to a collection is an error. The storage collection it exports to may 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.