Struct LogicalTablePendingRowsBatcher 

pub struct LogicalTablePendingRowsBatcher {

    workers: Arc<WorkerRegistry<BatchKey, WorkerCommand>>,
    flush_interval: Duration,
    flush_policy: TimingFlushPolicy,
    partition_manager: PartitionRuleManagerRef,
    node_manager: NodeManagerRef,
    catalog_manager: CatalogManagerRef,
    flow_notification_tx: FlowNotifier,
    flush_limiter: FlushLimiter,
    request_limiter: RequestLimiter,
    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<WorkerRegistry<BatchKey, WorkerCommand>>

flush_interval: Duration

flush_policy: TimingFlushPolicy

partition_manager: PartitionRuleManagerRef

node_manager: NodeManagerRef

catalog_manager: CatalogManagerRef

flow_notification_tx: FlowNotifier

flush_limiter: FlushLimiter

request_limiter: RequestLimiter

worker_channel_capacity: usize

prom_store_with_metric_engine: bool

schema_alterer: PendingRowsSchemaAltererRef

pending_rows_batch_sync: bool

shutdown: Sender<()>

Implementations

impl LogicalTablePendingRowsBatcher

pub(super) async fn build_and_align_table_batches( &self, requests: &RowInsertRequests, ctx: &QueryContextRef, ) -> Result<(Vec<(String, u32, RecordBatchWithTsIdx)>, 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.

impl LogicalTablePendingRowsBatcher

pub(super) fn collect_non_empty_table_rows( requests: &RowInsertRequests, ) -> (Vec<(&str, &Rows)>, usize)

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

impl LogicalTablePendingRowsBatcher

pub(super) fn collect_unique_table_schemas<'a>( table_rows: &[(&'a str, &'a Rows)], ) -> Result<Vec<(&'a str, &'a [ColumnSchema])>>

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

impl LogicalTablePendingRowsBatcher

pub(super) 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.

impl LogicalTablePendingRowsBatcher

pub(super) async fn create_missing_tables_and_refresh_schemas( &self, catalog: &str, schema: &str, ctx: &QueryContextRef, table_rows: &[(&str, &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.

impl LogicalTablePendingRowsBatcher

pub(super) fn enqueue_alter_for_new_tables( table_rows: &[(&str, &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.

impl LogicalTablePendingRowsBatcher

pub(super) 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.

impl LogicalTablePendingRowsBatcher

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

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

impl LogicalTablePendingRowsBatcher

pub fn try_new( partition_manager: PartitionRuleManagerRef, node_manager: NodeManagerRef, catalog_manager: CatalogManagerRef, table_flownode_set_cache: TableFlownodeSetCacheRef, 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, flow_notification_queue_capacity: NonZeroUsize, ) -> Option<Arc<Self>>

impl LogicalTablePendingRowsBatcher

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

async fn physical_time_index_unit_or_default( &self, ctx: &QueryContextRef, ) -> TimeUnit

Returns the physical metric table’s time index unit resolved from ctx, defaulting to millisecond when the table does not exist yet (the schema alterer auto-creates it as millisecond).

pub(crate) async fn accepts_bulk_destinations( &self, batches: impl Iterator<Item = &(QueryContextRef, RowInsertRequests)>, ) -> bool

Returns whether the bulk path can accept batches: every existing destination table must be a metric logical table bound to the physical table selected by its context. A destination bound to another physical table would be flushed through the selected physical’s regions, silently misplacing its rows, so such requests must stay on the ordinary insert path (which routes per destination). New tables are always fine: they are created on the selected physical table. Time index units need no check here — the bulk encode converts each request to its destination’s unit. Destinations are resolved once per distinct (schema, table).

pub async fn submit_with<T, F>( &self, requests: RowInsertRequests, ctx: QueryContextRef, after_prepare: impl FnOnce(RowInsertRequests) -> F, ) -> Result<(u64, T)>

where F: Future<Output = Result<T>> + Send,

Submits with request-level accounting after schema preparation and before queue admission. Acknowledgement follows the global batching policy.

impl LogicalTablePendingRowsBatcher

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

impl LogicalTablePendingRowsBatcher

fn spawn_worker( &self, key: BatchKey, tx: Sender<WorkerCommand>, rx: Receiver<WorkerCommand>, )

Trait Implementations

impl Drop for LogicalTablePendingRowsBatcher

fn drop(&mut self)

Executes the destructor for this type.

impl PromWriteBatcher for LogicalTablePendingRowsBatcher

fn submit<'life0, 'async_trait>( &'life0 self, requests: RowInsertRequests, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,