Aerospike: atomically create a list if it does not exists and append items to this list

user_2464031 0
Viewed 215

The bin in aerospike is a Map[String -> List]

I am trying to achieve the behavior as such:

bin.computeIfAbsent(key, k -> new List()).addAll(itemsToAdd)

Is there a way to do this atomically in Aerospike without implementing a UDF? If I read the documentation correctly, if it was a Map[String -> Map], I could have used CTX.mapKeyCreate to create the inner Map on demand, but I don't see anything similar for creating a List

UPD: Here is what I am trying to do

I have a stream of triplets:

{"pk","attr","value"}

I need to sink this stream into aerospike set aggregating by pk and attr in the following format:

{
   "PK":"pk",
   "mapBin": {
      "attr": ["value"]
    }
}

So let's say there are three items in the stream: {"pk","attr1","value1"},{"pk","attr2","value2"},{"pk","attr1","value3"}

They need to land in the aerospike as such:

{
   "PK":"pk",
   "mapBin": {
      "attr1": ["value1","value3"],
      "attr2": ["value2"]
    }
}

To insert a new item {"pk","attr2","value2"} I need to perform several actions:

  1. Get the list at mapBin[attr2]
  2. If it does not exist, insert empty List into the map
  3. Do ListOperation.append to append the item to existing list

Question is: is there a way to do it atomically without UDF?

aerospike java
2 Answers

You should be able to perform these actions atomically with the following (Java syntax):

record = client.operate(null, pk, 
    ListOperation.append("mapBin", Value.get("value2"), 
    CTX.mapKeyCreate(Value.get("attr2"), 
        MapOrder.UNORDERED)));

You can use ListOperation.appendItems() if there are multiple items to be appended to a list, and multiple ListOperations for different lists in the map within the same operate(), all of which will be executed atomically.

user_128169030

I tested the solution provided by @asnozzle above. (Server 5.7.0.11, Java client 5.1.11). (I put PK:pk as k:v in mapBin. I suppose you want it in its own bin.)

package com.aerospike;

import com.aerospike.client.*;
import com.aerospike.client.policy.WritePolicy;
import com.aerospike.client.cdt.MapPolicy;
import com.aerospike.client.cdt.CTX;
import com.aerospike.client.cdt.MapReturnType;
import com.aerospike.client.cdt.MapOrder;
import com.aerospike.client.cdt.ListOperation;
import com.aerospike.client.cdt.MapOperation;

public class TestExample {
    public static void main(String[] args) throws Exception {
      TestExample ce = new TestExample();
      ce.runExample();
      return;
    }

    public void runExample() throws Exception {
      AerospikeClient client = new AerospikeClient("127.0.0.1", 3000); 
      insertTuple(client, "pk", "attr1", "a1v1");
      insertTuple(client, "pk", "attr1", "a1v2");
      insertTuple(client, "pk", "attr2", "a2v1");
      insertTuple(client, "pk", "attr2", "a2v2");
      insertTuple(client, "pk", "attr2", "a2v3");
      insertTuple(client, "pk", "attr3", "a3v1");
    }
    private void insertTuple(AerospikeClient client, String s1, String s2, String s3) throws Exception {

        Key key = new Key("test", "testset", "k1");
        Record record = client.get(null, key);
        System.out.println("\nInitial Record: " + record);

        // Insert tuple data
        WritePolicy wPolicy = new WritePolicy(client.writePolicyDefault);
        MapPolicy mPolicy = new MapPolicy();
        client.operate(wPolicy, key, 
                       MapOperation.put(mPolicy, "mapBin", Value.get("PK"), Value.get(s1)),
                       ListOperation.append("mapBin", Value.get(s3), CTX.mapKeyCreate(Value.get(s2), MapOrder.UNORDERED))
                       ); 

        record = client.get(null,key);
        System.out.println("Get Record: " + record);
    }
}

And the output is:

Initial Record: null
Get Record: (gen:1),(exp:386490396),(bins:(mapBin:{attr1=[a1v1], PK=pk}))

Initial Record: (gen:1),(exp:386490396),(bins:(mapBin:{attr1=[a1v1], PK=pk}))
Get Record: (gen:2),(exp:386490396),(bins:(mapBin:{attr1=[a1v1, a1v2], PK=pk}))

Initial Record: (gen:2),(exp:386490396),(bins:(mapBin:{attr1=[a1v1, a1v2], PK=pk}))
Get Record: (gen:3),(exp:386490396),(bins:(mapBin:{attr2=[a2v1], attr1=[a1v1, a1v2], PK=pk}))

Initial Record: (gen:3),(exp:386490396),(bins:(mapBin:{attr2=[a2v1], attr1=[a1v1, a1v2], PK=pk}))
Get Record: (gen:4),(exp:386490396),(bins:(mapBin:{attr2=[a2v1, a2v2], attr1=[a1v1, a1v2], PK=pk}))

Initial Record: (gen:4),(exp:386490396),(bins:(mapBin:{attr2=[a2v1, a2v2], attr1=[a1v1, a1v2], PK=pk}))
Get Record: (gen:5),(exp:386490396),(bins:(mapBin:{attr2=[a2v1, a2v2, a2v3], attr1=[a1v1, a1v2], PK=pk}))

Initial Record: (gen:5),(exp:386490396),(bins:(mapBin:{attr2=[a2v1, a2v2, a2v3], attr1=[a1v1, a1v2], PK=pk}))
Get Record: (gen:6),(exp:386490396),(bins:(mapBin:{PK=pk, attr2=[a2v1, a2v2, a2v3], attr1=[a1v1, a1v2], attr3=[a3v1]}))
user_19171870
Related