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,
    auto_create_table: bool,
    pending_rows_batcher: Option<Arc<dyn PendingRowsBatcher>>,
}

Fields

catalog_manager: CatalogManagerRef

partition_manager: PartitionRuleManagerRef

node_manager: NodeManagerRef

table_flownode_set_cache: TableFlownodeSetCacheRef

auto_create_table: bool

Server-side upper bound for auto table creation on write. When false, missing tables are never auto-created regardless of the per-request auto_create_table hint. When true, the hint still applies.

pending_rows_batcher: Option<Arc<dyn PendingRowsBatcher>>

Implementations

impl Inserter

pub async fn flush_bulk_batch( &self, table_info: TableInfoRef, batch: RecordBatch, ctx: QueryContextRef, ) -> Result<AffectedRows>

Routes and writes a prepared table batch, returning the affected row count.

Callers must exclude instant-TTL tables and handle Flow notifications after successful writes. This execution helper does not perform either step.

pub async fn handle_bulk_insert( &self, table: TableRef, raw_flight_data: FlightData, record_batch: RecordBatch, schema_bytes: Bytes, skip_wal: bool, channel: Channel, ) -> Result<AffectedRows>

Handle bulk insert request.

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

impl Inserter

pub async fn can_batch_metric_rows( &self, requests: &RowInsertRequests, ctx: &QueryContextRef, physical_table: &str, ) -> Result<bool>

Checks the assumptions of the logical bulk path without changing tables. Unsupported requests retain ordinary insertion, including schema policy, defaults, instant TTL and row-based Flow delivery.

pub async fn meter_row_inserts( requests: &mut RowInsertRequests, ctx: &QueryContextRef, ) -> Result<u64>

Meters an original logical-table request before bulk routing, without cloning its rows or changing the request boundary used for accounting.

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

pub fn with_pending_rows_batcher( self, batcher: Option<Arc<dyn PendingRowsBatcher>>, ) -> Self

Installs the shared batcher; callers explicitly select its ingestion entry point.

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.

async fn submit_pending_rows( &self, requests: RowInsertRequests, table_infos: HashMap<TableId, Arc<TableInfo>>, ctx: QueryContextRef, batcher: &Arc<dyn PendingRowsBatcher>, ) -> Result<Output>

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.

fn table_batcher( &self, table_info: &TableInfoRef, ctx: &QueryContextRef, ) -> Option<&Arc<dyn PendingRowsBatcher>>

async fn submit_table_rows( &self, rows: Rows, table_info: TableInfoRef, ctx: QueryContextRef, batcher: &Arc<dyn PendingRowsBatcher>, ) -> Result<Output>

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>>

fn auto_create_disabled_reason( &self, ctx: &QueryContextRef, ) -> Result<Option<&'static str>>

Returns Some(reason) if the config or request hint disables automatic table creation. Exempt private system tables are handled by Self::is_auto_create_exempt_private_table.

fn is_auto_create_exempt_private_table(schema: &str, table: &str) -> bool

Returns whether a private system table may infer and reconcile its schema even when automatic table creation is disabled.

pub async fn ensure_bulk_insert_schema( &self, table: TableRef, request_schema: &ArrowSchema, ctx: &QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<TableRef>

Adds missing columns from a bulk stream’s schema and returns the refreshed table. Call once when initializing the stream, before writing its first batch. Does not infer new nested or dictionary columns.

pub async fn ensure_trace_table_on_demand( &self, table_name: &str, request_schema: Vec<ColumnSchema>, alter_existing: bool, ctx: &QueryContextRef, statement_executor: &StatementExecutor, ) -> Result<()>

Ensures a trace table has the request-global schema without requiring a padded data row to drive on-demand creation or alteration. When alter_existing is false, a table created after planning is left for the caller to re-plan.

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, align_time_index_unit: Option<TimeUnit>, ) -> 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.

align_time_index_unit is the selected physical metric table’s time index unit; passing Some (metric engine path only) rewrites each request’s time index column, inside this function’s existing table lookups (no extra catalog access): existing destination tables are converted to their own unit, and new tables to the given unit.

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

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, semantic_index: Option<&PerTableSemanticIndex>, ) -> 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, alter_existing: 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 is_single_value is true, it also rejects native-histogram/float kind changes. When both options are true, it considers 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>>

pub fn node_manager(&self) -> &NodeManagerRef

pub fn partition_manager(&self) -> &PartitionRuleManagerRef

pub fn table_flownode_set_cache(&self) -> &TableFlownodeSetCacheRef