From efc57a90fd1a66279dd8ab21156c2354537b2c13 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 11:10:07 +0530 Subject: [PATCH 1/2] feat: user predicates and aggregates in Python (#180) Copies minigraf-binding-template#9: MiniGrafUdfError, the three with_foreign traits, MiniGrafDb.register_predicate/register_aggregate over the core's try_register_*, and handle locks that only clone the database, with the template's Rust tests. Pins core v3 at 02243f16. The package exports Predicate, Aggregate, Accumulator and UdfError. pytest covers a predicate, an aggregate in :find and :over, UdfError and any other exception as QRY-014, a bad Ref from finish, a predicate querying the same database, INT-040, and two threads sharing one predicate. The consumer contract gains a predicate and an aggregate, CI runs cargo test, and the README gains "User functions". Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Urda7FD9CYYoxdeGGJa11x --- .github/workflows/ci.yml | 5 + Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 57 ++++++ consumer-test/contract.py | 46 ++++- minigraf/__init__.py | 16 ++ src/lib.rs | 389 ++++++++++++++++++++++++++++++++++++-- tests/test_udf.py | 173 +++++++++++++++++ 8 files changed, 667 insertions(+), 23 deletions(-) create mode 100644 tests/test_udf.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 22d02ab..98a8a7e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,6 +40,11 @@ jobs: with: python-version: '3.11' + # The shim's own Rust tests (UDF adapters, handle locks), copied from + # minigraf-binding-template. + - name: Rust tests + run: cargo test + - name: Install maturin run: pip install "maturin>=1.5" diff --git a/Cargo.lock b/Cargo.lock index 1f4935c..6a18185 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -410,7 +410,7 @@ checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" [[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 4e52891..bc50cd8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,7 @@ path = "src/uniffi_bindgen.rs" [dependencies] # 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/README.md b/README.md index 74d7707..7a65c6c 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,63 @@ elements is written `minigraf.Value.SET([minigraf.Value.KEYWORD(":role/admin")]) inside a set raises `MiniGrafError.Api` with `[API-021]`; a `list` is not a set. Set values come back as JSON arrays from `execute()` and as `Value.SET([...])` in fact records. +## User functions + +A predicate subclasses `minigraf.Predicate` and is called in Datalog as `[(name ?v)]`. +An aggregate subclasses `minigraf.Aggregate`; its `accumulator()` returns a new +`minigraf.Accumulator` (`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 +`minigraf.Value` (`Value.INT64`, `Value.TEXT`, …, with the value in `.value`), and +`finish` returns one. + +```python +from minigraf import Accumulator, Aggregate, Predicate, UdfError, Value + +class Adult(Predicate): + def test(self, value): + if not isinstance(value, Value.INT64): + raise UdfError.Failed("age is not an integer") + return value.value >= 18 + +class Product(Aggregate): + def accumulator(self): + return ProductAcc() + +class ProductAcc(Accumulator): + def __init__(self): + self.product = 1 + def step(self, value): + self.product *= value.value + def finish(self, count): # count: values stepped + return Value.INT64(self.product) + +db.register_predicate("adult?", Adult()) +db.register_aggregate("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]])") +``` + +- **Failures.** A callback that raises stops the query with `QRY-014`, raised as + `MiniGrafError.Query`; it is never read as `False` or a missing value. Raise + `UdfError.Failed(msg)` to choose the message. Any other exception also fails the call, + with its `repr` (such as `ValueError('bad')`) as the message. A `Value.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 freed. 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 `next_batch`, including a prepared query's cursor. + The binding releases the GIL while the core runs and the callback re-acquires it, so + queries on different threads may call the same object at once; 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 Python 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`. + ## Fact log and log writer `db.fact_log(filter)` streams every fact version (assertions and retractions, with diff --git a/consumer-test/contract.py b/consumer-test/contract.py index 7878091..03b1700 100644 --- a/consumer-test/contract.py +++ b/consumer-test/contract.py @@ -1,5 +1,6 @@ """The consumer contract every binding checks (spec §4): in memory, -file-backed across a reopen, and MiniGrafError for invalid Datalog. +file-backed across a reopen, MiniGrafError for invalid Datalog, and a user +predicate and aggregate (#180). Usage: python contract.py """ @@ -8,7 +9,7 @@ import os import sys -from minigraf import MiniGrafDb, MiniGrafError, Value +from minigraf import Accumulator, Aggregate, MiniGrafDb, MiniGrafError, Predicate, UdfError, Value QUERY = "(query [:find ?n :where [?e :name ?n]])" @@ -38,6 +39,47 @@ def names(db): assert json.loads(member.execute(s={1, 2}))["results"] == [[["x", 1]]], "a set binds" print("ok: set values come back and bind") +mem.execute("(transact [[:alice :age 30] [:bob :age 12]])") + + +class Adult(Predicate): + def test(self, value): + return value.value >= 18 + + +class Product(Aggregate): + def accumulator(self): + return ProductAcc() + + +class ProductAcc(Accumulator): + def __init__(self): + self.product = 1 + + def step(self, value): + if value.value == 0: + raise UdfError.Failed("zero") + self.product *= value.value + + def finish(self, count): + return Value.INT64(self.product) + + +mem.register_predicate("adult?", Adult()) +mem.register_aggregate("product", Product()) +adults = json.loads(mem.execute("(query [:find ?n :where [?e :age ?a] [(adult? ?a)] [?e :name ?n]])")) +assert adults["results"] == [["Alice"]], "a user predicate filters rows" +product = json.loads(mem.execute("(query [:find (product ?a) :where [?e :age ?a]])")) +assert product["results"] == [[360]], "a user aggregate folds values" +mem.execute("(transact [[:carol :age 0]])") +try: + mem.execute("(query [:find (product ?a) :where [?e :age ?a]])") +except MiniGrafError.Query as e: + assert e.msg.startswith("[QRY-014]") and "zero" in e.msg, "a raising UDF is QRY-014" +else: + sys.exit("a raising aggregate did not fail the query") +print("ok: user predicate and aggregate, a raise is QRY-014") + try: mem.execute("(query [:find") except MiniGrafError: diff --git a/minigraf/__init__.py b/minigraf/__init__.py index 2f693ae..460bcd7 100644 --- a/minigraf/__init__.py +++ b/minigraf/__init__.py @@ -5,13 +5,17 @@ FactFilter, FactOrder, FactRecord, + MiniGrafAccumulator, + MiniGrafAggregate, MiniGrafCursor, MiniGrafDb, MiniGrafError, MiniGrafFactLog, MiniGrafBindValue, MiniGrafLogWriter, + MiniGrafPredicate, MiniGrafPreparedQuery, + MiniGrafUdfError, MiniGrafValue, OpenOptions, SyncMode, @@ -24,6 +28,10 @@ Value = MiniGrafValue BindValue = MiniGrafBindValue PreparedQuery = MiniGrafPreparedQuery +Predicate = MiniGrafPredicate +Aggregate = MiniGrafAggregate +Accumulator = MiniGrafAccumulator +UdfError = MiniGrafUdfError #: ``valid_to`` of a fact that is valid forever. VALID_TIME_FOREVER = 2**63 - 1 @@ -223,6 +231,8 @@ def _writer_exit(self, exc_type, exc, tb): __all__ = [ "ANY_VALID_TIME", + "Accumulator", + "Aggregate", "BindValue", "Cursor", "Entity", @@ -231,19 +241,25 @@ def _writer_exit(self, exc_type, exc, tb): "FactOrder", "FactRecord", "LogWriter", + "MiniGrafAccumulator", + "MiniGrafAggregate", "MiniGrafCursor", "MiniGrafDb", "MiniGrafError", "MiniGrafFactLog", "MiniGrafBindValue", "MiniGrafLogWriter", + "MiniGrafPredicate", "MiniGrafPreparedQuery", + "MiniGrafUdfError", "MiniGrafValue", "OpenOptions", + "Predicate", "PreparedQuery", "SyncMode", "Timestamp", "TxCount", + "UdfError", "VALID_TIME_FOREVER", "WAL_CHECKPOINT_NEVER", "Value", 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"); + } } diff --git a/tests/test_udf.py b/tests/test_udf.py new file mode 100644 index 0000000..332af0e --- /dev/null +++ b/tests/test_udf.py @@ -0,0 +1,173 @@ +"""User functions (#180): predicates and aggregates written in Python.""" + +import json +import threading + +import pytest + +from minigraf import ( + Accumulator, + Aggregate, + MiniGrafDb, + MiniGrafError, + Predicate, + UdfError, + Value, +) + + +@pytest.fixture +def db(): + db = MiniGrafDb.open_in_memory() + 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 + + +def rows(result): + return sorted(json.loads(result)["results"]) + + +class IsEven(Predicate): + def test(self, value): + if not isinstance(value, Value.INT64): + raise UdfError.Failed("not an integer") + return value.value % 2 == 0 + + +class Total(Aggregate): + """Sums integers; step raises on ``fail_on``.""" + + def __init__(self, fail_on=None): + self.fail_on = fail_on + + def accumulator(self): + return TotalAcc(self.fail_on) + + +class TotalAcc(Accumulator): + def __init__(self, fail_on): + self.fail_on = fail_on + self.sum = 0 + + def step(self, value): + if value.value == self.fail_on: + raise UdfError.Failed(f"refused {value.value}") + self.sum += value.value + + def finish(self, count): + return Value.INT64(self.sum) + + +def test_predicate_filters_rows(db): + db.register_predicate("even?", IsEven()) + assert rows(db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])")) == [[2], [4]] + + +def test_aggregate_in_find_and_over(db): + db.register_aggregate("total", Total()) + assert rows(db.execute("(query [:find (total ?n) :where [?e :n ?n]])")) == [[10]] + 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. + assert rows(over) == [[1, 1], [2, 3], [3, 3], [4, 7]] + + +def test_udf_error_fails_the_query_with_qry_014(db): + db.register_predicate("even?", IsEven()) + db.execute('(transact [[:e :n "x"]])') + with pytest.raises(MiniGrafError.Query) as e: + db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])") + assert e.value.msg.startswith("[QRY-014]") + assert "not an integer" in e.value.msg + + +def test_any_exception_fails_the_query_with_qry_014(db): + class Broken(Predicate): + def test(self, value): + raise ValueError("own exception") + + db.register_predicate("broken?", Broken()) + with pytest.raises(MiniGrafError.Query) as e: + db.execute("(query [:find ?n :where [?e :n ?n] [(broken? ?n)]])") + assert e.value.msg.startswith("[QRY-014]") + assert "own exception" in e.value.msg + + +def test_aggregate_step_error_is_qry_014(db): + db.register_aggregate("total", Total(fail_on=2)) + with pytest.raises(MiniGrafError.Query) as e: + db.execute("(query [:find (total ?n) :where [?e :n ?n]])") + assert e.value.msg.startswith("[QRY-014]") + assert "refused 2" in e.value.msg + + +def test_bad_ref_from_finish_is_qry_014(db): + class BadRef(Aggregate): + def accumulator(self): + return BadRefAcc() + + class BadRefAcc(Accumulator): + def step(self, value): + pass + + def finish(self, count): + return Value.REF("nope") + + db.register_aggregate("bad", BadRef()) + with pytest.raises(MiniGrafError.Query) as e: + db.execute("(query [:find (bad ?n) :where [?e :n ?n]])") + assert e.value.msg.startswith("[QRY-014]") + + +def test_predicate_may_query_the_same_database(db): + class InGroup2(Predicate): + def test(self, value): + found = db.execute( + f"(query [:find ?e :where [?e :n {value.value}] [?e :g 2]])" + ) + return len(json.loads(found)["results"]) > 0 + + db.register_predicate("g2?", InGroup2()) + assert rows(db.execute("(query [:find ?n :where [?e :n ?n] [(g2? ?n)]])")) == [[3], [4]] + + +def test_a_name_registers_once(db): + db.register_predicate("even?", IsEven()) + with pytest.raises(MiniGrafError) as e: + db.register_predicate("even?", IsEven()) + assert e.value.msg.startswith("[INT-040]") + + +def test_two_threads_share_one_predicate(db): + calls = [] + + class Counting(Predicate): + def test(self, value): + calls.append(value.value) # list.append is atomic under the GIL + return value.value % 2 == 0 + + db.register_predicate("even?", Counting()) + results, errors = [], [] + + def run(): + try: + for _ in range(50): + results.append( + rows(db.execute("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])")) + ) + except Exception as e: # surfaced below; a thread's exception is lost otherwise + errors.append(e) + + threads = [threading.Thread(target=run) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] + assert results == [[[2], [4]]] * 100 + assert len(calls) == 400 From b1f45b26162022c042d687108ff4d320f20b6863 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sun, 25 Oct 2026 11:11:21 +0530 Subject: [PATCH 2/2] test: UDFs in cursors and prepared queries, two queries in one predicate at once, exact foreign-exception message (#180) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Urda7FD9CYYoxdeGGJa11x --- tests/test_udf.py | 51 +++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 49 insertions(+), 2 deletions(-) diff --git a/tests/test_udf.py b/tests/test_udf.py index 332af0e..e6adbd6 100644 --- a/tests/test_udf.py +++ b/tests/test_udf.py @@ -94,8 +94,7 @@ def test(self, value): db.register_predicate("broken?", Broken()) with pytest.raises(MiniGrafError.Query) as e: db.execute("(query [:find ?n :where [?e :n ?n] [(broken? ?n)]])") - assert e.value.msg.startswith("[QRY-014]") - assert "own exception" in e.value.msg + assert e.value.msg == "[QRY-014] user function `broken?` failed: ValueError('own exception')" def test_aggregate_step_error_is_qry_014(db): @@ -136,6 +135,25 @@ def test(self, value): assert rows(db.execute("(query [:find ?n :where [?e :n ?n] [(g2? ?n)]])")) == [[3], [4]] +def test_cursors_and_prepared_queries_call_user_functions(db): + db.register_predicate("even?", IsEven()) + db.register_aggregate("total", Total()) + with db.query("(query [:find ?n :where [?e :n ?n] [(even? ?n)]])") as cursor: + assert sorted(cursor) == [[2], [4]] + pq = db.prepare("(query [:find (total ?n) :where [?e :g $g] [?e :n ?n]])") + assert rows(pq.execute(g=1)) == [[3]] + with pq.query(g=2) as cursor: + assert list(cursor) == [[7]] + + +def test_a_user_function_failure_in_a_cursor_is_qry_014(db): + db.register_aggregate("total", Total(fail_on=2)) + with pytest.raises(MiniGrafError.Query) as e: + with db.query("(query [:find (total ?n) :where [?e :n ?n]])") as cursor: + list(cursor) + assert e.value.msg.startswith("[QRY-014]") + + def test_a_name_registers_once(db): db.register_predicate("even?", IsEven()) with pytest.raises(MiniGrafError) as e: @@ -143,6 +161,35 @@ def test_a_name_registers_once(db): assert e.value.msg.startswith("[INT-040]") +def test_two_queries_are_in_one_predicate_at_once(db): + # Each call waits until a call from the other thread arrives: this only + # finishes if neither query holds the GIL or a binding lock while the + # core runs. + barrier = threading.Barrier(2, timeout=10) + + class Meet(Predicate): + def test(self, value): + barrier.wait() + return True + + db.register_predicate("meet?", Meet()) + results, errors = [], [] + + def run(): + try: + results.append(rows(db.execute("(query [:find ?n :where [:a :n ?n] [(meet? ?n)]])"))) + except Exception as e: + errors.append(e) + + threads = [threading.Thread(target=run) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] + assert results == [[[1]], [[1]]] + + def test_two_threads_share_one_predicate(db): calls = []