pub struct SqlxBulkInserter  /* private fields */

Bulk inserter for SQLx that uses Kinetica’s HTTP API for efficient data loading.

SQLx uses the PostgreSQL wire protocol which is less efficient for bulk loading than Kinetica’s native HTTP API with Avro encoding. This helper bridges the gap by sending the records through a Kinetica client, connected and configured (TLS, HA failover and the rest) as for any other HTTP API use.

Uses GenericRecord for dynamic record creation.

Example

use gpudb_api::sqlx_support::SqlxBulkInserter;
use gpudb_api::{GenericRecord, Kinetica, KineticaError, KineticaOptions};
use std::sync::Arc;

// The HTTP API client the records are sent through
let mut options = KineticaOptions::new();
options.set_username("admin").set_password("password");
let kinetica = Arc::new(Kinetica::new(vec!["http://localhost:9191".to_string()], options).await?);

let bulk_inserter = SqlxBulkInserter::new(kinetica);

// Fetch type schema and create records
let record_type = Arc::new(bulk_inserter.fetch_type("my_table").await?);

// Setting a column fails on a name the type does not have
let records = (0..10000)
    .map(|i| {
        let mut r = record_type.new_record();
        r.set_by_name("id", i)?;
        r.set_by_name("name", format!("user_{i}"))?;
        Ok(r)
    })
    .collect::<Result<Vec<GenericRecord>, KineticaError>>()?;

// Bulk insert
let stats = bulk_inserter.insert("my_table", records, 10000).await?;
println!("Inserted {} records", stats.records_inserted);

Implementations

Source

impl SqlxBulkInserter

Source

pub fn new(kinetica: Arc<Kinetica>) -> Self

Create a bulk inserter that sends records through kinetica, a client of Kinetica’s HTTP API.

Source

pub fn kinetica(&self) -> &Arc<Kinetica>

Get a reference to the underlying Kinetica client.

Source

pub async fn fetch_type(&self, table_name: &str) -> Result<Type, KineticaError>

Fetch the Type for creating GenericRecords.

Returns a Type that can be used to create GenericRecord instances.

let record_type = Arc::new(inserter.fetch_type("my_table").await?);
let mut record = record_type.new_record();
record.set_by_name("id", 42)?;

Source

pub async fn insert( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, ) -> Result<BulkInsertStats, KineticaError>

Perform a bulk insert of GenericRecords with shard-aware multi-head ingest.

This uses the high-performance BulkInserter<GenericRecord> which provides:

Shard-aware routing to worker nodes (multi-head ingest)
Parallel batch processing
Avro binary encoding for maximum throughput

table_name - The name of the table to insert into
records - Vector of GenericRecords to insert
batch_size - Number of records per batch (recommended: 10000-50000)

let record_type = Arc::new(inserter.fetch_type("my_table").await?);
// Setting a column fails on a name the type does not have
let records = (0..1000).map(|i| {
    let mut r = record_type.new_record();
    r.set_by_name("id", i)?;
    Ok(r)
}).collect::<Result<Vec<GenericRecord>, KineticaError>>()?;
let stats = inserter.insert("my_table", records, 10000).await?;

Source

pub async fn insert_with_options( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, num_workers: usize, options: Option<BulkInserterOptions>, ) -> Result<BulkInsertStats, KineticaError>

Perform a bulk insert of GenericRecords with custom options and shard-aware routing.

Uses BulkInserter<GenericRecord> for optimal performance with multi-head ingest.

table_name - The name of the table to insert into
records - Vector of GenericRecords to insert
batch_size - Number of records per batch (recommended: 10000-50000)
num_workers - Number of parallel workers (recommended: 4-16)
options - Optional BulkInserterOptions for controlling insert behavior

Source

pub async fn upsert( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, ) -> Result<BulkInsertStats, KineticaError>

Perform a bulk insert with upsert mode (update on existing primary key).

Updates existing records if primary key already exists. Uses shard-aware multi-head ingest for optimal performance.

​Example

​Implementations

​impl SqlxBulkInserter

​pub fn new(kinetica: Arc<Kinetica>) -> Self

​pub fn kinetica(&self) -> &Arc<Kinetica>

​pub async fn fetch_type(&self, table_name: &str) -> Result<Type, KineticaError>

​pub async fn insert( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, ) -> Result<BulkInsertStats, KineticaError>

​pub async fn insert_with_options( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, num_workers: usize, options: Option<BulkInserterOptions>, ) -> Result<BulkInsertStats, KineticaError>

​pub async fn upsert( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, ) -> Result<BulkInsertStats, KineticaError>

Example

Implementations

impl SqlxBulkInserter

pub fn new(kinetica: Arc<Kinetica>) -> Self

pub fn kinetica(&self) -> &Arc<Kinetica>

pub async fn fetch_type(&self, table_name: &str) -> Result<Type, KineticaError>

pub async fn insert( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, ) -> Result<BulkInsertStats, KineticaError>

pub async fn insert_with_options( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, num_workers: usize, options: Option<BulkInserterOptions>, ) -> Result<BulkInsertStats, KineticaError>

pub async fn upsert( &self, table_name: &str, records: Vec<GenericRecord>, batch_size: usize, ) -> Result<BulkInsertStats, KineticaError>