Batched commands

For the complete documentation index see: llms.txt

All documentation pages available in markdown.

Jump to the Code block for a combined complete example.

Batched commands execute against multiple records issued as a single request. Batch reads support get, exists, getHeader, and operate requests. Batch writes, introduced in Aerospike 6.0.0, allow write requests against any keys, including updates, deletes, UDFs, and multi-operation operate commands.

Setup

The following examples will use the setup and record structure below to illustrate batch operations in an Aerospike database.

using Aerospike.Client;
using System;
using System.Collections;

// Define host configuration
Host config = new Host("127.0.0.1", 3000);
// Establishes a connection to the server
AerospikeClient client = new AerospikeClient(null, config);

The record structure:

Occurred: Integer
Reported: Integer
Posted: Integer
Report: Map
{
    shape: List,
    summary: String,
    city: String,
    state: String,
    duration: String
}
Location: GeoJSON

Policies

Policies are defined for the batch parent policy as well as batch read, batch write, batch delete, and batch UDF operations. Filter Expressions can be defined within each type of batch operation policy and the batch parent policy, along with other operation specific policies.

// Create a new batch policy
BatchPolicy batchPolicy = new BatchPolicy();
batchPolicy.filterExp = Exp.Build(
    // An example that will always return true
    Exp.GT(Exp.Val(2), Exp.Val(1))
);

// Create the batch write policy
BatchWritePolicy batchWritePolicy = new BatchWritePolicy();
batchWritePolicy.filterExp = Exp.Build(
    // An example that will always return true
    Exp.GT(Exp.Val(2), Exp.Val(1))
);

Two BatchPolicy fields shape how a batch actually runs:

maxConcurrentThreads

Defaults to 1, which sends each node’s portion of the batch serially. Set it to 0 to issue all node requests in parallel, or to a positive number to cap the worker threads. Raising it trades client threads and connections for latency, so size it against ClientPolicy.maxConnsPerNode.
respondAllKeys

Defaults to true, which tells the server to return a result for every key even after one of them fails. Setting it to false lets the server stop early, so some records come back with no result at all.

BatchPolicy batchPolicy = new(client.BatchParentPolicyWriteDefault)
{
  // 0 issues all node requests in parallel. Positive values cap the thread count.
  maxConcurrentThreads = 4
};

Handling partial failures

A batch is not all-or-nothing. Individual keys can fail while the rest of the batch succeeds, so a batch command that does not throw has not necessarily done everything you asked. Which signals you get depends on the overload:

Get(BatchPolicy, Key[]) returns Record[].

A null entry means the record was not found. There are no per-key result codes on this overload, so genuine failures throw instead.
Get(BatchPolicy, List<BatchRead>) and the Operate() overloads

Set BatchRecord.resultCode per key. ResultCode.KEY_NOT_FOUND_ERROR is an ordinary miss. Any other non-OK code is a failure for that key.
Operate() also reports an aggregate.

The BatchResults overload sets BatchResults.status, and the List<BatchRecord> overload returns a bool. Both are true only when every subcommand succeeded, and both are easy to discard by accident.
BatchRecord.inDoubt

Means a write may have been applied even though the client could not confirm it. Resolve an in-doubt record before retrying a non-idempotent operation.

An AerospikeException from a batch call signals a command-level or node-level failure, which is a different thing from a key that was not found. The examples below show each of these signals in place.

Requests

Exists

The following example creates an array of ten keys and checks for their existence in the database.

// Create batch of keys
Key[] keys = new Key[10];
for (int i = 0; i < 10; i++)
{
    keys[i] = new Key("sandbox", "ufodata", (i + 4995));
}

// Check if records exist
bool[] exists = client.Exists(batchPolicy, keys);

for (int i = 0; i < exists.Length; i++)
{
    if(!exists[i])
    {
        // Do something
        Console.WriteLine("Key: {0} does not exist", keys[i].userKey);
    }
}

// Close the connection to the server
client.Close();

Read records

The following example creates an array of ten keys and reads the records from the database; returning either the whole record or the specified report and location bins.

// Create batch of keys
Key[] keys = new Key[10];
for (int i = 0; i < 10; i++)
{
    keys[i] = new Key("sandbox", "ufodata", (i + 1));
}

// Read each whole record
Record[] records = client.Get(batchPolicy, keys);

// Or specifiy bins
// Record[] records = client.Get(batchPolicy, keys, "report", "location");

// Access the records
foreach (Record record in records)
{
    if(record != null)
    {
        // Do something
        Console.WriteLine("Record: {0}\\n", record.ToString().Split("bins:")[1]);
    }
}

// Close the connection to the server
client.Close();

Missing records and command failures

For list-based batch reads, inspect each record’s resultCode. KEY_NOT_FOUND_ERROR is an ordinary missing key. Other per-record codes indicate failures. A batch exception represents a command or node failure and is distinct from a missing record.

try
{
  client.Get(null, records);

  foreach (BatchRead read in records)
  {
    Key key = read.key;

    if (read.resultCode == ResultCode.OK)
    {
      Console.WriteLine($"Record: ns={key.ns} set={key.setName} key={key.userKey} bin={BinName} value={read.record?.GetValue(BinName)}");
    }
    else if (read.resultCode == ResultCode.KEY_NOT_FOUND_ERROR)
    {
      Console.WriteLine($"Record not found: ns={key.ns} set={key.setName} key={key.userKey}");
    }
    else
    {
      Console.Error.WriteLine($"Record failed: key={key.userKey} result={ResultCode.GetResultString(read.resultCode)}");
    }
  }
}
catch (AerospikeException ae)
{
  // A batch exception represents a command/node failure, not an ordinary missing key.
  Console.Error.WriteLine($"Batch command failed: {ae.Message}");
  throw;
}

Read commands

The following example creates an array of ten keys and accesses the city and state map keys to return their respective values from the report bin, for each record.

// Create batch of keys
Key[] keys = new Key[10];
for (int i = 0; i < 10; i++)
{
    keys[i] = new Key("sandbox", "ufodata", (i + 1));
}

// Create map key list
string[] mapKeys = {"city", "state"};

// Get 'city' and 'state' from report map for each record
BatchResults batchResult = client.Operate(batchPolicy, batchWritePolicy, keys,
    MapOperation.GetByKeyList("report", mapKeys, MapReturnType.VALUE)
);

// Access the records
foreach (BatchRecord batchRecord in batchResult.records)
{
    Record record = batchRecord.record;
    if(record != null)
    {
        // Do something
        Console.WriteLine("Record: {0}\\n", record.ToString().Split("bins:")[1]);
    }
}

// Close the connection to the server
client.Close();

Inspect aggregate and per-record status

BatchResults.status reports whether every subcommand succeeded. Always retain it, then inspect resultCode and inDoubt for each failed record. inDoubt means a write may have been applied even though the client could not confirm the result.

BatchResults batch = client.Operate(batchPolicy, null, keys,
  ListOperation.Append(ListPolicy.Default, BinName3, Value.Get(999)),
  ListOperation.Size(BinName3),
  ListOperation.GetByIndex(BinName3, -1, ListReturnType.VALUE));

Console.WriteLine($"All batch sub-commands succeeded: {batch.status}");

for (int i = 0; i < batch.records.Length; i++)
{
  BatchRecord br = batch.records[i];

  if (br.resultCode == 0)
  {
    IList results = br.record.GetList(BinName3);
    Console.WriteLine($"Result[{i}]: {results[1]},{results[2]}");
  }
  else
  {
    Console.WriteLine(
      $"Result[{i}]: error={ResultCode.GetResultString(br.resultCode)} " +
      $"inDoubt={br.inDoubt}");
  }
}

Read/write commands

The following example creates an array of ten keys and

Defines an Operation Expression that compares the occurred bin value against the provided value, 20211231, and verifies the posted bin exists to determine the boolean value of the new recent key being added to the report map.
Returns the report bin.

// Create batch of keys
Key[] keys = new Key[10];
for (int i = 0; i < 10; i++)
{
    keys[i] = new Key("sandbox", "ufodata", (i + 1));
}

// Define Operation Expressions
Expression exp = Exp.Build(
    MapExp.Put(MapPolicy.Default, Exp.Val("recent"),
        Exp.And(
            Exp.GT(Exp.IntBin("occurred"), Exp.Val(20211231)),
            Exp.BinExists("posted")
        ),
        Exp.MapBin("report")
    )
);

// Execute the write operation and return the report bin
BatchResults batchResult = client.Operate(batchPolicy, batchWritePolicy, keys,
    ExpOperation.Write("report", exp, ExpWriteFlags.DEFAULT),
    Operation.Get("report")
);

// Access the records
foreach (BatchRecord batchRecord in batchResult.records)
{
    Record record = batchRecord.record;
    if(record != null)
    {
        // Do something
        Console.WriteLine("Record: {0}\\n", record.ToString().Split("bins:")[1]);
    }
}

// Close the connection to the server
client.Close();

Deletes

The following example deletes the records from the database.

// Create batch of keys
Key[] keys = new Key[10];
for (int i = 0; i < 10; i++)
{
    keys[i] = new Key("sandbox", "ufodata", (i + 1));
}

// Delete records passing null to use the default BatchDeletePolicy
BatchResults batchResults = client.Delete(batchPolicy, null, keys);

// Close the connection to the server
client.Close();

Complex batched commands

The following example creates a list of four batch records that each use a differing set of operations.

The record with user defined key `4000`

uses the ops1 array that combines the Operation Expression.
uses exp1 which compares the occurred bin value against the provided value, 20211231 and verifies the posted bin exists to determine the boolean value of the new recent key being added to the report map.
returns the report bin.

The record with user defined key `4001`

uses the ops2 array which contains a read Operation Expression that gets the length of the shape list from the report map and returns the value in a computed bin named numShapes.

The record with user defined key `4002`

uses the ops3 array which combines a write operation that updates the posted bin value, with a map operation that updates the city value in the report map.
returns both the posted and report bins.

The record with user defined key `4003` is deleted from the database.

// Define Operation Expressions
Expression exp1 = Exp.Build(
    MapExp.Put(MapPolicy.Default, Exp.Val("recent"),
        Exp.And(
            Exp.GT(Exp.IntBin("occurred"), Exp.Val(20211231)),
            Exp.BinExists("posted")
        ),
        Exp.MapBin("report")
    )
);
Expression exp2 = Exp.Build(
    ListExp.Size(
        MapExp.GetByKey(MapReturnType.VALUE, Exp.Type.LIST, Exp.Val("shape"), Exp.MapBin("report"))
    )
);

// Define operations
Operation[] ops1 = Operation.Array(
    ExpOperation.Write("report", exp1, ExpWriteFlags.DEFAULT),
    Operation.Get("report")
);
Operation[] ops2 = Operation.Array(ExpOperation.Read("numShapes", exp2, ExpReadFlags.DEFAULT));
Operation[] ops3 = Operation.Array(
    Operation.Put(new Bin("posted", 20201108)),
    MapOperation.Put(MapPolicy.Default, "report", Value.Get("city"), Value.Get("Cedarville")),
    Operation.Get("posted"),
    Operation.Get("report")
);

// Create list of batch records to process
List<BatchRecord> batchRecordList = new List<BatchRecord>();
batchRecordList.Add(new BatchWrite(new Key("sandbox", "ufodata", 4000), ops1));
batchRecordList.Add(new BatchRead(new Key("sandbox", "ufodata", 4001), ops2));
batchRecordList.Add(new BatchWrite(new Key("sandbox", "ufodata", 4002), ops3));
batchRecordList.Add(new BatchDelete(new Key("sandbox", "ufodata", 4003)));

// Process the batch and retain whether every sub-command succeeded.
bool allSucceeded = client.Operate(batchPolicy, batchRecordList);
Console.WriteLine("All batch sub-commands succeeded: {0}", allSucceeded);

// Access the results
foreach (BatchRecord batchRecord in batchRecordList)
{
    Record record = batchRecord.record;
    if (batchRecord.resultCode == ResultCode.OK)
    {
        // Do something
        Console.WriteLine("Record: {0}\\n", record?.ToString().Split("bins:")[1]);
    }
    else
    {
        Console.WriteLine("Result: {0} | In doubt: {1}",
            ResultCode.GetResultString(batchRecord.resultCode), batchRecord.inDoubt);
    }
}

// Close the connection to the server
client.Close();

Code block

Expand this section for a single code block to execute a batch read/write operation

using Aerospike.Client;
using System;
using System.Collections;

// Define host configuration
Host config = new Host("127.0.0.1", 3000);
// Establishes a connection to the server
AerospikeClient client = new AerospikeClient(null, config);

// Create a new batch policy
BatchPolicy batchPolicy = new BatchPolicy();
batchPolicy.filterExp = Exp.Build(
    // An example that will always return true
    Exp.GT(Exp.Val(2), Exp.Val(1))
);

// Create the batch write policy
BatchWritePolicy batchWritePolicy = new BatchWritePolicy();
batchWritePolicy.filterExp = Exp.Build(
    // An example that will always return true
    Exp.GT(Exp.Val(2), Exp.Val(1))
);

// Create batch of keys
Key[] keys = new Key[10];
for (int i = 0; i < 10; i++)
{
    keys[i] = new Key("sandbox", "ufodata", (i + 1));
}

// Define Operation Expressions
Expression exp = Exp.Build(
    MapExp.Put(MapPolicy.Default, Exp.Val("recent"),
        Exp.And(
            Exp.GT(Exp.IntBin("occurred"), Exp.Val(20211231)),
            Exp.BinExists("posted")
        ),
        Exp.MapBin("report")
    )
);

// Execute the write operation and return the report bin
BatchResults batchResult = client.Operate(batchPolicy, batchWritePolicy, keys,
    ExpOperation.Write("report", exp, ExpWriteFlags.DEFAULT),
    Operation.Get("report")
);

// Access the records
foreach (BatchRecord batchRecord in batchResult.records)
{
    Record record = batchRecord.record;
    if(record != null)
    {
        // Do something
        Console.WriteLine("Record: {0}\\n", record.ToString().Split("bins:")[1]);
    }
}

// Close the connection to the server
client.Close();