Struct PendingRowsBatcher 

pub struct PendingRowsBatcher {

    workers: Arc<DashMap<BatchKey, PendingWorker>>,
    flush_interval: Duration,
    max_batch_rows: usize,
    partition_manager: PartitionRuleManagerRef,
    node_manager: NodeManagerRef,
    catalog_manager: CatalogManagerRef,
    flush_semaphore: Arc<Semaphore>,
    inflight_semaphore: Arc<Semaphore>,
    worker_channel_capacity: usize,
    prom_store_with_metric_engine: bool,
    schema_alterer: PendingRowsSchemaAltererRef,
    pending_rows_batch_sync: bool,
    shutdown: Sender<()>,
}

Prometheus remote write pending rows batcher.

Fields

workers: Arc<DashMap<BatchKey, PendingWorker>>

flush_interval: Duration

max_batch_rows: usize

partition_manager: PartitionRuleManagerRef

node_manager: NodeManagerRef

catalog_manager: CatalogManagerRef

flush_semaphore: Arc<Semaphore>

inflight_semaphore: Arc<Semaphore>

worker_channel_capacity: usize

prom_store_with_metric_engine: bool

schema_alterer: PendingRowsSchemaAltererRef

pending_rows_batch_sync: bool

shutdown: Sender<()>

Implementations

impl PendingRowsBatcher

pub fn try_new( partition_manager: PartitionRuleManagerRef, node_manager: NodeManagerRef, catalog_manager: CatalogManagerRef, prom_store_with_metric_engine: bool, schema_alterer: PendingRowsSchemaAltererRef, flush_interval: Duration, max_batch_rows: usize, max_concurrent_flushes: usize, worker_channel_capacity: usize, max_inflight_requests: usize, ) -> Option<Arc<Self>>

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

async fn build_and_align_table_batches( &self, requests: RowInsertRequests, ctx: &QueryContextRef, ) -> Result<(Vec<(String, u32, RecordBatch)>, usize)>

Converts proto RowInsertRequests directly into aligned RecordBatches in a single pass, handling table creation, schema alteration, column renaming, reordering, and null-filling without building intermediate RecordBatches.

fn collect_non_empty_table_rows( requests: RowInsertRequests, ) -> (Vec<(String, Rows)>, usize)

Extracts non-empty (table_name, rows) pairs and computes total row count across the retained entries.

fn collect_unique_table_schemas( table_rows: &[(String, Rows)], ) -> Result<Vec<(&str, &[ColumnSchema])>>

Returns unique (table_name, proto_schema) pairs while keeping the first-seen schema for duplicate table names.

async fn plan_table_resolution( &self, catalog: &str, schema: &str, ctx: &QueryContextRef, unique_tables: &[(&str, &[ColumnSchema])], ) -> Result<TableResolutionPlan>

Resolves table metadata and classifies each table into existing, to-create, and to-alter groups used by subsequent DDL steps.

async fn create_missing_tables_and_refresh_schemas( &self, catalog: &str, schema: &str, ctx: &QueryContextRef, table_rows: &[(String, Rows)], plan: &mut TableResolutionPlan, ) -> Result<()>

Batch-creates missing tables, refreshes their schema metadata, and enqueues follow-up alters for extra tag columns discovered in later rows.

fn enqueue_alter_for_new_tables( table_rows: &[(String, Rows)], plan: &mut TableResolutionPlan, ) -> Result<()>

For newly created tables, re-checks all row schemas and appends alter operations when additional tag columns are still missing.

async fn alter_tables_and_refresh_schemas( &self, catalog: &str, schema: &str, ctx: &QueryContextRef, plan: &mut TableResolutionPlan, ) -> Result<()>

Batch-alters tables that have missing tag columns and refreshes the in-memory schema map used for row alignment.

fn build_aligned_batches( table_rows: &[(String, Rows)], region_schemas: &HashMap<String, (Arc<ArrowSchema>, u32)>, ) -> Result<Vec<(String, u32, RecordBatch)>>

Converts proto rows to RecordBatch values aligned to resolved region schemas and returns (table_name, table_id, batch) tuples.

fn get_or_spawn_worker(&self, key: BatchKey) -> PendingWorker

fn spawn_worker(&self, key: BatchKey) -> PendingWorker

Trait Implementations

impl Drop for PendingRowsBatcher

fn drop(&mut self)

Executes the destructor for this type.