BulkInserter

Creates a new shard-aware BulkInserter instance with default options.

This is a convenience method that calls with_options with None for options. For more control over insert behavior, use with_options directly.

• kinetica - Shared Kinetica client
• table_name - Target table for inserts (including schema, e.g., “my_schema.my_table”)
• batch_size - Records per batch (recommended: 50000)
• num_workers - Number of flush workers, i.e. of batches sent concurrently; used exactly as given. 0 sizes the pool at one worker per available CPU.

Creates a new shard-aware BulkInserter instance with the specified options.

This constructor:

1. Validates user permissions for the table
2. Validates batch size
3. Checks if the table is replicated
4. Configures multi-head ingest (if available and table is not replicated)
5. Sets up worker queues and routing

• kinetica - Shared Kinetica client
• table_name - Target table for inserts (including schema, e.g., “my_schema.my_table”)
• batch_size - Records per batch (recommended: 50000); must be > 0
• num_workers - Number of flush workers, i.e. of batches sent concurrently; used exactly as given. 0 sizes the pool at one worker per available CPU.
• options - Optional BulkInserterOptions for configuring insert behavior

Returns an error if:

• User doesn’t have required permissions (insert, and update if using upsert)
• Table doesn’t exist
• Batch size is less than 1
• Unable to configure multi-head ingest

Creates a new shard-aware BulkInserter instance with an explicit schema.

This constructor is useful for types that cannot implement Default, such as GenericRecord which requires a Type reference at construction time.

• kinetica - Shared Kinetica client
• table_name - Target table for inserts (including schema, e.g., “my_schema.my_table”)
• batch_size - Records per batch (recommended: 50000); must be > 0
• num_workers - Number of flush workers, i.e. of batches sent concurrently; used exactly as given. 0 sizes the pool at one worker per available CPU.
• options - Optional BulkInserterOptions for configuring insert behavior
• schema_str - The Avro schema JSON string for the record type

use gpudb_api::{BulkInserter, GenericRecord, Type};

let record_type = Arc::new(Type::from_table(&kinetica, "my_table").await?);
let schema_str = record_type.schema_string();

let inserter = BulkInserter::<GenericRecord>::with_schema(
    kinetica,
    "my_table".into(),
    50000,
    8,
    None,
    schema_str,
).await?;

Creates a JSON inserter with the default JsonOptions and options.

num_workers is as for BulkInserter::with_schema.

Creates a JSON inserter. With JsonOptions::is_validate_json, each record inserted must be a JSON object.

Inserts a single record with shard-aware routing.

The record’s shard key is computed using ShardKeyExtractor::shard_key_values(), and the record is routed to the worker that owns the corresponding shard.

Returns an error if the record cannot be routed — see Self::route_to_worker. The record is not queued in that case, and comes back in the QueueError.

Also returns an error after close when this record fills a batch: the batch cannot be sent, and its records come back.

This method uses ArcSwap::load() to get the current routing state, which provides:

• Lock-free atomic read of the routing configuration
• Consistent routing even during HA failover (no torn reads)
• No race conditions with concurrent routing updates

A JSON inserter whose JsonOptions ask for it first checks that the record is a JSON object.

Inserts a single record with backpressure control.

This async method will wait if the maximum in-flight batches limit is reached, providing memory-bounded operation for high-throughput scenarios.

This method uses ArcSwap::load() to get the current routing state atomically.

Inserts a batch of records with shard-aware routing.

This method is more efficient than calling insert() in a loop because:

1. Worker indices are computed in parallel using rayon
2. Records are grouped by destination before any lock is taken
3. Each destination’s queue is locked once per call, not once per record

The records go into the calling thread’s stripe of each queue, as insert puts them, and a batch is sent each time that stripe reaches the batch size. No batch is larger than the batch size.

• records - A vector of records to insert. Each record will be wrapped in Arc internally.

The records before the first that cannot be routed are queued, and that record and every one after it come back in the QueueError with its error. A list whose JSON does not validate is not queued at all, and comes back whole.

let records: Vec<MyRecord> = (0..100_000)
    .map(|i| MyRecord { id: i, name: format!("name_{}", i) })
    .collect();
inserter.insert_batch(records)?;

Inserts a batch of records that are already wrapped in Arc.

This is useful when records are shared or when the caller already has Arc-wrapped records.

• records - A vector of Arc-wrapped records to insert.

Returns the total number of records successfully inserted.

Returns the total number of records successfully updated (for upserts).

Returns the table name.

Returns the batch size.

Returns whether the target table is replicated.

Returns whether multi-head ingest is enabled.

Returns whether the head node is being used (replicated table or multi-head disabled).

Returns the current maximum retries for failed insertions.

Sets the maximum retries for failed insertions.

Returns a reference to the insert options.

The JSON options of a JSON inserter, or those set since.

Replaces the JSON options, which a JSON inserter reads for each record it is given. They do not make an inserter of Avro records a JSON one.

Returns the flush options.

The timed flush options.

Replaces the timed flush options, None restoring the defaults (no timed flush), and restarts the timed flush with them. A scheduled timed flush is stopped first, and a new one starts if the new interval is positive, its first run one interval from now.

Returns whether return_individual_errors mode is active.

Returns the number of records currently buffered in worker queues.

These are records that have been queued via insert() but not yet sent to the flush workers. Call flush() to send these records.

Returns the estimated number of records in-flight (sent but not yet confirmed).

This is an estimate based on pending_batches * batch_size. The actual number may be slightly different due to partial batches.

The connection this inserter sends through.

Drains and returns all queued errors. This clears the error queue, so subsequent calls return new errors only.

Returns the number of queued errors without draining.

Returns the total number of records that failed to insert. This sums the record_count from all errors in the queue.

Drains and returns all queued warnings.

Returns the number of queued warnings without draining.

Returns whether timed flush is currently running.

Stops the timed flush task.

Returns the maximum number of in-flight batches configured.

Returns the number of available in-flight permits.

Returns backpressure metrics for monitoring.

This provides visibility into the backpressure state of the inserter, useful for monitoring and tuning max_in_flight_batches.

let metrics = inserter.backpressure_metrics();
println!("In-flight: {}/{} ({:.1}% utilized)",
    metrics.in_flight_batches,
    metrics.max_in_flight_batches,
    metrics.utilization_pct
);

Flushes all buffered records to the server and waits until they have been acknowledged.

When this returns Ok(()), every record queued before the call has either been inserted or recorded in the error queue — so a subsequent read is guaranteed to see them.

Check error_count or drain_errors afterwards to see whether any records failed; a returned Ok means “all batches finished”, not “all batches succeeded”.

Use flush_no_wait if you deliberately want to hand the batches off without waiting.

Hands all buffered records to the flush workers without waiting for them to complete.

Prefer flush — after this returns, the records have not necessarily reached the server, so reading the table back may not see them. This exists for background flushing (the timed-flush task uses it) where blocking would stall the caller.

Returns the number of pending batches (sent but not yet completed).

Closes the BulkInserter, flushing all records and waiting for completion.

This method:

1. Flushes all buffered records to the flush workers
2. Waits for all pending batches to complete
3. Stops the flush workers and the timed flush

The inserter cannot send anything afterwards: an insert still buffers, but one that fills a batch, or a flush() with records buffered, fails.

IMPORTANT: This method will wait indefinitely until all pending batches complete. It only returns when:

1. All records have been successfully inserted, OR
2. All records have either succeeded or failed (with errors queued)

Failed records are tracked in the error queue and can be retrieved via drain_errors(). The method guarantees that inserted + updated + failed = queued.

Returns Ok(()) when all batches have completed (success or failure). Check error_count() or drain_errors() to see if any records failed.

Closes the BulkInserter with an optional timeout.

If timeout is reached, returns an error but records may still be in-flight. Prefer using close() without timeout to ensure all records are flushed.

• timeout - Maximum time to wait for pending batches to complete

Returns Ok(()) if all batches completed, or Err if timeout was reached.

Returns the number of shard key columns used for routing.

Returns the names of shard key columns.

Returns the routing table size (number of shards).

Returns the number of worker queues.

Returns the number of times the connection has switched clusters due to failover.

Returns the current HA ring size (number of clusters).

Returns the current head node URL.