Batched commands

Multiple records can be read in a single batch call.

let bins = ["name", "age"];
let mut batch_reads = vec![];
for i in 0..10 {
  let key = as_key!("test", "test", i);
  batch_reads.push(BatchRead::new(key, &bins));
}
match client.batch_get(&BatchPolicy::default(), batch_reads).await {
    Ok(results) => {
      for result in results {
        match result.records {
          Some(record) => println!("{:?} => {:?}", result.key, record.bins),
          None => println!("No such record: {:?}", result.key),
        }
      }
    }
    Err(err) => println!("Error executing batch request: {}", err),
}

📖 API reference: Client::batch_get | BatchPolicy::default

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 all nodes return the record data, the records are returned to the caller. The list of records returned is in the same order the keys are passed in. If a record is not found in the database, the batch read’s record entry is None.

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().

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

let results = client.batch(&BatchPolicy::default(), &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

Each item returned by Client::batch() is a BatchRecord with fields key, record, result_code, and in_doubt. Check result_code, not a status field:

use aerospike::ResultCode;

for result in results {
    match result.result_code {
        Some(ResultCode::Ok) | None => {
            if let Some(record) = result.record {
                println!("{:?} => {:?}", result.key, record.bins);
            }
        }
        Some(code) => eprintln!("batch op failed for {:?}: {code:?}", result.key),
    }
}

📖 API reference: 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 = "2.0.0"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

In your src/main.rs file:

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").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()));
    }

    let results = client.batch(&BatchPolicy::default(), &batch_ops).await?;
    for result in results {
        match result.record {
            Some(record) => println!("{:?} => {:?}", result.key, record.bins),
            None => println!("No such record: {:?}", 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)}