Background queries

Jump to the Code block for a combined complete example.

A client application can also issue an asynchronous background query to the database and apply a series of write transaction operations to each record. This is more efficient than a query to retrieve records followed by updates to them for cases where data needs to be manipulated on the client side. Transactional operations are typically more efficient than using Lua UDFs because the server doesn’t need to translate internal objects to another language. Many client libraries also provide an API to poll for the completion of a background query.

See Background Queries for more information.

:::caution Production impact A background query writes to every record that matches the statement, across every node in the cluster. By default it runs at full cluster speed (RecordsPerSecond = 0), which can saturate node resources and raise latency for live traffic. The Go client has no API to cancel the job after QueryExecute returns.

Set RecordsPerSecond before you start the job, and run the query against a small set first to confirm both its scope and the write it applies. :::

Setup

The following examples use the shared setup and record structure to illustrate background queries in an Aerospike database.

import (
    "fmt"
    "log"

    "github.com/aerospike/aerospike-client-go/v8"
)

// Establishes a connection to the server
client, err := aerospike.NewClient("127.0.0.1", 3000)
if err != nil {
  log.Fatal(err)
}
defer client.Close()

The record structure:

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

Policies

Background queries can define policies to pass to the executed task.

The following example creates a policy that defines a filter expression looking for records that do not already have a numShapes bin.

// Create new write policy
writePolicy := aerospike.NewWritePolicy(0, 0)
writePolicy.FilterExpression = aerospike.ExpNot(
    aerospike.ExpBinExists("numShapes"))

queryPolicy := aerospike.NewQueryPolicy()
queryPolicy.RecordsPerSecond = 500

Query

Like basic queries, background queries can run on the primary index or, using a filter, on a secondary index.

Primary index

The following example creates a primary index background query using the Filter Expression defined in the policies example to find all records without a numShapes bin, then get the length of the shape key in the report map and write that value to a new bin called numShapes.

// Create statement
stmt := aerospike.NewStatement("sandbox", "ufodata")

exp := aerospike.ExpListSize(
    aerospike.ExpMapGetByKey(aerospike.MapReturnType.VALUE, aerospike.ExpTypeLIST, aerospike.ExpStringVal("shape"), aerospike.ExpMapBin("report")))

task, err := client.QueryExecute(queryPolicy, writePolicy, stmt, aerospike.ExpWriteOp("numShapes", exp, aerospike.ExpWriteFlagDefault))
if err != nil {
  log.Fatal(err)
}

Secondary index

The following example uses a secondary index created on the occurred integer bin. Use INTEGER for integer bins on Database 8.1.3 and later, or NUMERIC on earlier versions. Database 7.0 and later also support BLOB for []byte bins.

Caution

Creating a secondary index builds it across every data node and consumes CPU, memory, and I/O until the build completes, which raises latency for concurrent workloads. Create the index once before the application starts, during a maintenance window on a cluster serving live traffic. Do not call CreateIndex from application startup code that runs on every deployment.

idxTask, err := client.CreateIndex(nil, "sandbox", "ufodata", "occurred_idx", "occurred", aerospike.INTEGER)
if err != nil {
  log.Fatal(err)
}
<-idxTask.OnComplete()

The following example creates a secondary index background query using a filter expression that checks for a posted bin on records with an occurred value inclusively between 20210101 and 20211231, then updates the report map by adding a recent key with a value of true.

// Create new write policy
writePolicy := aerospike.NewWritePolicy(0, 0)
writePolicy.FilterExpression = aerospike.ExpBinExists("posted")

// Create statement
stmt := aerospike.NewStatement("sandbox", "ufodata")

// Set index filter
stmt.SetFilter(aerospike.NewRangeFilter("occurred", 20210101, 20211231))

task, err := client.QueryExecute(queryPolicy, writePolicy, stmt,
    aerospike.MapPutOp(aerospike.DefaultMapPolicy(), "report", aerospike.NewValue("recent"), aerospike.NewValue(true)))
if err != nil {
  log.Fatal(err)
}

Tracking

IsDone reports that the job finished running. It does not report whether every record was written successfully. Records can fail the write while the job still completes.

Wait on OnComplete rather than polling IsDone once. Persist task.TaskId() if the process can restart while the job is still running. After restart, reconstruct an ExecuteTask with that ID and the same statement (including any secondary-index filter) and continue to poll.

The statement must match the original because the client selects which server module to poll from the statement’s filter: a statement with no filter is treated as a scan, and one with a filter is treated as a query. Reconstructing the task with a missing or different filter polls the wrong module, so the job’s status is reported incorrectly or never resolves.

// Block until the job finishes or an error is returned.
if err := <-task.OnComplete(); err != nil {
    log.Fatal(err)
}
fmt.Print("Query complete")

taskID := task.TaskId()
stmt := aerospike.NewStatement("sandbox", "ufodata")
resumed := aerospike.NewExecuteTask(client.Cluster(), stmt, taskID)

if err := <-resumed.OnComplete(); err != nil {
    log.Fatal(err)
}

Code block

Expand this section for a single code block to execute a background query

import (
    "fmt"
    "log"

    "github.com/aerospike/aerospike-client-go/v8"
)

func main() {
    // Establishes a connection to the server
    client, err := aerospike.NewClient("127.0.0.1", 3000)
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    writePolicy := aerospike.NewWritePolicy(0, 0)
    writePolicy.FilterExpression = aerospike.ExpNot(
        aerospike.ExpBinExists("numShapes"))

    queryPolicy := aerospike.NewQueryPolicy()
    queryPolicy.RecordsPerSecond = 500

    // Create statement
    stmt := aerospike.NewStatement("sandbox", "ufodata")

    exp := aerospike.ExpListSize(
        aerospike.ExpMapGetByKey(aerospike.MapReturnType.VALUE, aerospike.ExpTypeLIST, aerospike.ExpStringVal("shape"), aerospike.ExpMapBin("report")))

    task, err := client.QueryExecute(queryPolicy, writePolicy, stmt, aerospike.ExpWriteOp("numShapes", exp, aerospike.ExpWriteFlagDefault))
    if err != nil {
        log.Fatal(err)
    }

    if err := <-task.OnComplete(); err != nil {
        log.Fatal(err)
    }
    fmt.Print("Query complete")
}