Struct BatchingEngine 

pub struct BatchingEngine {
    runtime: RwLock<FlowRuntimeRegistry>,
    pub(crate) frontend_client: Arc<FrontendClient>,
    flow_metadata_manager: FlowMetadataManagerRef,
    table_meta: TableMetadataManagerRef,
    catalog_manager: CatalogManagerRef,
    query_engine: QueryEngineRef,
    pub(crate) batch_opts: Arc<BatchingModeOptions>,
}

Batching mode Engine, responsible for driving all the batching mode tasks

TODO(discord9): determine how to configure refresh rate

Fields

runtime: RwLock<FlowRuntimeRegistry>

frontend_client: Arc<FrontendClient>

frontend client for insert request

flow_metadata_manager: FlowMetadataManagerRef

table_meta: TableMetadataManagerRef

catalog_manager: CatalogManagerRef

query_engine: QueryEngineRef

batch_opts: Arc<BatchingModeOptions>

Batching mode options for control how batching mode query works

Implementations

impl BatchingEngine

pub fn new( frontend_client: Arc<FrontendClient>, query_engine: QueryEngineRef, flow_metadata_manager: FlowMetadataManagerRef, table_meta: TableMetadataManagerRef, catalog_manager: CatalogManagerRef, batch_opts: BatchingModeOptions, ) -> Self

pub async fn get_last_exec_time_map(&self) -> BTreeMap<u64, i64>

Returns last execution timestamps (millisecond) for all batching flows.

pub async fn handle_mark_dirty_time_window( &self, reqs: DirtyWindowRequests, ) -> Result<(), Error>

Mark dirty time windows for batching flows.

Both timestamps and time_ranges ([start_inclusive, end_exclusive)) in each DirtyWindowRequest are bare i64s interpreted in the source table’s time index column native unit, resolved via table metadata.

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

impl BatchingEngine

fn batch_opts_for_flow_options( &self, flow_options: &HashMap<String, String>, ) -> Result<Arc<BatchingModeOptions>, Error>

fn table_options_enable_append_mode( extra_options: &HashMap<String, String>, ) -> bool

fn ensure_sql_flow_has_twe_or_eval_interval( eval_interval: Option<i64>, has_time_window_expr: bool, ) -> Result<(), Error>

SQL flows without a usable time-window expression can only run as an explicit full-query flow, so require EVAL INTERVAL at creation time.

fn ensure_incremental_source_append_only( batch_opts: &BatchingModeOptions, table_name: &[String; 3], extra_options: &HashMap<String, String>, ) -> Result<(), Error>

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

async fn check_is_tql_table( &self, query: &LogicalPlan, query_ctx: &QueryContext, ) -> Result<(), Error>

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

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

Only flush the dirty windows of the flow task with given flow id, by running the query on it. As flush the whole time range is usually prohibitively expensive.

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

Determine if the batching mode flow task exists with given flow id

async fn rollback_flow_runtime_if_current( &self, flow_id: u64, task: &BatchingTask, )

Trait Implementations

impl FlowEngine for BatchingEngine

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 FlowStatProvider for BatchingEngine

async fn flow_stat(&self) -> FlowStat

Returns current runtime stats of an engine.