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

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

Performance

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

Insert Methods

The BulkInserter provides four insertion methods:

Method	Use Case	Queue Throughput
`insert`	Single record streaming	~3-7M rec/s
`insert_with_backpressure`	Memory-bounded streaming	~3-7M rec/s
`insert_batch`	Bulk loading	~20-30M rec/s
`insert_batch_arc`	Pre-wrapped records	~30-35M rec/s

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 and BulkInserter::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:

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:

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:

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

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

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