Struct FlownodeContext

pub struct FlownodeContext {
    pub source_to_tasks: BTreeMap<TableId, BTreeSet<u64>>,
    pub flow_to_sink: BTreeMap<u64, [String; 3]>,
    pub flow_plans: BTreeMap<u64, TypedPlan>,
    pub sink_to_flow: BTreeMap<[String; 3], u64>,
    pub source_sender: BTreeMap<TableId, SourceSender>,
    pub sink_receiver: BTreeMap<[String; 3], (UnboundedSender<Batch>, UnboundedReceiver<Batch>)>,
    pub table_source: Box<dyn FlowTableSource>,
    pub table_repr: IdToNameMap,
    pub query_context: Option<Arc<QueryContext>>,
}

A context that holds the information of the dataflow

Fields

source_to_tasks: BTreeMap<TableId, BTreeSet<u64>>

mapping from source table to tasks, useful for schedule which task to run when a source table is updated

flow_to_sink: BTreeMap<u64, [String; 3]>

mapping from task to sink table, useful for sending data back to the client when a task is done running

flow_plans: BTreeMap<u64, TypedPlan>

sink_to_flow: BTreeMap<[String; 3], u64>

source_sender: BTreeMap<TableId, SourceSender>

broadcast sender for source table, any incoming write request will be sent to the source table’s corresponding sender

Note that we are getting insert requests with table id, so we should use table id as the key

sink_receiver: BTreeMap<[String; 3], (UnboundedSender<Batch>, UnboundedReceiver<Batch>)>

broadcast receiver for sink table, there should only be one receiver, and it will receive all the data from the sink table

and send it back to the client, since we are mocking the sink table as a client, we should use table name as the key note that the sink receiver should only have one, and we are using broadcast as mpsc channel here

table_source: Box<dyn FlowTableSource>

can query the schema of the table source, from metasrv with local cache

table_repr: IdToNameMap

All the tables that have been registered in the worker

query_context: Option<Arc<QueryContext>>

Implementations

impl FlownodeContext

pub fn new(table_source: Box<dyn FlowTableSource>) -> Self

pub fn get_flow_ids(&self, table_id: TableId) -> Option<&BTreeSet<u64>>

impl FlownodeContext

pub async fn send( &self, table_id: TableId, rows: Vec<(Row, i64, i64)>, batch_datatypes: &[ConcreteDataType], ) -> Result<usize, Error>

return number of rows it actual send(including what’s in the buffer)

TODO(discord9): make this concurrent

pub async fn send_rb( &self, table_id: TableId, batch: RecordBatch, ) -> Result<usize, Error>

pub async fn flush_all_sender(&self) -> Result<usize, Error>

flush all sender’s buf

return numbers being sent

impl FlownodeContext

pub fn register_task_src_sink( &mut self, task_id: u64, source_table_ids: &[TableId], sink_table_name: [String; 3], )

mapping source table to task, and sink table to task in worker context

also add their corresponding broadcast sender/receiver

pub fn add_flow_plan(&mut self, task_id: u64, plan: TypedPlan)

add flow plan to worker context

pub fn get_flow_plan(&self, task_id: &u64) -> Option<TypedPlan>

pub fn remove_flow(&mut self, task_id: u64)

remove flow from worker context

pub fn add_source_sender_if_not_exist(&mut self, table_id: TableId)

try add source sender, if already exist, do nothing

pub fn add_sink_receiver(&mut self, table_name: [String; 3])

pub fn get_source_by_global_id( &self, id: &GlobalId, ) -> Result<&SourceSender, Error>

pub fn get_sink_by_global_id( &self, id: &GlobalId, ) -> Result<UnboundedSender<Batch>, Error>

impl FlownodeContext

pub async fn table( &self, name: &[String; 3], ) -> Result<(GlobalId, RelationDesc), Error>

Retrieves a GlobalId and table schema representing a table previously registered by calling the [register_table] function.

Returns an error if no table has been registered with the provided names

pub async fn assign_global_id_to_table( &mut self, srv_map: &ManagedTableSource, table_name: Option<[String; 3]>, table_id: Option<TableId>, ) -> Result<GlobalId, Error>

Assign a global id to a table, if already assigned, return the existing global id

require at least one of table_name or table_id to be Some

and will try to fetch the schema from table info manager(if table exist now)

NOTE: this will not actually render the table into collection referred as GlobalId merely creating a mapping from table id to global id

pub fn new_global_id(&self) -> GlobalId

Get a new global id

Trait Implementations

impl Debug for FlownodeContext

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.