> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kinetica.com/llms.txt
> Use this file to discover all available pages before exploring further.

# bulk_inserter

> bulk_inserter — Kinetica Rust API

[gpudb\_api](https://docs.kinetica.com/content/api/rust/index.md)
[Source](https://github.com/kineticadb/kinetica-api-rust/blob/master/crates/gpudb-api/src/bulk_inserter.rs#L1-L5143) 

Shard-aware high-performance bulk insert using serde\_avro\_fast for encoding.`BulkInserter` routes records to the correct worker based on shard key computation, enabling true multi-head ingest where each record is sent directly to the worker that owns its shard.

## How Shard-Aware Routing Works

<ol>
  <li>On initialization, fetch the routing table via `admin_show_shards`</li>
  <li>Fetch the table schema via `show_table` to identify shard key columns</li>
  <li>Build a `RecordKeyBuilder` to compute routing keys</li>
  <li>For each record, compute its shard key hash using all shard key columns</li>
  <li>Use the routing table to determine which worker owns that shard</li>
  <li>Route the record directly to that worker</li>
</ol>

## Performance

<ul>
  <li>Shard-aware routing eliminates server-side record redistribution</li>
  <li>Each worker receives only records that belong to its shards</li>
  <li>Faster Avro encoding with `serde_avro_fast` (\~6-7x faster than apache-avro)</li>
  <li>Parallel encoding with rayon for maximum throughput</li>
</ul>

## Insert Methods

The `BulkInserter` provides four insertion methods:

<table>
  <thead>
    <tr>
      <th>Method</th>
      <th>Use Case</th>
      <th>Queue Throughput</th>
    </tr>
  </thead>

  <tbody>
    <tr>
      <td>[`insert`](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter#method.insert)</td>
      <td>Single record streaming</td>
      <td>\~3-7M rec/s</td>
    </tr>

    <tr>
      <td>[`insert_with_backpressure`](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter#method.insert_with_backpressure)</td>
      <td>Memory-bounded streaming</td>
      <td>\~3-7M rec/s</td>
    </tr>

    <tr>
      <td>[`insert_batch`](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter#method.insert_batch)</td>
      <td>Bulk loading</td>
      <td>\~20-30M rec/s</td>
    </tr>

    <tr>
      <td>[`insert_batch_arc`](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter#method.insert_batch_arc)</td>
      <td>Pre-wrapped records</td>
      <td>\~30-35M rec/s</td>
    </tr>
  </tbody>
</table>

The batch methods use parallel shard key computation and reduce lock contention by grouping records before acquiring locks. This provides up to 8x faster client-side queue throughput for large batches.

## JSON Records

[`BulkInserter::new_json`](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter#method.new_json) and [`BulkInserter::with_json_options`](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter#method.with_json_options) make a `BulkInserter<String>` taking JSON objects as text: each record goes to the rank of a random shard, and each batch is sent as one JSON array to `/insert/records/json`.

## Further Features

### Backpressure Control

Use `insert_with_backpressure()` for memory-bounded operation:

```rust theme={null}
let mut options = BulkInserterOptions::new();
options.set_max_in_flight_batches(50);  // Limit concurrent batches

let inserter = BulkInserter::with_options(kinetica, "my_table".into(), 50000, 8, Some(options)).await?;

for record in records {
    inserter.insert_with_backpressure(Arc::new(record)).await?;
}
```

### Timed Flush

Configure automatic periodic flush:

```rust theme={null}
let mut options = BulkInserterOptions::new();
options.set_flush_options(FlushOptions::new(false, 30));  // Flush every 30 seconds

let inserter = BulkInserter::with_options(kinetica, "my_table".into(), 50000, 8, Some(options)).await?;
// Timed flush runs automatically in background
```

### Error Queue

Non-blocking async error collection:

```rust theme={null}
// After insertion, check for errors
let errors = inserter.drain_errors();
for err in errors {
    eprintln!("Error: {} at {}", err.message, err.worker_url);
}
```

### Batch Callbacks

Real-time monitoring via `BatchInsertionListener`:

```rust theme={null}
use gpudb_api::{BatchInsertionListener, BatchInsertionResult};

struct MyListener;
impl BatchInsertionListener for MyListener {
    fn on_batch_inserted(&self, result: &BatchInsertionResult) {
        println!("Batch: {} records, success={}", result.batch_size, result.success);
    }
}
```

## Example

```rust theme={null}
use gpudb_api::{Kinetica, BulkInserter, BulkInserterOptions, KineticaRecord, ShardKeyExtractor, ShardKeyValue, ShardKeyValues};
use smallvec::smallvec;
use std::sync::Arc;

#[derive(serde::Serialize, Default)]
struct MyRecord { id: i32, name: String }

impl ShardKeyExtractor for MyRecord {
    fn shard_key_values(&self) -> ShardKeyValues<'_> {
        smallvec![("id", ShardKeyValue::Int(self.id))]
    }
}

// Simple usage with defaults
let inserter = BulkInserter::new(
    kinetica.clone(),
    "my_schema.my_table".into(),
    50000,  // batch size
    16      // num workers
).await?;

// Or with options
let mut options = BulkInserterOptions::default();
options.set_update_on_existing_pk(true);
let inserter_with_options = BulkInserter::<MyRecord>::with_options(
    kinetica,
    "my_schema.my_table".into(),
    50000,
    16,
    Some(options),
).await?;

// Single record insertion (good for streaming)
for i in 0..1_000_000 {
    let record = Arc::new(MyRecord { id: i, name: format!("name_{}", i) });
    inserter.insert(record)?;
}

// Or batch insertion (faster for bulk loading)
let records: Vec<MyRecord> = (0..100_000)
    .map(|i| MyRecord { id: i, name: format!("name_{}", i) })
    .collect();
inserter.insert_batch(records)?;

inserter.flush().await?;
```

## Re-exports

`pub use crate::bulk_inserter_common::ShardKeyExtractor;``pub use crate::bulk_inserter_common::ShardKeyValue;``pub use crate::bulk_inserter_common::ShardKeyValues;`

## Modules

[insert\_<wbr />records\_<wbr />options](https://docs.kinetica.com/content/api/rust/bulk_inserter/insert_records_options/index.md)Constants for InsertRecordsRequest options.

## Structs

[Backpressure<wbr />Metrics](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BackpressureMetrics.md)Metrics for monitoring backpressure behavior.[Batch<wbr />Insertion<wbr />Result](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BatchInsertionResult.md)Result of a batch insertion attempt. This is passed to the batch listener callback after each batch completes.[Bulk<wbr />Inserter](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserter.md)Shard-aware high-performance bulk inserter.[Bulk<wbr />Inserter<wbr />Options](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.BulkInserterOptions.md)Options for configuring the BulkInserter.[Flush<wbr />Options](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.FlushOptions.md)Options for controlling automatic flush behavior.[FnBatch<wbr />Listener](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.FnBatchListener.md)Simple function-based listener wrapper[Insert<wbr />Error](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.InsertError.md)An error that occurred during batch insertion. These are collected in the error queue for async error handling.[Queue<wbr />Error](https://docs.kinetica.com/content/api/rust/bulk_inserter/struct.QueueError.md)Records the inserter did not queue, and why, for the caller to retry or report.

## Traits

[Batch<wbr />Insertion<wbr />Listener](https://docs.kinetica.com/content/api/rust/bulk_inserter/trait.BatchInsertionListener.md)Callback trait for batch insertion events. Implement this trait to receive notifications when batches complete.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.