Struct dataflow::render::context::Context

pub struct Context<S: Scope, V: Data, T> where
    T: Timestamp + Lattice,
    S::Timestamp: Lattice + Refines<T>,  {
    pub debug_name: String,
    pub dataflow_id: usize,
    pub as_of_frontier: Antichain<Timestamp>,
    pub bindings: BTreeMap<Id, CollectionBundle<S, V, T>>,
}

Dataflow-local collections and arrangements.

A context means to wrap available data assets and present them in an easy-to-use manner. These assets include dataflow-local collections and arrangements, as well as imported arrangements from outside the dataflow.

Context has two timestamp types, one from S::Timestamp and one from T, where the former must refine the latter. The former is the timestamp used by the scope in question, and the latter is the timestamp of imported traces. The two may be different in the case of regions or iteration.

Fields

debug_name: String

The debug name of the dataflow associated with this context.

dataflow_id: usize

The Timely ID of the dataflow associated with this context.

as_of_frontier: Antichain<Timestamp>

Indicates a frontier that can be used to compact input timestamps without affecting the results. We should apply it, to sources and imported traces, both because it improves performance, and because potentially incorrect results are visible in sinks.

bindings: BTreeMap<Id, CollectionBundle<S, V, T>>

Bindings of identifiers to collections.

Implementations

impl<S: Scope, V: Data, T> Context<S, V, T> where
    T: Timestamp + Lattice,
    S::Timestamp: Lattice + Refines<T>, 

pub fn for_dataflow<Plan>(
    dataflow: &DataflowDescription<Plan>,
    dataflow_id: usize
) -> Self

Creates a new empty Context.

pub fn insert_id(
    &mut self,
    id: Id,
    collection: CollectionBundle<S, V, T>
) -> Option<CollectionBundle<S, V, T>>

Insert a collection bundle by an identifier.

This is expected to be used to install external collections (sources, indexes, other views), as well as for Let bindings of local collections.

pub fn remove_id(&mut self, id: Id) -> Option<CollectionBundle<S, V, T>>

Remove a collection bundle by an identifier.

The primary use of this method is uninstalling Let bindings.

pub fn update_id(&mut self, id: Id, collection: CollectionBundle<S, V, T>)

Melds a collection bundle to whatever exists.

pub fn lookup_id(&self, id: Id) -> Option<CollectionBundle<S, V, T>>

Look up a collection bundle by an identifier.

impl<G> Context<G, Row, Timestamp> where
    G: Scope<Timestamp = Timestamp>, 

pub fn render_flat_map(
    &mut self,
    input: CollectionBundle<G, Row, G::Timestamp>,
    func: TableFunc,
    exprs: Vec<MirScalarExpr>,
    mfp: MapFilterProject,
    input_key: Option<Vec<MirScalarExpr>>
) -> CollectionBundle<G, Row, G::Timestamp>

Renders relation_expr followed by map_filter_project if provided.

impl<G> Context<G, Row, Timestamp> where
    G: Scope<Timestamp = Timestamp>, 

pub fn render_delta_join(
    &mut self,
    inputs: Vec<CollectionBundle<G, Row, G::Timestamp>>,
    join_plan: DeltaJoinPlan,
    scope: &mut G
) -> CollectionBundle<G, Row, G::Timestamp>

Renders MirRelationExpr:Join using dogs^3 delta query dataflows.

The join is followed by the application of map_filter_project, whose implementation will be pushed in to the join pipeline if at all possible.

impl<G, T> Context<G, Row, T> where
    G: Scope,
    G::Timestamp: Lattice + Refines<T>,
    T: Timestamp + Lattice, 

pub fn render_join(
    &mut self,
    inputs: Vec<CollectionBundle<G, Row, T>>,
    linear_plan: LinearJoinPlan,
    scope: &mut G
) -> CollectionBundle<G, Row, T>

fn differential_join(
    &mut self,
    joined: JoinedFlavor<G, T>,
    lookup_relation: CollectionBundle<G, Row, T>,
    _: LinearStagePlan,
    errors: &mut Vec<Collection<G, DataflowError>>
) -> Collection<G, Row>

Looks up the arrangement for the next input and joins it to the arranged version of the join of previous inputs. This is split into its own method to enable reuse of code with different types of prev_keyed.

fn differential_join_inner<J, Tr2>(
    &mut self,
    prev_keyed: J,
    next_input: Arranged<G, Tr2>,
    closure: JoinClosure
) -> (Collection<G, Row>, Collection<G, DataflowError>) where
    J: JoinCore<G, Row, Row, Diff>,
    Tr2: TraceReader<Key = Row, Val = Row, Time = G::Timestamp, R = Diff> + Clone + 'static,
    Tr2::Batch: BatchReader<Row, Tr2::Val, G::Timestamp, Diff> + 'static,
    Tr2::Cursor: Cursor<Row, Tr2::Val, G::Timestamp, Diff> + 'static, 

Joins the arrangement for next_input to the arranged version of the join of previous inputs. This is split into its own method to enable reuse of code with different types of next_input.

impl<G, T> Context<G, Row, T> where
    G: Scope,
    G::Timestamp: Lattice + Refines<T>,
    T: Timestamp + Lattice, 

pub fn render_reduce(
    &mut self,
    input: CollectionBundle<G, Row, T>,
    key_val_plan: KeyValPlan,
    reduce_plan: ReducePlan,
    input_key: Option<Vec<MirScalarExpr>>
) -> CollectionBundle<G, Row, T>

Renders a MirRelationExpr::Reduce using various non-obvious techniques to minimize worst-case incremental update times and memory footprint.

impl<G> Context<G, Row, Timestamp> where
    G: Scope<Timestamp = Timestamp>, 

pub(crate) fn export_sink(
    &mut self,
    compute_state: &mut ComputeState,
    tokens: &mut BTreeMap<GlobalId, Rc<dyn Any>>,
    import_ids: HashSet<GlobalId>,
    sink_id: GlobalId,
    sink: &SinkDesc,
    metrics: &SinkBaseMetrics
)

Export the sink described by sink from the rendering context.

impl<G, T> Context<G, Row, T> where
    G: Scope,
    G::Timestamp: Lattice + Refines<T>,
    T: Timestamp + Lattice, 

pub fn render_threshold(
    &self,
    input: CollectionBundle<G, Row, T>,
    threshold_plan: ThresholdPlan
) -> CollectionBundle<G, Row, T>

impl<G> Context<G, Row, Timestamp> where
    G: Scope<Timestamp = Timestamp>, 

pub fn render_topk(
    &mut self,
    input: CollectionBundle<G, Row, G::Timestamp>,
    top_k_plan: TopKPlan
) -> CollectionBundle<G, Row, G::Timestamp>

impl<'g, G> Context<Child<'g, G, G::Timestamp>, Row, Timestamp> where
    G: Scope<Timestamp = Timestamp>, 

fn import_index(
    &mut self,
    compute_state: &mut ComputeState,
    tokens: &mut BTreeMap<GlobalId, Rc<dyn Any>>,
    scope: &mut G,
    region: &mut Child<'g, G, G::Timestamp>,
    idx_id: GlobalId,
    idx: &IndexDesc
)

fn build_object(
    &mut self,
    scope: &mut Child<'g, G, G::Timestamp>,
    object: BuildDesc<Plan>
)

fn export_index(
    &mut self,
    compute_state: &mut ComputeState,
    tokens: &mut BTreeMap<GlobalId, Rc<dyn Any>>,
    import_ids: HashSet<GlobalId>,
    idx_id: GlobalId,
    idx: &IndexDesc
)

impl<G> Context<G, Row, Timestamp> where
    G: Scope<Timestamp = Timestamp>, 

pub fn render_plan(
    &mut self,
    plan: Plan,
    scope: &mut G,
    worker_index: usize
) -> CollectionBundle<G, Row, G::Timestamp>

Renders a plan to a differential dataflow, producing the collection of results.

The return type reflects the uncertainty about the data representation, perhaps as a stream of data, perhaps as an arrangement, perhaps as a stream of batches.