adityamukho · Oct 25, 2026 · Oct 25, 2026 · Oct 25, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -17,7 +17,7 @@ path = "src/uniffi_bindgen.rs"
 [dependencies]
 # Pin to the exact released version. The release workflow updates this line.
 # Unreleased v3 core (format v8). Switches to minigraf = "=3.0.0" at the v3.0.0 release.
-minigraf = { git = "https://github.com/project-minigraf/minigraf", rev = "7c8c554096209ddc2bf1342b9a4eef3ef77c67c3" }
+minigraf = { git = "https://github.com/project-minigraf/minigraf", rev = "02243f1656eedd987611a9babdfee4a8c6e22b6b" }
 uniffi = { version = "0.31.1", features = ["cli"] }
 thiserror = "2.0.18"
 serde_json = "1.0.149"

diff --git a/consumer-test/src/test/java/ContractTest.java b/consumer-test/src/test/java/ContractTest.java
@@ -3,6 +3,7 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import io.github.project_minigraf.minigraf.MiniGrafAccumulator;
 import io.github.project_minigraf.minigraf.MiniGrafDb;
 import io.github.project_minigraf.minigraf.MiniGrafException;
 import io.github.project_minigraf.minigraf.MiniGrafPreparedQuery;
@@ -67,6 +68,40 @@ void setValuesComeBackAndBind() throws Exception {
         }
     }
 
+    @Test
+    void userPredicateAndAggregate() throws Exception {
+        try (MiniGrafDb db = Minigraf.openInMemory()) {
+            db.execute("(transact [[:alice :name \"Alice\"] [:alice :age 30] [:bob :name \"Bob\"] [:bob :age 12]])");
+            db.registerPredicate("adult?", v -> ((MiniGrafValue.Int64) v).getValue() >= 18);
+            db.registerAggregate("product", () -> new MiniGrafAccumulator() {
+                private long product = 1;
+
+                @Override
+                public void step(MiniGrafValue v) {
+                    long n = ((MiniGrafValue.Int64) v).getValue();
+                    if (n == 0) {
+                        throw new IllegalArgumentException("zero");
+                    }
+                    product *= n;
+                }
+
+                @Override
+                public MiniGrafValue finish(long count) {
+                    return new MiniGrafValue.Int64(product);
+                }
+            });
+            String adults = db.execute("(query [:find ?n :where [?e :age ?a] [(adult? ?a)] [?e :name ?n]])");
+            assertTrue(adults.contains("\"results\":[[\"Alice\"]]"), "a user predicate filters rows");
+            String product = db.execute("(query [:find (product ?a) :where [?e :age ?a]])");
+            assertTrue(product.contains("\"results\":[[360]]"), "a user aggregate folds values");
+            db.execute("(transact [[:carol :age 0]])");
+            MiniGrafException e = assertThrows(MiniGrafException.Query.class,
+                    () -> db.execute("(query [:find (product ?a) :where [?e :age ?a]])"));
+            String msg = Minigraf.errorMessage(e);
+            assertTrue(msg.startsWith("[QRY-014]") && msg.contains("zero"), "a throwing UDF is QRY-014");
+        }
+    }
+
     @Test
     void invalidDatalogThrowsMiniGrafException() throws Exception {
         try (MiniGrafDb db = Minigraf.openInMemory()) {

diff --git a/java/README.md b/java/README.md
@@ -162,6 +162,73 @@ values come back as JSON arrays from `execute` and as `MiniGrafValue.Set` in fac
 Errors are `MiniGrafException.Parse`, `.Query`, `.Storage`, `.Api` (a bad argument, a
 closed object, a bind mistake) or `.Other`.
 
+## User functions
+
+A predicate implements `MiniGrafPredicate` (`test(value): Boolean`) and is called in
+Datalog as `[(name ?v)]`. An aggregate implements `MiniGrafAggregate`; its
+`accumulator()` returns a new `MiniGrafAccumulator` (`step(value)`, `finish(count)`) for
+each group or window partition, and it is used as `(name ?v)` in `:find` or with
+`:over`. Values arrive as `MiniGrafValue` (`Int64`, `Text`, …, with the value in
+`value`), and `finish` returns one.
+
+```kotlin
+object Adult : MiniGrafPredicate {
+    override fun test(value: MiniGrafValue): Boolean = when (value) {
+        is MiniGrafValue.Int64 -> value.value >= 18
+        else -> throw MiniGrafUdfException.Failed("age is not an integer")
+    }
+}
+
+object Product : MiniGrafAggregate {
+    override fun accumulator(): MiniGrafAccumulator = object : MiniGrafAccumulator {
+        private var product = 1L
+        override fun step(value: MiniGrafValue) { product *= (value as MiniGrafValue.Int64).value }
+        override fun finish(count: Long): MiniGrafValue = MiniGrafValue.Int64(product) // count: values stepped
+    }
+}
+
+db.registerPredicate("adult?", Adult)
+db.registerAggregate("product", Product)
+db.execute("(query [:find ?n :where [?e :age ?a] [(adult? ?a)] [?e :name ?n]])")
+db.execute("(query [:find ?g (product ?a) :where [?e :group ?g] [?e :age ?a]])")
+db.execute("(query [:find ?a (product ?a :over (:order-by ?a)) :where [?e :age ?a]])")
+```
+
+From Java the interfaces take lambdas where they have one method:
+
+```java
+db.registerPredicate("adult?", v -> {
+    if (!(v instanceof MiniGrafValue.Int64)) throw new IllegalArgumentException("age is not an integer");
+    return ((MiniGrafValue.Int64) v).getValue() >= 18;
+});
+db.registerAggregate("product", () -> new MiniGrafAccumulator() {
+    private long product = 1;
+    @Override public void step(MiniGrafValue v) { product *= ((MiniGrafValue.Int64) v).getValue(); }
+    @Override public MiniGrafValue finish(long count) { return new MiniGrafValue.Int64(product); }
+});
+```
+
+- **Failures.** A callback that throws stops the query with `QRY-014`, raised as
+  `MiniGrafException.Query`; it is never read as `false` or a missing value. From Kotlin,
+  throw `MiniGrafUdfException.Failed(msg)` to choose the message. Any other exception
+  also fails the call, with its first line (class and message, such as
+  `java.lang.IllegalArgumentException: bad`) as the message. The interfaces declare no
+  checked exception, so Java throws an unchecked one. A `MiniGrafValue.Ref` from `finish`
+  whose value is not a UUID is a failing call too.
+- **Lifecycle.** The database keeps a strong reference to each registered object until
+  the database and every cursor and prepared query opened from it are closed. There is
+  no unregister. Predicates and aggregates have separate names, and registering a name
+  twice in either is an error (`INT-040`). A query sees the functions registered when it
+  opened; a prepared query, those registered when each run starts.
+- **Threading.** A callback runs on the thread that calls `execute` (of the database or
+  a prepared query) or a cursor's `nextBatch`, including a prepared query's cursor.
+  Queries on different threads may call the same object at once, so it must be
+  thread-safe. One accumulator is used by one query on one thread at a time.
+- **Cost.** Each call crosses the FFI once: one JVM call per value.
+- **Re-entry.** A callback may query the same database, open a cursor or register another
+  function. Inside a write transaction, the calls the core refuses there still return
+  `INT-001`.
+
 ## Links
 
 - [Full Java/JVM integration guide](https://github.com/project-minigraf/minigraf/wiki/Use-Cases#java--jvm)

diff --git a/java/src/test/java/io/github/project_minigraf/minigraf/JavaInteropTest.java b/java/src/test/java/io/github/project_minigraf/minigraf/JavaInteropTest.java
@@ -103,4 +103,38 @@ void preparedQueryFromJava() throws Exception {
         assertTrue(Minigraf.errorMessage(e).startsWith("[API-022]"));
         assertThrows(IllegalArgumentException.class, () -> Minigraf.execute(pq, Map.of("n", new Object())));
     }
+
+    @Test
+    void userFunctionsFromJava() throws Exception {
+        MiniGrafDb db = Minigraf.openInMemory();
+        db.execute("(transact [[:a :n 1] [:b :n 2] [:c :n 3]])");
+        db.registerPredicate("odd?", v -> ((MiniGrafValue.Int64) v).getValue() % 2 == 1);
+        assertTrue(db.execute("(query [:find ?n :where [?e :n ?n] [(odd? ?n)] [(> ?n 1)]])")
+                .contains("\"results\":[[3]]"));
+
+        db.registerAggregate("product", () -> new MiniGrafAccumulator() {
+            private long product = 1;
+
+            @Override
+            public void step(MiniGrafValue v) {
+                product *= ((MiniGrafValue.Int64) v).getValue();
+            }
+
+            @Override
+            public MiniGrafValue finish(long count) {
+                return new MiniGrafValue.Int64(product);
+            }
+        });
+        assertTrue(db.execute("(query [:find (product ?n) :where [?e :n ?n]])").contains("\"results\":[[6]]"));
+
+        // The interfaces declare no checked exception, so Java throws an unchecked one.
+        db.registerPredicate("broken?", v -> {
+            throw new IllegalArgumentException("from Java");
+        });
+        MiniGrafException e = assertThrows(MiniGrafException.class,
+                () -> db.execute("(query [:find ?n :where [?e :n ?n] [(broken? ?n)]])"));
+        assertTrue(e instanceof MiniGrafException.Query);
+        assertEquals("[QRY-014] user function `broken?` failed: java.lang.IllegalArgumentException: from Java",
+                Minigraf.errorMessage(e));
+    }
 }
diff --git a/java/src/test/kotlin/io/github/project_minigraf/minigraf/UdfTest.kt b/java/src/test/kotlin/io/github/project_minigraf/minigraf/UdfTest.kt
@@ -0,0 +1,211 @@
+package io.github.project_minigraf.minigraf
+
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.fasterxml.jackson.module.kotlin.readValue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CyclicBarrier
+import java.util.concurrent.TimeUnit
+import kotlin.concurrent.thread
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+private val json = jacksonObjectMapper()
+
+private fun rows(s: String): List<List<Any?>> = json.readValue<Map<String, Any>>(s)["results"]
+    .let { @Suppress("UNCHECKED_CAST") (it as List<List<Any?>>) }
+    .sortedBy { it.toString() }
+
+private object IsEven : MiniGrafPredicate {
+    override fun test(value: MiniGrafValue): Boolean = when (value) {
+        is MiniGrafValue.Int64 -> value.value % 2 == 0L
+        else -> throw MiniGrafUdfException.Failed("not an integer")
+    }
+}
+
+/** Sums integers; `step` throws on [failOn]. */
+private class Total(private val failOn: Long? = null) : MiniGrafAggregate {
+    override fun accumulator(): MiniGrafAccumulator = object : MiniGrafAccumulator {
+        private var sum = 0L
+
+        override fun step(value: MiniGrafValue) {
+            val n = (value as MiniGrafValue.Int64).value
+            if (n == failOn) throw MiniGrafUdfException.Failed("refused $n")
+            sum += n
+        }
+
+        override fun finish(count: Long): MiniGrafValue = MiniGrafValue.Int64(sum)
+    }
+}
+
+/** User functions (minigraf#180): predicates and aggregates written in Kotlin. */
+class UdfTest {
+
+    private fun db(): MiniGrafDb {
+        val db = MiniGrafDb.openInMemory()
+        db.execute(
+            "(transact [[:a :n 1] [:b :n 2] [:c :n 3] [:d :n 4]" +
+                " [:a :g 1] [:b :g 1] [:c :g 2] [:d :g 2]])",
+        )
+        return db
+    }
+
+    private fun queryError(block: () -> Unit): String =
+        (assertThrows<MiniGrafException.Query> { block() }).text
+
+    @Test
+    fun predicateFiltersRows() {
+        val db = db()
+        db.registerPredicate("even?", IsEven)
+        assertEquals(listOf(listOf<Any?>(2), listOf<Any?>(4)), rows(db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])")))
+    }
+
+    @Test
+    fun aggregateInFindAndOver() {
+        val db = db()
+        db.registerAggregate("total", Total())
+        assertEquals(listOf(listOf<Any?>(10)), rows(db.execute("(query [:find (total ?n) :where [?e :n ?n]])")))
+        val over = db.execute(
+            "(query [:find ?n (total ?n :over (:partition-by ?g :order-by ?n))" +
+                " :where [?e :n ?n] [?e :g ?g]])",
+        )
+        // A running total per group: 1, 1+2 in group 1; 3, 3+4 in group 2.
+        assertEquals(listOf(listOf<Any?>(1, 1), listOf<Any?>(2, 3), listOf<Any?>(3, 3), listOf<Any?>(4, 7)), rows(over))
+    }
+
+    @Test
+    fun udfExceptionFailsTheQueryWithQry014() {
+        val db = db()
+        db.registerPredicate("even?", IsEven)
+        db.execute("(transact [[:e :n \"x\"]])")
+        val msg = queryError { db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])") }
+        assertTrue(msg.startsWith("[QRY-014]"), "starts with QRY-014")
+        assertTrue(msg.contains("not an integer"), "carries the callback's message")
+    }
+
+    @Test
+    fun anyExceptionFailsTheQueryWithQry014() {
+        val db = db()
+        db.registerPredicate(
+            "broken?",
+            object : MiniGrafPredicate {
+                override fun test(value: MiniGrafValue): Boolean = throw IllegalStateException("own exception")
+            },
+        )
+        val msg = queryError { db.execute("(query [:find ?n :where [?e :n ?n] [(broken? ?n)]])") }
+        assertEquals("[QRY-014] user function `broken?` failed: java.lang.IllegalStateException: own exception", msg)
+    }
+
+    @Test
+    fun aggregateStepFailureIsQry014() {
+        val db = db()
+        db.registerAggregate("total", Total(failOn = 2))
+        val msg = queryError { db.execute("(query [:find (total ?n) :where [?e :n ?n]])") }
+        assertTrue(msg.startsWith("[QRY-014]"), "starts with QRY-014")
+        assertTrue(msg.contains("refused 2"), "carries the step's message")
+    }
+
+    @Test
+    fun badRefFromFinishIsQry014() {
+        val db = db()
+        db.registerAggregate(
+            "bad",
+            object : MiniGrafAggregate {
+                override fun accumulator(): MiniGrafAccumulator = object : MiniGrafAccumulator {
+                    override fun step(value: MiniGrafValue) {}
+                    override fun finish(count: Long): MiniGrafValue = MiniGrafValue.Ref("nope")
+                }
+            },
+        )
+        val msg = queryError { db.execute("(query [:find (bad ?n) :where [?e :n ?n]])") }
+        assertTrue(msg.startsWith("[QRY-014]"), "starts with QRY-014")
+    }
+
+    @Test
+    fun predicateMayQueryTheSameDatabase() {
+        val db = db()
+        db.registerPredicate(
+            "g2?",
+            object : MiniGrafPredicate {
+                override fun test(value: MiniGrafValue): Boolean {
+                    val n = (value as MiniGrafValue.Int64).value
+                    return rows(db.execute("(query [:find ?e :where [?e :n $n] [?e :g 2]])")).isNotEmpty()
+                }
+            },
+        )
+        assertEquals(listOf(listOf<Any?>(3), listOf<Any?>(4)), rows(db.execute("(query [:find ?n :where [?e :n ?n] [(g2? ?n)]])")))
+    }
+
+    @Test
+    fun cursorsAndPreparedQueriesCallUserFunctions() {
+        val db = db()
+        db.registerPredicate("even?", IsEven)
+        db.registerAggregate("total", Total())
+        db.query("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])").use { cursor ->
+            val found = cursor.batches(1).flatMap { json.readValue<List<List<Any?>>>(it) }.sortedBy { it.toString() }
+            assertEquals(listOf(listOf<Any?>(2), listOf<Any?>(4)), found)
+        }
+        val pq = db.prepare("(query [:find (total ?n) :where [?e :g \$g] [?e :n ?n]])")
+        assertEquals(listOf(listOf<Any?>(3)), rows(pq.execute("g" to 1)))
+        pq.query("g" to 2).use { cursor -> assertEquals(listOf("[[7]]"), cursor.batches(10).toList()) }
+    }
+
+    @Test
+    fun aPreparedQuerySeesFunctionsRegisteredBeforeEachRun() {
+        val db = db()
+        val pq = db.prepare("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])")
+        assertThrows<MiniGrafException> { pq.execute() }
+        db.registerPredicate("even?", IsEven)
+        assertEquals(listOf(listOf<Any?>(2), listOf<Any?>(4)), rows(pq.execute()))
+    }
+
+    @Test
+    fun userFunctionFailureInACursorIsQry014() {
+        val db = db()
+        db.registerAggregate("total", Total(failOn = 2))
+        val msg = queryError {
+            db.query("(query [:find (total ?n) :where [?e :n ?n]])").use { it.batches(10).toList() }
+        }
+        assertTrue(msg.startsWith("[QRY-014]"), "starts with QRY-014")
+    }
+
+    @Test
+    fun aNameRegistersOnce() {
+        val db = db()
+        db.registerPredicate("even?", IsEven)
+        val e = assertThrows<MiniGrafException> { db.registerPredicate("even?", IsEven) }
+        assertTrue(e.text.startsWith("[INT-040]"), "duplicate name is INT-040")
+    }
+
+    @Test
+    fun twoQueriesAreInOnePredicateAtOnce() {
+        // Each call waits until a call from the other thread arrives: this only
+        // finishes if neither query holds a binding lock while the core runs.
+        val db = db()
+        val barrier = CyclicBarrier(2)
+        db.registerPredicate(
+            "meet?",
+            object : MiniGrafPredicate {
+                override fun test(value: MiniGrafValue): Boolean {
+                    barrier.await(10, TimeUnit.SECONDS)
+                    return true
+                }
+            },
+        )
+        val results = ConcurrentLinkedQueue<List<List<Any?>>>()
+        val errors = ConcurrentLinkedQueue<Throwable>()
+        val threads = (1..2).map {
+            thread {
+                try {
+                    results.add(rows(db.execute("(query [:find ?n :where [:a :n ?n] [(meet? ?n)]])")))
+                } catch (e: Throwable) {
+                    errors.add(e)
+                }
+            }
+        }
+        threads.forEach { it.join() }
+        assertTrue(errors.isEmpty(), "no thread failed")
+        assertEquals(listOf(listOf(listOf<Any?>(1)), listOf(listOf<Any?>(1))), results.toList())
+    }
+}