Struct Inserter

pub struct Inserter {
    catalog_manager: CatalogManagerRef,
    pub(crate) partition_manager: PartitionRuleManagerRef,
    pub(crate) node_manager: NodeManagerRef,
    pub(crate) table_flownode_set_cache: TableFlownodeSetCacheRef,
}

Fields

catalog_manager: CatalogManagerRef

partition_manager: PartitionRuleManagerRef

node_manager: NodeManagerRef

table_flownode_set_cache: TableFlownodeSetCacheRef

Implementations

impl Inserter

pub async fn handle_bulk_insert( &self, table: TableRef, decoder: &mut FlightDecoder, data: FlightData, ) -> Result<AffectedRows>

Handle bulk insert request.

fn maybe_update_flow_dirty_window( &self, table_info: TableInfoRef, record_batch: RecordBatch, )

impl Inserter

pub fn new( catalog_manager: CatalogManagerRef, partition_manager: PartitionRuleManagerRef, node_manager: NodeManagerRef, table_flownode_set_cache: TableFlownodeSetCacheRef, ) -> Self

pub async fn handle_column_inserts( &self, requests: InsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<Output>

pub async fn handle_row_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, accommodate_existing_schema: bool, is_single_value: bool, ) -> Result<Output>

Handles row inserts request and creates a physical table on demand.

pub async fn handle_log_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<Output>

Handles row inserts request and creates a log table on demand.

pub async fn handle_trace_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<Output>

pub async fn handle_last_non_null_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, accommodate_existing_schema: bool, is_single_value: bool, ) -> Result<Output>

Handles row inserts request and creates a table with last_non_null merge mode on demand.

async fn handle_row_inserts_with_create_type( &self, requests: RowInsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, create_type: AutoCreateTableType, accommodate_existing_schema: bool, is_single_value: bool, ) -> Result<Output>

Handles row inserts request with specified AutoCreateTableType.

pub async fn handle_metric_row_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, statement_executor: &StatementExecutor, physical_table: String, ) -> Result<Output>

Handles row inserts request with metric engine.

pub async fn handle_table_insert( &self, request: TableInsertRequest, ctx: QueryContextRef, ) -> Result<Output>

pub async fn handle_statement_insert( &self, insert: &Insert, ctx: &QueryContextRef, ) -> Result<Output>

impl Inserter

async fn do_request( &self, requests: InstantAndNormalInsertRequests, table_infos: &HashMap<TableId, Arc<TableInfo>>, ctx: &QueryContextRef, ) -> Result<Output>

async fn group_requests_by_peer( &self, requests: RegionInsertRequests, ) -> Result<HashMap<Peer, RegionInsertRequests>>

async fn create_or_alter_tables_on_demand( &self, requests: &mut RowInsertRequests, ctx: &QueryContextRef, auto_create_table_type: AutoCreateTableType, statement_executor: &StatementExecutor, accommodate_existing_schema: bool, is_single_value: bool, ) -> Result<CreateAlterTableResult>

Creates or alter tables on demand:

• if table does not exist, create table by inferred CreateExpr
• if table exist, check if schema matches. If any new column found, alter table by inferred AlterExpr

Returns a mapping from table name to table id, where table name is the table name involved in the requests. This mapping is used in the conversion of RowToRegion.

accommodate_existing_schema is used to determine if the existing schema should override the new schema. It only works for TIME_INDEX and single VALUE columns. This is for the case where the user creates a table with custom schema, and then inserts data with endpoints that have default schema setting, like prometheus remote write. This will modify the RowInsertRequests in place. is_single_value indicates whether the default schema only contains single value column so we can accommodate it.

async fn create_physical_table_on_demand( &self, ctx: &QueryContextRef, physical_table: String, statement_executor: &StatementExecutor, ) -> Result<()>

async fn get_table( &self, catalog: &str, schema: &str, table: &str, ) -> Result<Option<TableRef>>

fn get_create_table_expr_on_demand( &self, req: &RowInsertRequest, create_type: &AutoCreateTableType, ctx: &QueryContextRef, ) -> Result<CreateTableExpr>

fn get_alter_table_expr_on_demand( &self, req: &mut RowInsertRequest, table: &TableRef, ctx: &QueryContextRef, accommodate_existing_schema: bool, is_single_value: bool, ) -> Result<Option<AlterTableExpr>>

Returns an alter table expression if it finds new columns in the request. When accommodate_existing_schema is false, it always adds columns if not exist. When accommodate_existing_schema is true, it may modify the input req to accommodate it with existing schema. See create_or_alter_tables_on_demand for more details. When accommodate_existing_schema is true and is_single_value is true, it also consider fields when modifying the input req.

async fn create_physical_table( &self, create_table_expr: CreateTableExpr, partitions: Option<Partitions>, ctx: &QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<TableRef>

Creates a table with options.

async fn create_logical_tables( &self, create_table_exprs: Vec<CreateTableExpr>, ctx: &QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<Vec<TableRef>>