Struct FlowDualEngine 

pub struct FlowDualEngine {
    streaming_engine: Arc<StreamingEngine>,
    batching_engine: Arc<BatchingEngine>,
    src_table2flow: RwLock<SrcTableToFlow>,
    flow_metadata_manager: Arc<FlowMetadataManager>,
    catalog_manager: Arc<dyn CatalogManager>,
    check_task: Mutex<Option<ConsistentCheckTask>>,
    plugins: Plugins,
    done_recovering: AtomicBool,
}

Manage both streaming and batching mode engine

including create/drop/flush flow and redirect insert requests to the appropriate engine

Fields

streaming_engine: Arc<StreamingEngine>

batching_engine: Arc<BatchingEngine>

src_table2flow: RwLock<SrcTableToFlow>

helper struct for faster query flow by table id or vice versa

flow_metadata_manager: Arc<FlowMetadataManager>

catalog_manager: Arc<dyn CatalogManager>

check_task: Mutex<Option<ConsistentCheckTask>>

plugins: Plugins

done_recovering: AtomicBool

Implementations

impl FlowDualEngine

pub fn new( streaming_engine: Arc<StreamingEngine>, batching_engine: Arc<BatchingEngine>, flow_metadata_manager: Arc<FlowMetadataManager>, catalog_manager: Arc<dyn CatalogManager>, plugins: Plugins, ) -> Self

pub fn set_done_recovering(&self)

Set done_recovering to true indicate that we are ready to handle requests

pub fn is_recover_done(&self) -> bool

Check if done_recovering is true

async fn wait_for_all_flow_recover( &self, waiting_req_cnt: usize, ) -> Result<(), Error>

wait for recovering to be done, this will only happen when flownode just started

pub fn plugins(&self) -> &Plugins

pub fn is_distributed(&self) -> bool

Determine if the engine is in distributed mode

pub fn streaming_engine(&self) -> Arc<StreamingEngine>

pub fn batching_engine(&self) -> Arc<BatchingEngine>

async fn wait_for_available_frontend( &self, timeout: Duration, ) -> Result<(), Error>

In distributed mode, scan periodically(1s) until available frontend is found, or timeout, in standalone mode, return immediately notice here if any frontend appear in cluster info this function will return immediately

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

Try to sync with check task, this is only used in drop flow&flush flow, so a flow id is required

the need to sync is to make sure flush flow actually get called

async fn check_flow_consistent( &self, allow_create: bool, allow_drop: bool, ) -> Result<(), Error>

Spawn a task to consistently check if all flow tasks in metasrv is created on flownode, so on startup, this will create all missing flow tasks, and constantly check at a interval

pub async fn start_flow_consistent_check_task( self: &Arc<Self>, ) -> Result<(), Error>

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

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

TODO(discord9): also add a exists api using flow metadata manager’s exists method

Trait Implementations

impl FlowEngine for FlowDualEngine

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

async fn handle_mark_window_dirty( &self, req: DirtyWindowRequests, ) -> Result<(), Error>

impl Flownode for FlowDualEngine

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,

fn handle_mark_window_dirty<'life0, 'async_trait>( &'life0 self, req: DirtyWindowRequests, ) -> Pin<Box<dyn Future<Output = MetaResult<FlowResponse>> + Send + 'async_trait>>

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

Handles requests to mark time window as dirty.