Batched commands

Multiple records can be read in a single batch call.

let read_policy = BatchReadPolicy::default();
let bins = Bins::from(["name", "age"]);
let mut batch_ops = vec![];
for i in 0..10 {
    let key = as_key!("test", "test", i);
    batch_ops.push(BatchOperation::read(&read_policy, key, bins.clone()));
}
use aerospike::ResultCode;

match client.batch(&BatchPolicy::default(), &mut batch_ops).await {
    Ok(()) => {
        for op in &batch_ops {
            let result = op.batch_record();
            match op.result_code() {
                Some(ResultCode::Ok) => {
                    if let Some(record) = &result.record {
                        println!("{:?} => {:?}", result.key, record.bins);
                    }
                }
                Some(ResultCode::KeyNotFoundError) => {
                    println!("No such record: {:?}", result.key);
                }
                Some(code) => {
                    eprintln!("batch read failed for {:?}: {code:?}", result.key);
                }
                None => {
                    eprintln!("no result for {:?}: row was not answered", result.key);
                }
            }
        }
    }
    Err(err) => println!("Error executing batch request: {}", err),
}

Some(Ok) is a record that came back. Some(KeyNotFoundError) is a missing key. None means that row was never answered. The batch call can still return Ok(()) when some keys are missing. 📖 API reference: Client::batch | BatchOperation::read | BatchPolicy::default | BatchRecord

This call groups keys based on which Aerospike Server node can best handle the request, and uses the client’s thread pool to concurrently handle all requests to each node. After the call returns, read each row from the BatchOperation values you passed in (same order as the keys). A missing record often has ResultCode::KeyNotFoundError with record set to None. See Per-record results.

Caution

Always check each row’s result_code and record, even when Client::batch() returns Ok(()). A successful batch call can still include per-key failures or missing records.

Batch write and delete

Client::batch() also supports writes and deletes in the same call. Each operation type uses its own policy type. Do not pass WritePolicy to BatchOperation::delete().

Caution

Batch writes and deletes apply per key. If one key in the batch fails, other keys in the same call are not rolled back. For all-or-nothing updates across multiple records, use Transactions.

Batch write

Pass a Vec<Operation> (for example from operations::put), not Vec<Bin>:

use aerospike::{
    as_bin, as_key, operations,
    BatchDeletePolicy, BatchOperation, BatchPolicy, BatchWritePolicy,
};

let write_policy = BatchWritePolicy::default();
let mut batch_ops = Vec::new();

let key = as_key!("test", "users", 1);
let ops = vec![
    operations::put(&as_bin!("name", "Ada")),
    operations::put(&as_bin!("score", 100_i64)),
];
batch_ops.push(BatchOperation::write(&write_policy, key, ops));

client.batch(&BatchPolicy::default(), &mut batch_ops).await?;

Batch delete

Use BatchDeletePolicy:

let delete_policy = BatchDeletePolicy::default();
let key = as_key!("test", "users", 1);
batch_ops.push(BatchOperation::delete(&delete_policy, key));

Per-record results

After Client::batch() returns, read each row from the BatchOperation values you passed in. Use BatchOperation::result_code(), record(), and related methods, or read the key and record fields and the result_code() and in_doubt() methods on the BatchRecord from batch_record(). Check result_code(), not a status field:

use aerospike::ResultCode;

client.batch(&BatchPolicy::default(), &mut batch_ops).await?;

for op in &batch_ops {
    let result = op.batch_record();
    match op.result_code() {
        Some(ResultCode::Ok) => {
            if let Some(record) = &result.record {
                println!("{:?} => {:?}", result.key, record.bins);
            }
        }
        Some(code) => eprintln!("batch op failed for {:?}: {code:?}", result.key),
        None => eprintln!("no result for {:?}: row was not answered", result.key),
    }
}

Stream results with batch_foreach

When you want a callback for each row as results arrive, use Client::batch_foreach(). Build a Vec<BatchOperation> and pass it as &mut; the callback runs once per row. Return false from the future to stop early.

client
    .batch_foreach(
        &BatchPolicy::default(),
        &mut batch_ops,
        |index, row| {
            println!("row {}: {:?}", index, row.result_code());
            async { true }
        },
    )
    .await?;

📖 API reference: Client::batch_foreach | BatchOperation::write | BatchOperation::delete | BatchWritePolicy | BatchDeletePolicy | BatchRecord | operations::put

Complete example

In your Cargo.toml file:

[package]
name = "aerospike-batch-example"
version = "0.1.0"
edition = "2021"

[dependencies]
aerospike = { version = "3.0.0", features = ["rt-tokio"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

In your src/main.rs file:

#[macro_use]
extern crate aerospike;

use aerospike::{
    as_bin, as_key, BatchOperation, BatchPolicy, BatchReadPolicy, Bins,
    Client, ClientPolicy, WritePolicy,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new(&ClientPolicy::default(), &"127.0.0.1:3000").await?;

    let records = vec![
        (1, "Ada", 31),
        (2, "Linus", 42),
        (3, "Grace", 37),
    ];

    for (id, name, age) in records {
        let key = as_key!("test", "users", id);
        let bins = vec![as_bin!("name", name), as_bin!("age", age)];
        client.put(&WritePolicy::default(), &key, &bins).await?;
    }

    let read_policy = BatchReadPolicy::default();
    let bin_names = Bins::Some(vec!["name".into(), "age".into()]);
    let mut batch_ops = Vec::new();
    for id in [1, 2, 3] {
        let key = as_key!("test", "users", id);
        batch_ops.push(BatchOperation::read(&read_policy, key, bin_names.clone()));
    }

    client.batch(&BatchPolicy::default(), &mut batch_ops).await?;
    use aerospike::ResultCode;
    for op in &batch_ops {
        let result = op.batch_record();
        match op.result_code() {
            Some(ResultCode::Ok) => {
                if let Some(record) = &result.record {
                    println!("{:?} => {:?}", result.key, record.bins);
                }
            }
            Some(ResultCode::KeyNotFoundError) => {
                println!("No such record: {:?}", result.key);
            }
            Some(code) => eprintln!("batch read failed for {:?}: {code:?}", result.key),
            None => eprintln!("no result for {:?}: row was not answered", result.key),
        }
    }

    client.close().await?;
    Ok(())
}

📖 API reference: Client::new | Client::put | BatchOperation::read | Client::batch | Client::close

Expected output

Output is similar to (bin order may vary):

Key { namespace: "test", set_name: "users", user_key: Some(Int(1)), .. } => {"name": String("Ada"), "age": Int(31)}
Key { namespace: "test", set_name: "users", user_key: Some(Int(2)), .. } => {"name": String("Linus"), "age": Int(42)}
Key { namespace: "test", set_name: "users", user_key: Some(Int(3)), .. } => {"name": String("Grace"), "age": Int(37)}