pub struct BatchingSqlxClient  /* private fields */

A SQLx client that automatically batches INSERT statements and routes them through BulkInserter<GenericRecord> for high-performance shard-aware ingestion.

This client intercepts INSERT statements executed in a loop, accumulates them, and flushes them through the high-performance BulkInserter when the batch size is reached. Non-INSERT statements are executed immediately via SQLx.

Features

Auto-batching: INSERT statements are automatically accumulated
Shard-aware routing: Uses BulkInserter<GenericRecord> for multi-head ingest
Mixed statements: SELECT, CREATE, etc. work normally alongside INSERTs
Performance: Achieves high throughput for bulk INSERT operations

Example

use gpudb_api::sqlx_support::{BatchingSqlxClient, connect};
use gpudb_api::{Kinetica, KineticaOptions};
use std::sync::Arc;

// The SQLx pool for other statements, and the HTTP API client INSERT
// batches are flushed through
let pool = connect("localhost", "admin", "password").await?;
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 mut client = BatchingSqlxClient::new(pool, kinetica, 10000, 8);

// Execute INSERT statements in a loop - they get auto-batched
for i in 0..100000 {
    client.execute(&format!(
        "INSERT INTO my_table (id, value) VALUES ({}, {})", i, i * 10
    )).await?;
}

// Flush remaining records and get stats
let stats = client.close().await?;
println!("Inserted {} records at {:.0} rec/sec",
    stats.records_inserted, stats.records_per_second);

Implementations

Source

impl BatchingSqlxClient

Source

pub fn new( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, ) -> Self

Create a new batching SQLx client.

pool - SQLx PostgreSQL connection pool, for statements other than INSERT
kinetica - The Kinetica HTTP API client INSERT batches are flushed through
batch_size - Number of records to accumulate before flushing (recommended: 10000-50000)
num_workers - Number of parallel workers for BulkInserter (recommended: 4-16)

Source

pub fn with_options( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, options: BulkInserterOptions, ) -> Self

Create a batching client with custom options (e.g., upsert mode).

Source

pub fn with_hints( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, hints: SqlHints, ) -> Self

Create a batching client with SQL hints.

use gpudb_api::sqlx_support::{BatchingSqlxClient, connect};
use gpudb_api::{Kinetica, KineticaOptions, SqlHints};
use std::sync::Arc;

let pool = connect("localhost", "admin", "password").await?;
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 hints = SqlHints::new()
    .update_on_existing_pk(true)
    .distributed_operations(true);

let mut client = BatchingSqlxClient::with_hints(pool, kinetica, 10000, 8, hints);

Source

pub fn with_options_and_hints( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, options: BulkInserterOptions, hints: SqlHints, ) -> Self

Create a batching client with both BulkInserter options and SQL hints.

Source

pub fn set_hints(&mut self, hints: SqlHints)

Set SQL hints for the client.

Note: For INSERT statements, hints like update_on_existing_pk are automatically applied through the BulkInserter options.

Source

pub fn hints(&self) -> &SqlHints

Get the current SQL hints.

Source

pub async fn execute( &mut self, sql: &str, ) -> Result<Option<Vec<PgRow>>, BatchingSqlxError>

Execute a SQL statement.

INSERT statements are accumulated and batched automatically
Other statements (SELECT, CREATE, etc.) are executed immediately via SQLx

When the batch size is reached, records are automatically flushed through BulkInserter<GenericRecord> with shard-aware routing.

Source

pub async fn flush(&mut self) -> Result<(), BatchingSqlxError>

Flush all pending records for all tables.

Source

pub async fn close(self) -> Result<BatchingSqlxStats, BatchingSqlxError>

Flush all pending records and return final statistics.

This should be called when you’re done inserting to ensure all records are flushed.

Source

pub fn pending_count(&self) -> usize

Get the number of pending (unflushed) records across all tables.

Source

pub fn stats(&self) -> BatchingSqlxStats

Get current statistics without closing.

Source

pub fn pool(&self) -> &KineticaPool

Get a reference to the SQLx pool for direct queries.

​Features

​Example

​Implementations

​impl BatchingSqlxClient

​pub fn new( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, ) -> Self

​pub fn with_options( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, options: BulkInserterOptions, ) -> Self

​pub fn with_hints( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, hints: SqlHints, ) -> Self

​pub fn with_options_and_hints( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, options: BulkInserterOptions, hints: SqlHints, ) -> Self

​pub fn set_hints(&mut self, hints: SqlHints)

​pub fn hints(&self) -> &SqlHints

​pub async fn execute( &mut self, sql: &str, ) -> Result<Option<Vec<PgRow>>, BatchingSqlxError>

​pub async fn flush(&mut self) -> Result<(), BatchingSqlxError>

​pub async fn close(self) -> Result<BatchingSqlxStats, BatchingSqlxError>

​pub fn pending_count(&self) -> usize

​pub fn stats(&self) -> BatchingSqlxStats

​pub fn pool(&self) -> &KineticaPool

Features

Example

Implementations

impl BatchingSqlxClient

pub fn new( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, ) -> Self

pub fn with_options( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, options: BulkInserterOptions, ) -> Self

pub fn with_hints( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, hints: SqlHints, ) -> Self

pub fn with_options_and_hints( pool: KineticaPool, kinetica: Arc<Kinetica>, batch_size: usize, num_workers: usize, options: BulkInserterOptions, hints: SqlHints, ) -> Self

pub fn set_hints(&mut self, hints: SqlHints)

pub fn hints(&self) -> &SqlHints

pub async fn execute( &mut self, sql: &str, ) -> Result<Option<Vec<PgRow>>, BatchingSqlxError>

pub async fn flush(&mut self) -> Result<(), BatchingSqlxError>

pub async fn close(self) -> Result<BatchingSqlxStats, BatchingSqlxError>

pub fn pending_count(&self) -> usize

pub fn stats(&self) -> BatchingSqlxStats

pub fn pool(&self) -> &KineticaPool