feat: user predicates and aggregates (#180)

.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"
 

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"

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

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 <empty directory>
 """
@@ -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:

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",