Struct Context

pub struct Context<'referred, 'df> {
    pub id: GlobalId,
    pub df: &'referred mut Hydroflow<'df>,
    pub compute_state: &'referred mut DataflowState,
    pub input_collection: BTreeMap<GlobalId, CollectionBundle>,
    pub local_scope: Vec<BTreeMap<LocalId, CollectionBundle>>,
    pub input_collection_batch: BTreeMap<GlobalId, CollectionBundle<Batch>>,
    pub local_scope_batch: Vec<BTreeMap<LocalId, CollectionBundle<Batch>>>,
    pub err_collector: ErrCollector,
}

The Context for build a Operator with id of GlobalId

Fields

id: GlobalId

df: &'referred mut Hydroflow<'df>

compute_state: &'referred mut DataflowState

input_collection: BTreeMap<GlobalId, CollectionBundle>

a list of all collections being used in the operator

TODO(discord9): remove extra clone by counting usage and remove it on last usage?

local_scope: Vec<BTreeMap<LocalId, CollectionBundle>>

used by Get/Let Plan for getting/setting local variables

TODO(discord9): consider if use Vec<(LocalId, CollectionBundle)> instead

input_collection_batch: BTreeMap<GlobalId, CollectionBundle<Batch>>

a list of all collections being used in the operator

TODO(discord9): remove extra clone by counting usage and remove it on last usage?

local_scope_batch: Vec<BTreeMap<LocalId, CollectionBundle<Batch>>>

used by Get/Let Plan for getting/setting local variables

TODO(discord9): consider if use Vec<(LocalId, CollectionBundle)> instead

err_collector: ErrCollector

Implementations

impl Context<'_, '_>

pub fn render_mfp_batch( &mut self, input: Box<TypedPlan>, mfp: MapFilterProject, _output_type: &RelationType, ) -> Result<CollectionBundle<Batch>, Error>

Like render_mfp but in batch mode

pub fn render_mfp( &mut self, input: Box<TypedPlan>, mfp: MapFilterProject, ) -> Result<CollectionBundle, Error>

render MapFilterProject, will only emit the rows once. Assume all incoming row’s sys time being `now`` and ignore the row’s stated sys time TODO(discord9): schedule mfp operator to run when temporal filter need

MapFilterProject(mfp for short) is scheduled to run when there is enough amount of input updates or when a future update in it’s output buffer(a Arrangement) is supposed to emit now.

impl Context<'_, '_>

const REDUCE_BATCH: &'static str = "reduce_batch"

const REDUCE: &'static str = "reduce"

pub fn render_reduce_batch( &mut self, input: Box<TypedPlan>, key_val_plan: &KeyValPlan, reduce_plan: &ReducePlan, output_type: &RelationType, ) -> Result<CollectionBundle<Batch>, Error>

Like render_reduce, but for batch mode, and only barebone implementation no support for distinct aggregation for now

pub fn render_reduce( &mut self, input: Box<TypedPlan>, key_val_plan: KeyValPlan, reduce_plan: ReducePlan, output_type: RelationType, ) -> Result<CollectionBundle, Error>

render Plan::Reduce into executable dataflow

fn add_accum_distinct_input_arrange( &mut self, reduce_plan: &ReducePlan, ) -> Option<Vec<ArrangeHandler>>

Contrast to it name, it’s for adding distinct input for accumulable reduce plan with distinct input, like select COUNT(DISTINCT col) from table

The return value is optional a list of arrangement, which is created for distinct input, and should be the same length as the distinct aggregation in accumulable reduce plan

impl Context<'_, '_>

pub fn render_source_batch( &mut self, src_recv: Receiver<Batch>, ) -> Result<CollectionBundle<Batch>, Error>

simply send the batch to downstream, without fancy features like buffering

pub fn render_source( &mut self, src_recv: Receiver<(Row, i64, i64)>, ) -> Result<CollectionBundle, Error>

Render a source which comes from brocast channel into the dataflow will immediately send updates not greater than now and buffer the rest in arrangement

pub fn render_unbounded_sink_batch( &mut self, bundle: CollectionBundle<Batch>, sender: UnboundedSender<Batch>, )

pub fn render_unbounded_sink( &mut self, bundle: CollectionBundle, sender: UnboundedSender<(Row, i64, i64)>, )

impl Context<'_, '_>

pub fn insert_global(&mut self, id: GlobalId, collection: CollectionBundle)

pub fn insert_local(&mut self, id: LocalId, collection: CollectionBundle)

pub fn insert_global_batch( &mut self, id: GlobalId, collection: CollectionBundle<Batch>, )

pub fn insert_local_batch( &mut self, id: LocalId, collection: CollectionBundle<Batch>, )

impl Context<'_, '_>

pub fn render_plan_batch( &mut self, plan: TypedPlan, ) -> Result<CollectionBundle<Batch>, Error>

Like render_plan but in Batch Mode

pub fn render_plan( &mut self, plan: TypedPlan, ) -> Result<CollectionBundle, Error>

Interpret plan to dataflow and prepare them for execution

return the output handler of this plan

pub fn render_constant_batch( &mut self, rows: Vec<(Row, i64, i64)>, output_type: &RelationType, ) -> CollectionBundle<Batch>

render Constant, take all rows that have a timestamp not greater than the current time This function is primarily used for testing Always assume input is sorted by timestamp

pub fn render_constant( &mut self, rows: Vec<(Row, i64, i64)>, ) -> CollectionBundle

render Constant, take all rows that have a timestamp not greater than the current time

Always assume input is sorted by timestamp

pub fn get_batch_by_id( &mut self, id: Id, ) -> Result<CollectionBundle<Batch>, Error>

pub fn get_by_id(&mut self, id: Id) -> Result<CollectionBundle, Error>

pub fn eval_batch_let( &mut self, id: LocalId, value: Box<TypedPlan>, body: Box<TypedPlan>, ) -> Result<CollectionBundle<Batch>, Error>

Eval Let operator, useful for assigning a value to a local variable

pub fn eval_let( &mut self, id: LocalId, value: Box<TypedPlan>, body: Box<TypedPlan>, ) -> Result<CollectionBundle, Error>

Eval Let operator, useful for assigning a value to a local variable

Trait Implementations

impl Drop for Context<'_, '_>

fn drop(&mut self)

Executes the destructor for this type.