Struct StreamingEngine

pub struct StreamingEngine {

    pub worker_handles: Vec<WorkerHandle>,
    worker_selector: Mutex<usize>,
    pub query_engine: Arc<dyn QueryEngine>,
    table_info_source: ManagedTableSource,
    frontend_invoker: RwLock<Option<FrontendInvoker>>,
    node_context: RwLock<FlownodeContext>,
    refill_tasks: RwLock<BTreeMap<u64, RefillTask>>,
    flow_err_collectors: RwLock<BTreeMap<u64, ErrCollector>>,
    src_send_buf_lens: RwLock<BTreeMap<TableId, Receiver<usize>>>,
    tick_manager: FlowTickManager,
    pub node_id: Option<u32>,
    flush_lock: RwLock<()>,
    state_report_handler: RwLock<Option<Receiver<Sender<FlowStat>>>>,
}

FlowNodeManager manages the state of all tasks in the flow node, which should be run on the same thread

The choice of timestamp is just using current system timestamp for now

Fields

worker_handles: Vec<WorkerHandle>

The handler to the worker that will run the dataflow which is !Send so a handle is used

worker_selector: Mutex<usize>

The selector to select a worker to run the dataflow

query_engine: Arc<dyn QueryEngine>

The query engine that will be used to parse the query and convert it to a dataflow plan

table_info_source: ManagedTableSource

Getting table name and table schema from table info manager

frontend_invoker: RwLock<Option<FrontendInvoker>>

node_context: RwLock<FlownodeContext>

contains mapping from table name to global id, and table schema

refill_tasks: RwLock<BTreeMap<u64, RefillTask>>

Contains all refill tasks

flow_err_collectors: RwLock<BTreeMap<u64, ErrCollector>>

src_send_buf_lens: RwLock<BTreeMap<TableId, Receiver<usize>>>

tick_manager: FlowTickManager

node_id: Option<u32>

This node id is only available in distributed mode, on standalone mode this is guaranteed to be None

flush_lock: RwLock<()>

Lock for flushing, will be read by handle_inserts and write by flush_flow

So that a series of event like inserts -> flush can be handled correctly

state_report_handler: RwLock<Option<Receiver<Sender<FlowStat>>>>

receive a oneshot sender to send state size report

Implementations

impl StreamingEngine

async fn handle_inserts_inner( &self, request: InsertRequests, ) -> Result<(), Error>

impl StreamingEngine

pub async fn create_and_start_refill_flow_tasks( self: &FlowStreamingEngineRef, flow_metadata_manager: &FlowMetadataManagerRef, catalog_manager: &CatalogManagerRef, ) -> Result<(), Error>

Create and start refill flow tasks in background

pub async fn create_refill_flow_tasks( &self, flow_metadata_manager: &FlowMetadataManagerRef, catalog_manager: &CatalogManagerRef, ) -> Result<Vec<RefillTask>, Error>

Create a series of tasks to refill flow

pub(crate) async fn starting_refill_flows( self: &FlowStreamingEngineRef, tasks: Vec<RefillTask>, ) -> Result<(), Error>

Starting to refill flows, if any error occurs, will rebuild the flow and retry

impl StreamingEngine

pub async fn gen_state_report(&self) -> FlowStat

impl StreamingEngine

pub(crate) async fn get_worker_handle_for_create_flow(&self) -> &WorkerHandle

Get a worker handle for creating flow, using round robin to select a worker

pub(crate) async fn create_table_from_relation( &self, flow_name: &str, table_name: &[String; 3], relation_desc: &RelationDesc, ) -> Result<bool, Error>

Create table from given schema(will adjust to add auto column if needed), return true if table is created

pub(crate) async fn try_fetch_existing_table( &self, table_name: &[String; 3], ) -> Result<Option<(bool, Vec<ColumnSchema>)>, Error>

Try fetch table with adjusted schema(added auto column if needed)

pub(crate) async fn submit_create_sink_table_ddl( &self, create_table: CreateTableExpr, ) -> Result<(), Error>

submit a create table ddl

impl StreamingEngine

Building FlownodeManager

pub async fn set_frontend_invoker(&self, frontend: FrontendInvoker)

set frontend invoker

pub fn new( node_id: Option<u32>, query_engine: Arc<dyn QueryEngine>, table_meta: TableMetadataManagerRef, ) -> Self

Create without setting frontend_invoker

pub async fn with_state_report_handler( self, handler: Receiver<Sender<FlowStat>>, ) -> Self

pub fn new_with_workers<'s>( node_id: Option<u32>, query_engine: Arc<dyn QueryEngine>, table_meta: TableMetadataManagerRef, num_workers: usize, ) -> (Self, Vec<Worker<'s>>)

Create a flownode manager with one worker

pub fn add_worker_handle(&mut self, handle: WorkerHandle)

add a worker handler to manager, meaning this corresponding worker is under it’s manage

impl StreamingEngine

This impl block contains methods to send writeback requests to frontend

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

Return the number of requests it made

pub async fn generate_writeback_request( &self, ) -> Result<BTreeMap<[String; 3], Vec<DiffRequest>>, Error>

Generate writeback request for all sink table

async fn fetch_table_pk_schema( &self, table_name: &[String; 3], ) -> Result<Option<(Vec<String>, Option<usize>, Vec<ColumnSchema>)>, Error>

Fetch table schema and primary key from table info source, if table not exist return None

async fn adjust_auto_created_table_schema( &self, schema: &RelationDesc, ) -> Result<(Vec<String>, Vec<ColumnSchema>, bool), Error>

return (primary keys, schema and if the table have a placeholder timestamp column) schema of the table comes from flow’s output plan

adjust to add update_at column and ts placeholder if needed

impl StreamingEngine

Flow Runtime related methods

async fn start_state_report_handler(self: Arc<Self>) -> Option<JoinHandle<()>>

Start state report handler, which will receive a sender from HeartbeatTask to send state size report back

if heartbeat task is shutdown, this future will exit too

pub fn run_background( self: Arc<Self>, shutdown: Option<Receiver<()>>, ) -> JoinHandle<()>

run in common_runtime background runtime

pub async fn log_all_errors(&self)

log all flow errors

pub async fn run(&self, shutdown: Option<Receiver<()>>)

Trigger dataflow running, and then send writeback request to the source sender

note that this method didn’t handle input mirror request, as this should be handled by grpc server

pub async fn run_available(&self, blocking: bool) -> Result<usize, Error>

Run all available subgraph in the flow node This will try to run all dataflow in this node

set blocking to true to wait until worker finish running false to just trigger run and return immediately return numbers of rows send to worker(Inaccuary) TODO(discord9): add flag for subgraph that have input since last run

pub async fn handle_write_request( &self, region_id: RegionId, rows: Vec<(Row, i64, i64)>, batch_datatypes: &[ConcreteDataType], ) -> Result<(), Error>

send write request to related source sender

impl StreamingEngine

Create&Remove flow

pub async fn remove_flow_inner(&self, flow_id: u64) -> Result<(), Error>

remove a flow by it’s id

pub async fn create_flow_inner( &self, args: CreateFlowArgs, ) -> Result<Option<u64>, Error>

Return task id if a new task is created, otherwise return None

steps to create task:

1. parse query into typed plan(and optional parse expire_after expr)
2. render source/sink with output table id and used input table id

pub async fn flush_flow_inner(&self, flow_id: u64) -> Result<usize, Error>

pub async fn flow_exist_inner(&self, flow_id: u64) -> Result<bool, Error>

Trait Implementations

impl FlowEngine for StreamingEngine

async fn create_flow(&self, args: CreateFlowArgs) -> Result<Option<u64>, Error>

Create a flow using the provided arguments, return previous flow id if exists and is replaced

async fn remove_flow(&self, flow_id: u64) -> Result<(), Error>

Remove a flow by its ID

async fn flush_flow(&self, flow_id: u64) -> Result<usize, Error>

Flush the flow, return the number of rows flushed

async fn flow_exist(&self, flow_id: u64) -> Result<bool, Error>

Check if the flow exists

async fn list_flows(&self) -> Result<impl IntoIterator<Item = u64>, Error>

List all flows

async fn handle_flow_inserts( &self, request: InsertRequests, ) -> Result<(), Error>

Handle the insert requests for the flow

impl Flownode for StreamingEngine

fn handle<'life0, 'async_trait>( &'life0 self, request: FlowRequest, ) -> Pin<Box<dyn Future<Output = MetaResult<FlowResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn handle_inserts<'life0, 'async_trait>( &'life0 self, request: InsertRequests, ) -> Pin<Box<dyn Future<Output = MetaResult<FlowResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,