Atomic operation to set minimum value in an Aerospike bin

user_582436 0
Viewed 225

I need an atomic 'set minimum' operation for Aerospike, where I give a bin name and a numeric argument, and whichever is lower, the current value of the bin or the argument, is set and returned.

The following Lua UDF should work

test.lua

function set_min(rec, bin_name, value)
    if aerospike:exists(rec) then
        local min = rec[bin_name]
        if min > value then
            rec[bin_name] = value
            aerospike:update(rec)
        end
    else
        rec[bin_name] = value
        aerospike:create(rec)
    end
    return rec[bin_name]
end

Run with the arguments 11, 9, 5, 7:

aql> execute test.set_min('minval', 11) on test.set-min where PK=2
+---------+
| set_min |
+---------+
| 11      |
+---------+
1 row in set (0.001 secs)

OK

aql> execute test.set_min('minval', 9) on test.set-min where PK=2
+---------+
| set_min |
+---------+
| 9       |
+---------+
1 row in set (0.001 secs)

OK

aql> execute test.set_min('minval', 5) on test.set-min where PK=2
+---------+
| set_min |
+---------+
| 5       |
+---------+
1 row in set (0.001 secs)

OK

aql> execute test.set_min('minval', 7) on test.set-min where PK=2
+---------+
| set_min |
+---------+
| 5       |
+---------+
1 row in set (0.000 secs)

Is there another way to do this?

aerospike user-defined-functions lua types
1 Answers

In any database, a user-defined function will run slower than native operations. This is no different in Aerospike, where Lua UDFs will have higher latencies and will not scale as well as native operations.

Aerospike's List and Map data types have extensive (and growing) APIs of atomic operations. These operations can be combined into a single multi-operation transaction (using the operate() method).

We can leverage ordered List to do the same atomic operation as the UDF above, in a way that runs faster and scales better.

set_min.py

from __future__ import print_function
import aerospike
from aerospike import exception as e
from aerospike_helpers.operations import list_operations as lh
import pprint
import sys

def set_min(bin_name, val):
    list_policy = {
        "list_order": aerospike.LIST_ORDERED,
        "write_flags": (aerospike.LIST_WRITE_ADD_UNIQUE |
                        aerospike.LIST_WRITE_PARTIAL |
                        aerospike.LIST_WRITE_NO_FAIL)
    }
    ops = [
        lh.list_append(bin_name, val, list_policy),
        lh.list_remove_by_rank_range(bin_name, 0, aerospike.LIST_RETURN_NONE,
            1, True),
        lh.list_get_by_rank(bin_name, 0, aerospike.LIST_RETURN_VALUE)
    ]
    return ops

config = {'hosts': [('172.16.39.132', 3000)]}
client = aerospike.client(config).connect()
pp = pprint.PrettyPrinter(indent=2)
key = ('test', 'set-min', 1)

key, meta, bins = client.operate(key, set_min('minval', 11))
pp.pprint(bins['minval'])

key, meta, bins = client.operate(key, set_min('minval', 9))
pp.pprint(bins['minval'])

key, meta, bins = client.operate(key, set_min('minval', 5))
pp.pprint(bins['minval'])

key, meta, bins = client.operate(key, set_min('minval', 7))
pp.pprint(bins['minval'])

client.close()

Run with the arguments 11, 9, 5, 7:

11
9
5
5
  1. Using an ordered list, a unique value is added to the list, gracefully failing if this value already exists. The list should have one or two elements now.
  2. The list is trimmed to hold only the lowest ranked element.
  3. The lowest ranked element (should be only one in the list) is returned.

These three operations happen atomically under a record lock.

For reference see the docs for the Python client.

user_5824360
Related