From b0f82bc567329373ef15c4aabc62cebefd0bb4d2 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 11:35:53 +0530 Subject: [PATCH 1/2] feat: user predicates and aggregates (minigraf#180) MiniGrafPredicate, MiniGrafAggregate and MiniGrafAccumulator, implementable from Kotlin and Java (lambdas for the one-method interfaces), and MiniGrafDb.registerPredicate/registerAggregate over the core's try_register_*. A throwing callback stops the query with QRY-014. Every MiniGrafDb method locks its handle only to clone the database, so a callback can call back into it. Shim code and Rust tests copied from minigraf-binding-template#9; core pin 7c8c5540 -> 02243f16. Kotlin and Java tests, consumer contract (predicate, aggregate, a throw is QRY-014), README "User functions". Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Urda7FD9CYYoxdeGGJa11x --- Cargo.lock | 2 +- Cargo.toml | 2 +- consumer-test/src/test/java/ContractTest.java | 35 ++ java/README.md | 67 +++ .../minigraf/JavaInteropTest.java | 34 ++ .../project_minigraf/minigraf/UdfTest.kt | 202 +++++++++ src/lib.rs | 389 +++++++++++++++++- 7 files changed, 710 insertions(+), 21 deletions(-) create mode 100644 java/src/test/kotlin/io/github/project_minigraf/minigraf/UdfTest.kt diff --git a/Cargo.lock b/Cargo.lock index 9054e7d..60217f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -440,7 +440,7 @@ checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" [[package]] name = "minigraf" version = "2.0.5" -source = "git+https://github.com/project-minigraf/minigraf?rev=7c8c554096209ddc2bf1342b9a4eef3ef77c67c3#7c8c554096209ddc2bf1342b9a4eef3ef77c67c3" +source = "git+https://github.com/project-minigraf/minigraf?rev=02243f1656eedd987611a9babdfee4a8c6e22b6b#02243f1656eedd987611a9babdfee4a8c6e22b6b" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index ff07eb0..0736cff 100644 --- 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 index 6fc338d..1206f66 100644 --- 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 index 4dadbba..9ca5c2f 100644 --- 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. +- **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 index 87d1671..258613c 100644 --- 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 new file mode 100644 index 0000000..3e58ae7 --- /dev/null +++ b/java/src/test/kotlin/io/github/project_minigraf/minigraf/UdfTest.kt @@ -0,0 +1,202 @@ +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> = json.readValue>(s)["results"] + .let { @Suppress("UNCHECKED_CAST") (it as List>) } + .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 { block() }).text + + @Test + fun predicateFiltersRows() { + val db = db() + db.registerPredicate("even?", IsEven) + assertEquals(listOf(listOf(2), listOf(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(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(1, 1), listOf(2, 3), listOf(3, 3), listOf(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(3), listOf(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>>(it) }.sortedBy { it.toString() } + assertEquals(listOf(listOf(2), listOf(4)), found) + } + val pq = db.prepare("(query [:find (total ?n) :where [?e :g \$g] [?e :n ?n]])") + assertEquals(listOf(listOf(3)), rows(pq.execute("g" to 1))) + pq.query("g" to 2).use { cursor -> assertEquals(listOf("[[7]]"), cursor.batches(10).toList()) } + } + + @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 { 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>>() + val errors = ConcurrentLinkedQueue() + 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(1)), listOf(listOf(1))), results.toList()) + } +} diff --git a/src/lib.rs b/src/lib.rs index 462c5b9..6593cfd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -456,14 +456,7 @@ impl MiniGrafDb { } pub fn execute(&self, datalog: String) -> Result { - let result = self - .inner - .lock() - .map_err(|_| MiniGrafError::Other { - msg: "mutex poisoned".into(), - })? - .execute(&datalog) - .map_err(MiniGrafError::from)?; + let result = self.db()?.execute(&datalog)?; Ok(query_result_to_json(result)) } @@ -482,7 +475,7 @@ impl MiniGrafDb { /// Open a cursor over a `(query ...)`. Its answer is fixed when it opens. pub fn query(&self, datalog: String) -> Result, MiniGrafError> { - let cursor = lock(&self.inner)?.query(&datalog)?; + let cursor = self.db()?.query(&datalog)?; Ok(Arc::new(MiniGrafCursor { vars: cursor.vars().to_vec(), inner: Mutex::new(Some(cursor)), @@ -493,7 +486,7 @@ impl MiniGrafDb { /// the log is closed or read to the end. pub fn fact_log(&self, filter: FactFilter) -> Result, MiniGrafError> { let filter = filter.to_core()?; - let log = lock(&self.inner)?.fact_log(&filter)?; + let log = self.db()?.fact_log(&filter)?; Ok(Arc::new(MiniGrafFactLog { inner: Mutex::new(Some(log)), })) @@ -503,7 +496,7 @@ impl MiniGrafDb { /// different bind values in its `$name` slots. A transact, retract or /// rule is API-005/006/007; a slot in an attribute position is API-023. pub fn prepare(&self, datalog: String) -> Result, MiniGrafError> { - let inner = lock(&self.inner)?.prepare(&datalog)?; + let inner = self.db()?.prepare(&datalog)?; Ok(Arc::new(MiniGrafPreparedQuery { _db: self.inner.clone(), inner, @@ -512,20 +505,109 @@ impl MiniGrafDb { /// The transaction counter that `:as-of N` compares against. pub fn current_tx_count(&self) -> Result { - Ok(from_u64(lock(&self.inner)?.current_tx_count())) + Ok(from_u64(self.db()?.current_tx_count())) } pub fn checkpoint(&self) -> Result<(), MiniGrafError> { - self.inner - .lock() - .map_err(|_| MiniGrafError::Other { - msg: "mutex poisoned".into(), - })? - .checkpoint() - .map_err(MiniGrafError::from) + Ok(self.db()?.checkpoint()?) + } + + /// Register `predicate` under `name`, called in Datalog as + /// `[(name ?v)]`. A predicate that throws stops the query with QRY-014. + /// Queries already open keep the functions they opened with. + pub fn register_predicate( + &self, + name: String, + predicate: Arc, + ) -> Result<(), MiniGrafError> { + Ok(self.db()?.try_register_predicate(&name, move |v| { + Ok(predicate.test(MiniGrafValue::from(v.clone()))?) + })?) + } + + /// Register `aggregate` under `name`, used in `:find` as `(name ?v)` and + /// as a window function with `:over`. Each group or window partition + /// gets its own accumulator. A throw stops the query with QRY-014. + pub fn register_aggregate( + &self, + name: String, + aggregate: Arc, + ) -> Result<(), MiniGrafError> { + Ok(self.db()?.try_register_aggregate( + &name, + move || Ok(aggregate.accumulator()?), + |acc: &mut Arc, v| { + Ok(acc.step(MiniGrafValue::from(v.clone()))?) + }, + |acc, count| { + // A count of values held in memory stays far below i64::MAX. + let count = i64::try_from(count).unwrap_or(i64::MAX); + acc.finish(count)? + .to_core() + .map_err(|e| minigraf::UdfError::new(e.to_string())) + }, + )?) + } +} + +impl MiniGrafDb { + /// The database, cloned out of the handle so that no binding lock is + /// held while the core runs: a user function may call back into this + /// handle. + fn db(&self) -> Result { + Ok(lock(&self.inner)?.clone()) + } +} + +// ─── User functions ────────────────────────────────────────────────────────── + +/// Why a user function failed. A callback that throws anything else is +/// mapped to this. The query stops with QRY-014, which carries `msg`. +#[derive(Debug, thiserror::Error, uniffi::Error)] +pub enum MiniGrafUdfError { + #[error("{msg}")] + Failed { msg: String }, +} + +/// Any other exception. Kotlin sends its stack trace, so only the first +/// line is kept: the exception's class and message. +impl From for MiniGrafUdfError { + fn from(e: uniffi::UnexpectedUniFFICallbackError) -> Self { + let msg = e.reason.lines().next().unwrap_or_default().to_string(); + MiniGrafUdfError::Failed { msg } } } +impl From for minigraf::UdfError { + fn from(e: MiniGrafUdfError) -> Self { + let MiniGrafUdfError::Failed { msg } = e; + minigraf::UdfError::new(msg) + } +} + +/// A user predicate. It may be called from any thread that queries the +/// database, concurrently, so it must be thread-safe. +#[uniffi::export(with_foreign)] +pub trait MiniGrafPredicate: Send + Sync { + fn test(&self, value: MiniGrafValue) -> Result; +} + +/// A user aggregate: a factory of accumulators. +#[uniffi::export(with_foreign)] +pub trait MiniGrafAggregate: Send + Sync { + /// A new, empty accumulator: one per group, or per window partition. + fn accumulator(&self) -> Result, MiniGrafUdfError>; +} + +/// The running state of one aggregate. One query uses it, on one thread at +/// a time. +#[uniffi::export(with_foreign)] +pub trait MiniGrafAccumulator: Send + Sync { + fn step(&self, value: MiniGrafValue) -> Result<(), MiniGrafUdfError>; + /// `count`: values stepped so far. + fn finish(&self, count: i64) -> Result; +} + // ─── MiniGrafPreparedQuery ─────────────────────────────────────────────────── /// A query parsed once by `MiniGrafDb.prepare`, run with bind values. It @@ -1430,4 +1512,273 @@ mod tests { 1 ); } + + // ─── User functions (#180) ─────────────────────────────────────────── + + struct IsEven; + impl MiniGrafPredicate for IsEven { + fn test(&self, value: MiniGrafValue) -> Result { + match value { + MiniGrafValue::Int64 { value } => Ok(value % 2 == 0), + _ => Err(MiniGrafUdfError::Failed { + msg: "not an integer".into(), + }), + } + } + } + + /// Sums integers; fails in `step` on `fail_on`, and `finish` returns + /// `out` when set. + struct Total { + fail_on: Option, + out: Option, + steps: Arc, + } + struct TotalAcc { + fail_on: Option, + out: Option, + steps: Arc, + sum: Mutex, + } + impl MiniGrafAggregate for Total { + fn accumulator(&self) -> Result, MiniGrafUdfError> { + Ok(Arc::new(TotalAcc { + fail_on: self.fail_on, + out: self.out.clone(), + steps: self.steps.clone(), + sum: Mutex::new(0), + })) + } + } + impl MiniGrafAccumulator for TotalAcc { + fn step(&self, value: MiniGrafValue) -> Result<(), MiniGrafUdfError> { + self.steps.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let MiniGrafValue::Int64 { value } = value else { + return Err(MiniGrafUdfError::Failed { + msg: "not an integer".into(), + }); + }; + if Some(value) == self.fail_on { + return Err(MiniGrafUdfError::Failed { + msg: format!("refused {value}"), + }); + } + *self.sum.lock().unwrap() += value; + Ok(()) + } + fn finish(&self, count: i64) -> Result { + if let Some(out) = &self.out { + return Ok(out.clone()); + } + Ok(MiniGrafValue::Text { + value: format!("{}/{count}", *self.sum.lock().unwrap()), + }) + } + } + + fn total(fail_on: Option, out: Option) -> Arc { + Arc::new(Total { + fail_on, + out, + steps: Arc::default(), + }) + } + + fn udf_db() -> Arc { + let db = MiniGrafDb::open_in_memory().expect("open"); + 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]])" + .into(), + ) + .expect("transact"); + db + } + + fn results(json: &str) -> Vec { + let v: serde_json::Value = serde_json::from_str(json).expect("json"); + v["results"].as_array().expect("results").clone() + } + + #[test] + fn predicate_filters_rows() { + let db = udf_db(); + db.register_predicate("even?".into(), Arc::new(IsEven)) + .expect("register"); + let mut rows = results( + &db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])".into()) + .expect("query"), + ); + rows.sort_by_key(|r| r[0].as_i64()); + assert_eq!(rows, vec![serde_json::json!([2]), serde_json::json!([4])]); + } + + #[test] + fn failing_predicate_is_qry_014() { + let db = udf_db(); + db.execute(r#"(transact [[:e :n "x"]])"#.into()) + .expect("transact"); + db.register_predicate("even?".into(), Arc::new(IsEven)) + .expect("register"); + let e = err(db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])".into())); + let MiniGrafError::Query { msg } = e else { + panic!("expected a query error"); + }; + assert!(msg.contains("QRY-014") && msg.contains("even?") && msg.contains("not an integer")); + } + + #[test] + fn aggregate_in_find_and_over() { + let db = udf_db(); + db.register_aggregate("total".into(), total(None, None)) + .expect("register"); + let rows = results( + &db.execute("(query [:find (total ?n) :where [?e :n ?n]])".into()) + .expect("find"), + ); + assert_eq!(rows, vec![serde_json::json!(["10/4"])]); + let mut rows = results( + &db.execute( + "(query [:find ?n (total ?n :over (:partition-by ?g :order-by ?n)) :where [?e :n ?n] [?e :g ?g]])" + .into(), + ) + .expect("over"), + ); + rows.sort_by_key(|r| r[0].as_i64()); + assert_eq!( + rows, + vec![ + serde_json::json!([1, "1/1"]), + serde_json::json!([2, "3/2"]), + serde_json::json!([3, "3/1"]), + serde_json::json!([4, "7/2"]), + ] + ); + } + + #[test] + fn failing_step_is_qry_014_and_stops() { + let db = udf_db(); + let agg = total(Some(2), None); + db.register_aggregate("total".into(), agg.clone()) + .expect("register"); + let e = err(db.execute("(query [:find (total ?n) :where [?e :n ?n]])".into())); + let MiniGrafError::Query { msg } = e else { + panic!("expected a query error"); + }; + assert!(msg.contains("QRY-014") && msg.contains("refused 2")); + assert!( + agg.steps.load(std::sync::atomic::Ordering::SeqCst) <= 4, + "no step after the whole input" + ); + } + + #[test] + fn bad_ref_from_finish_is_qry_014() { + let db = udf_db(); + let bad = MiniGrafValue::Ref { + value: "nope".into(), + }; + db.register_aggregate("total".into(), total(None, Some(bad))) + .expect("register"); + let e = err(db.execute("(query [:find (total ?n) :where [?e :n ?n]])".into())); + let MiniGrafError::Query { msg } = e else { + panic!("expected a query error"); + }; + assert!(msg.contains("QRY-014") && msg.contains(r#"ref value "nope" is not a UUID"#)); + } + + #[test] + fn finish_ref_reaches_the_result() { + let db = udf_db(); + let id = "6f1c2a3b-4d5e-4f60-8172-839ab4c5d6e7".to_string(); + db.register_aggregate( + "total".into(), + total(None, Some(MiniGrafValue::Ref { value: id.clone() })), + ) + .expect("register"); + let rows = results( + &db.execute("(query [:find (total ?n) :where [?e :n ?n]])".into()) + .expect("find"), + ); + assert_eq!(rows, vec![serde_json::json!([id])]); + } + + /// Queries the database it is registered on from inside the call. + struct Reenter(std::sync::Weak, std::sync::atomic::AtomicBool); + impl MiniGrafPredicate for Reenter { + fn test(&self, value: MiniGrafValue) -> Result { + let db = self.0.upgrade().ok_or(MiniGrafUdfError::Failed { + msg: "db gone".into(), + })?; + let MiniGrafValue::Int64 { value } = value else { + return Ok(false); + }; + let fail = |e: MiniGrafError| MiniGrafUdfError::Failed { msg: e.to_string() }; + let json = db + .execute(format!( + "(query [:find ?e :where [?e :n {value}] [?e :g 2]])" + )) + .map_err(fail)?; + let cursor = db + .query("(query [:find ?n :where [?e :n ?n]])".into()) + .map_err(fail)?; + let _ = cursor.next_batch(10).map_err(fail)?; + if !self.1.swap(true, std::sync::atomic::Ordering::SeqCst) { + db.register_predicate("late?".into(), Arc::new(IsEven)) + .map_err(fail)?; + } + Ok(!results(&json).is_empty()) + } + } + + #[test] + fn predicate_can_query_its_own_database() { + let db = udf_db(); + db.register_predicate( + "g2?".into(), + Arc::new(Reenter(Arc::downgrade(&db), Default::default())), + ) + .expect("register"); + let (tx, rx) = std::sync::mpsc::channel(); + let db2 = db.clone(); + std::thread::spawn(move || { + let r = db2.execute("(query [:find ?n :where [?e :n ?n] [(g2? ?n)]])".into()); + let _ = tx.send(r.map(|json| results(&json))); + }); + let mut rows = rx + .recv_timeout(std::time::Duration::from_secs(30)) + .expect("no deadlock") + .expect("query"); + rows.sort_by_key(|r| r[0].as_i64()); + assert_eq!(rows, vec![serde_json::json!([3]), serde_json::json!([4])]); + let later = db + .execute("(query [:find ?n :where [?e :n ?n] [(late? ?n)]])".into()) + .expect("registered from inside the call"); + assert_eq!(results(&later).len(), 2); + } + + #[test] + fn registering_a_name_twice_is_int_040() { + let db = udf_db(); + db.register_predicate("even?".into(), Arc::new(IsEven)) + .expect("register"); + let e = err(db.register_predicate("even?".into(), Arc::new(IsEven))); + assert!(msg(e).starts_with("[INT-040]")); + db.register_aggregate("total".into(), total(None, None)) + .expect("aggregates have their own names"); + let e = err(db.register_aggregate("total".into(), total(None, None))); + assert!(msg(e).starts_with("[INT-040]")); + } + + #[test] + fn unexpected_callback_error_becomes_failed() { + let e = MiniGrafUdfError::from(uniffi::UnexpectedUniFFICallbackError { + reason: "boom".into(), + }); + assert_eq!(e.to_string(), "boom"); + let e = MiniGrafUdfError::from(uniffi::UnexpectedUniFFICallbackError { + reason: "java.lang.IllegalStateException: no\n\tat Foo.test(Foo.kt:3)\n".into(), + }); + assert_eq!(e.to_string(), "java.lang.IllegalStateException: no"); + } } From 4c7eb3b2481d1855a370f993ddc0f4790eab8e02 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 11:39:17 +0530 Subject: [PATCH 2/2] test: a prepared query sees functions registered after prepare (Codex review) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Urda7FD9CYYoxdeGGJa11x --- java/README.md | 2 +- .../io/github/project_minigraf/minigraf/UdfTest.kt | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/java/README.md b/java/README.md index 9ca5c2f..d1cb2cd 100644 --- a/java/README.md +++ b/java/README.md @@ -219,7 +219,7 @@ db.registerAggregate("product", () -> new MiniGrafAccumulator() { 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. + 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 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 index 3e58ae7..dfa9359 100644 --- a/java/src/test/kotlin/io/github/project_minigraf/minigraf/UdfTest.kt +++ b/java/src/test/kotlin/io/github/project_minigraf/minigraf/UdfTest.kt @@ -151,6 +151,15 @@ class UdfTest { 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 { pq.execute() } + db.registerPredicate("even?", IsEven) + assertEquals(listOf(listOf(2), listOf(4)), rows(pq.execute())) + } + @Test fun userFunctionFailureInACursorIsQry014() { val db = db()