Zystkmt

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?

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.