Struct BatchingTask

pub struct BatchingTask {
    pub config: Arc<TaskConfig>,
    pub state: Arc<RwLock<TaskState>>,
}

Fields

config: Arc<TaskConfig>

state: Arc<RwLock<TaskState>>

Implementations

impl BatchingTask

pub fn try_new( flow_id: u64, query: &str, plan: LogicalPlan, time_window_expr: Option<TimeWindowExpr>, expire_after: Option<i64>, sink_table_name: [String; 3], source_table_names: Vec<[String; 3]>, query_ctx: QueryContextRef, catalog_manager: CatalogManagerRef, shutdown_rx: Receiver<()>, ) -> Result<Self, Error>

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

mark time window range (now - expire_after, now) as dirty (or (0, now) if expire_after not set)

useful for flush_flow to flush dirty time windows range

pub async fn check_or_create_sink_table( &self, engine: &QueryEngineRef, frontend_client: &Arc<FrontendClient>, ) -> Result<Option<(u32, Duration)>, Error>

Create sink table if not exists

async fn is_table_exist(&self, table_name: &[String; 3]) -> Result<bool, Error>

pub async fn gen_exec_once( &self, engine: &QueryEngineRef, frontend_client: &Arc<FrontendClient>, ) -> Result<Option<(u32, Duration)>, Error>

pub async fn gen_insert_plan( &self, engine: &QueryEngineRef, ) -> Result<Option<LogicalPlan>, Error>

pub async fn create_table( &self, frontend_client: &Arc<FrontendClient>, expr: CreateTableExpr, ) -> Result<(), Error>

pub async fn execute_logical_plan( &self, frontend_client: &Arc<FrontendClient>, plan: &LogicalPlan, ) -> Result<Option<(u32, Duration)>, Error>

pub async fn start_executing_loop( &self, engine: QueryEngineRef, frontend_client: Arc<FrontendClient>, )

start executing query in a loop, break when receive shutdown signal

any error will be logged when executing query

async fn gen_create_table_expr( &self, engine: QueryEngineRef, ) -> Result<CreateTableExpr, Error>

Generate the create table SQL

the auto created table will automatically added a update_at Milliseconds DEFAULT now() column in the end (for compatibility with flow streaming mode)

and it will use first timestamp column as time index, all other columns will be added as normal columns and nullable

async fn gen_query_with_time_window( &self, engine: QueryEngineRef, sink_table_schema: &Arc<Schema>, ) -> Result<Option<(LogicalPlan, usize)>, Error>

will merge and use the first ten time window in query

Trait Implementations

impl Clone for BatchingTask

fn clone(&self) -> BatchingTask

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.