Apply a UDF

Applies to

• Aerospike Developer SDK preview (Java 21+ and Python 3.10+)
• Aerospike Database 6.0 or later unless a section states otherwise

Code examples

Examples assume a connected session from Connect to Aerospike. Each usage guide lists shared imports in a tabbed block near the top of the page; snippet blocks add imports only for types not already shown. The first code block in each section includes dataset setup; later blocks in that section reuse those values. When a page includes a Complete example section, that block is fully self-contained.

Record UDFs run on a single record. The record may already exist or be created by the UDF. Register the Lua module on the cluster first. See Manage UDFs.

Except where noted, snippets on this page use the imports below. A snippet lists additional import lines only when it needs a type not shown here. When this page includes a Complete example section, that block is fully self-contained with every import required to run it.

Java

import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.Key;
import com.aerospike.client.sdk.RecordStream;
import com.aerospike.client.sdk.Session;

import java.util.Optional;

Python

from aerospike_sdk import DataSet

Invoke a UDF on one record

Chain executeUdf() / execute_udf(), function(), optional passing(), and execute(). The server passes the record as the first argument to the Lua function; values in passing() follow.

readBin function

The following Lua runs on the server and is shared by both the Java and Python examples. Define it in example.lua (see Manage UDFs):

function readBin(r, name)
    return r[name]
end

readBin returns the value of bin name on record r.

Java

DataSet udfDemo = DataSet.of("test", "udf_demo");
Key key = udfDemo.id("demo:1");

session.upsert(key).bin("name").setTo("Aerospike").execute();

RecordStream rs = session.executeUdf(key)
    .function("example", "readBin")
    .passing("name")
    .execute();

Optional<Object> value = rs.getFirstUdfResult();
System.out.println(value.orElse(null));

📖 API reference: Session.executeUdf | RecordStream.getFirstUdfResult

Python

udf_demo = DataSet.of("test", "udf_demo")
key = udf_demo.id("demo:1")

await session.upsert(key).put({"name": "Aerospike"}).execute()

stream = await (
    session.execute_udf(key)
    .function("example", "readBin")
    .passing("name")
    .execute()
)
value = await stream.first_udf_result()
print(value)

📖 API reference: Session.execute_udf | RecordStream.first_udf_result

Multiple arguments

When the UDF accepts extra parameters, pass them with passing().

multiplyAndAdd function

The following Lua runs on the server and is shared by both the Java and Python examples. Define it in example.lua (see Manage UDFs):

function multiplyAndAdd(r, a, b)
    return r['factor'] * a + b
end

multiplyAndAdd multiplies the record’s factor bin by a and adds b, so here 3 * 10 + 5 = 35.

Java

DataSet udfDemo = DataSet.of("test", "udf_demo");
Key key = udfDemo.id("demo:2");

session.upsert(key).bin("factor").setTo(3).execute();

RecordStream rs = session.executeUdf(key)
    .function("example", "multiplyAndAdd")
    .passing(10, 5)
    .execute();

Optional<Object> result = rs.getFirstUdfResult();
System.out.println(result.orElse(null));  // 35

Python

udf_demo = DataSet.of("test", "udf_demo")
key = udf_demo.id("demo:2")

await session.upsert(key).put({"factor": 3}).execute()

stream = await (
    session.execute_udf(key)
    .function("example", "multiplyAndAdd")
    .passing(10, 5)
    .execute()
)
result = await stream.first_udf_result()
print(result)  # 35

Batch UDF

Note

Server version 6.0.0 and above is required for batch UDF operations.

Pass multiple keys to run the same function on each record.

Java

DataSet udfDemo = DataSet.of("test", "udf_demo");
Key k1 = udfDemo.id("demo:10");
Key k2 = udfDemo.id("demo:11");

session.upsert(k1).bin("name").setTo("Alice").execute();
session.upsert(k2).bin("name").setTo("Bob").execute();

RecordStream rs = session.executeUdf(k1, k2)
    .function("example", "readBin")
    .passing("name")
    .execute();

while (rs.hasNext()) {
    var rr = rs.next();
    System.out.println(rr.udfResultOrThrow());
}

Python

udf_demo = DataSet.of("test", "udf_demo")
k1 = udf_demo.id("demo:10")
k2 = udf_demo.id("demo:11")

await session.upsert(k1).put({"name": "Alice"}).execute()
await session.upsert(k2).put({"name": "Bob"}).execute()

stream = await (
    session.execute_udf(k1, k2)
    .function("example", "readBin")
    .passing("name")
    .execute()
)
async for r in stream:
    print(r.udf_result)
stream.close()

Complete example: register, write, and read

This example registers example.lua, writes a record, and invokes readBin.

Java

import com.aerospike.client.sdk.Cluster;
import com.aerospike.client.sdk.ClusterDefinition;
import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.Key;
import com.aerospike.client.sdk.RecordStream;
import com.aerospike.client.sdk.Session;
import com.aerospike.client.sdk.policy.Behavior;
import com.aerospike.client.sdk.task.RegisterTask;

import java.util.Optional;

public class UdfReadBinExample {
    private static final String EXAMPLE_LUA = """
        function readBin(r, name)
            return r[name]
        end

        function multiplyAndAdd(r, a, b)
            return r['factor'] * a + b
        end
        """;

    public static void main(String[] args) {
        try (Cluster cluster = new ClusterDefinition("localhost", 3000).connect()) {
            Session session = cluster.createSession(Behavior.DEFAULT);

            RegisterTask task = session.registerUdfString(EXAMPLE_LUA, "example.lua");
            task.waitTillComplete();

            DataSet udfDemo = DataSet.of("test", "udf_demo");
            Key key = udfDemo.id("demo:1");

            session.upsert(key).bin("name").setTo("Aerospike").execute();

            RecordStream rs = session.executeUdf(key)
                .function("example", "readBin")
                .passing("name")
                .execute();

            Optional<Object> value = rs.getFirstUdfResult();
            System.out.println(value.orElse(null));
        }
    }
}

Python

import asyncio

from aerospike_sdk import Behavior, Client, DataSet

EXAMPLE_LUA = b"""\
function readBin(r, name)
    return r[name]
end

function multiplyAndAdd(r, a, b)
    return r['factor'] * a + b
end
"""

async def main():
    async with Client("localhost:3000") as client:
        task = await client.register_udf(EXAMPLE_LUA, "example.lua")
        await task.wait_till_complete()

        session = client.create_session(Behavior.DEFAULT)
        udf_demo = DataSet.of("test", "udf_demo")
        key = udf_demo.id("demo:1")

        await session.upsert(key).put({"name": "Aerospike"}).execute()

        stream = await (
            session.execute_udf(key)
            .function("example", "readBin")
            .passing("name")
            .execute()
        )
        value = await stream.first_udf_result()
        print(value)


if __name__ == "__main__":
    asyncio.run(main())

Complete example: multiplyAndAdd

Java

import com.aerospike.client.sdk.Cluster;
import com.aerospike.client.sdk.ClusterDefinition;
import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.Key;
import com.aerospike.client.sdk.RecordStream;
import com.aerospike.client.sdk.Session;
import com.aerospike.client.sdk.policy.Behavior;
import com.aerospike.client.sdk.task.RegisterTask;

import java.util.Optional;

public class UdfMultiplyAndAddExample {
    private static final String EXAMPLE_LUA = """
        function multiplyAndAdd(r, a, b)
            return r['factor'] * a + b
        end
        """;

    public static void main(String[] args) {
        try (Cluster cluster = new ClusterDefinition("localhost", 3000).connect()) {
            Session session = cluster.createSession(Behavior.DEFAULT);

            RegisterTask task = session.registerUdfString(EXAMPLE_LUA, "example.lua");
            task.waitTillComplete();

            DataSet udfDemo = DataSet.of("test", "udf_demo");
            Key key = udfDemo.id("demo:2");

            session.upsert(key).bin("factor").setTo(3).execute();

            RecordStream rs = session.executeUdf(key)
                .function("example", "multiplyAndAdd")
                .passing(10, 5)
                .execute();

            Optional<Object> result = rs.getFirstUdfResult();
            System.out.println(result.orElse(null));
        }
    }
}

Python

import asyncio

from aerospike_sdk import Behavior, Client, DataSet

EXAMPLE_LUA = b"""\
function multiplyAndAdd(r, a, b)
    return r['factor'] * a + b
end
"""

async def main():
    async with Client("localhost:3000") as client:
        task = await client.register_udf(EXAMPLE_LUA, "example.lua")
        await task.wait_till_complete()

        session = client.create_session(Behavior.DEFAULT)
        udf_demo = DataSet.of("test", "udf_demo")
        key = udf_demo.id("demo:2")

        await session.upsert(key).put({"factor": 3}).execute()

        stream = await (
            session.execute_udf(key)
            .function("example", "multiplyAndAdd")
            .passing(10, 5)
            .execute()
        )
        result = await stream.first_udf_result()
        print(result)


if __name__ == "__main__":
    asyncio.run(main())

Complete example: inline writeUnique

Register unique.lua from a string, invoke writeUnique twice on the same key, and observe that the second call does not overwrite an existing record. See Manage UDFs for the writeUnique definition.

Java

import com.aerospike.client.sdk.Cluster;
import com.aerospike.client.sdk.ClusterDefinition;
import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.Key;
import com.aerospike.client.sdk.Record;
import com.aerospike.client.sdk.RecordResult;
import com.aerospike.client.sdk.RecordStream;
import com.aerospike.client.sdk.Session;
import com.aerospike.client.sdk.policy.Behavior;
import com.aerospike.client.sdk.task.RegisterTask;

public class UdfWriteUniqueExample {
    private static final String UNIQUE_LUA = """
        function writeUnique(r, name, value)
            if not aerospike:exists(r) then
                aerospike:create(r)
                r[name] = value
                aerospike:update(r)
            end
        end
        """;

    public static void main(String[] args) {
        try (Cluster cluster = new ClusterDefinition("localhost", 3000).connect()) {
            Session session = cluster.createSession(Behavior.DEFAULT);

            RegisterTask task = session.registerUdfString(UNIQUE_LUA, "unique.lua");
            task.waitTillComplete();

            DataSet udfDemo = DataSet.of("test", "udf_demo");
            Key key = udfDemo.id("demo:3");

            session.delete(key).execute();

            for (int i = 0; i < 2; i++) {
                session.executeUdf(key)
                    .function("unique", "writeUnique")
                    .passing("status", "created")
                    .execute();
            }

            RecordStream rs = session.query(key).execute();
            RecordResult rr = rs.next();
            Record rec = rr.recordOrThrow();
            System.out.println(rec.getString("status"));
        }
    }
}

Python

import asyncio

from aerospike_sdk import Behavior, Client, DataSet

UNIQUE_LUA = b"""\
function writeUnique(r, name, value)
    if not aerospike:exists(r) then
        aerospike:create(r)
        r[name] = value
        aerospike:update(r)
    end
end
"""

async def main():
    async with Client("localhost:3000") as client:
        task = await client.register_udf(UNIQUE_LUA, "unique.lua")
        await task.wait_till_complete()

        session = client.create_session(Behavior.DEFAULT)
        udf_demo = DataSet.of("test", "udf_demo")
        key = udf_demo.id("demo:3")

        await session.delete(key).execute()

        for _ in range(2):
            await (
                session.execute_udf(key)
                .function("unique", "writeUnique")
                .passing("status", "created")
                .execute()
            )

        rec = await (await session.query(key).execute()).first_or_raise()
        print(rec.record.bins.get("status"))


if __name__ == "__main__":
    asyncio.run(main())