janus/datalog

An incremental Datalog engine on top of Rama. You register queries against a stream of [e a v] datoms; the engine emits, per microbatch, the change (a Z-set delta) of each query’s result — and can maintain those changes directly into top-k windows. It never recomputes a query from scratch: anything that would otherwise materialise a view and diff snapshots gets the diff directly. The same engine also answers one-shot pull queries — run a query once against the current index snapshot and get the full result back — for request/response reads.

Everything is accessible from a single namespace — janus.datalog.api.

Setup

Add the dependency:

;; deps.edn
{janus/datalog {:local/root "../datalog"}}

Require the API namespace:

(require '[janus.datalog.api :as api])

The model

The engine consumes two event streams:

Datoms: [[e a v] w] tuples — a fact about entity e plus a Z-set weight: +1 asserts the fact, -1 retracts it. The datoms depot must be partitioned on e (e.g. (declare-depot setup *datoms (hash-by ffirst))); every engine pstate inherits that partitioning.
Query (de)registrations: [external-id query weight] tuples. external-id is your identity for the subscription (any value); a positive weight registers, a negative one retracts. Internally the engine content-addresses query bodies: two external ids registering the same (canonicalised) query share one plan, one subscription set, and one stored row — matches fan back out per external id.

Per microbatch, each registered query emits the rows its result gained (w > 0) or lost (w < 0) — nothing when nothing changed.

The query language

A datalog query is always an extended {:find … :where …} map.

'{:find  [?n ?e]
  :where [[?e :n ?n]]}

Feed specs may use a simpler vector-of-filter-specs shorthand; the janus/feeds component expands that to a full query via janus.feeds.query/expand before registration.

Logic variables are symbols starting with ?. The :where vector is a conjunction of clauses:

Clause Shape Example

Clause	Shape	Example
Pattern	`[e :attr v]` — each of `e` / `v` an lvar or a constant; `v` may be a set of constants (membership)	`[?e :status #{:published :draft}]`
Predicate	`[(op args…)]` — filters rows; `<`, `<=`, `>`, `>=` over a pattern’s value var compile into index range scans	`[(< 2020 ?y)]`
Function expression	`[(op args…) ?out]` — binds `?out` to the result; `op` resolves in `clojure.core` unless namespace-qualified	`[(- ?t 6) ?t0]`

Pattern

[e :attr v] — each of e / v an lvar or a constant; v may be a set of constants (membership)

[?e :status #{:published :draft}]

Predicate

[(op args…)] — filters rows; <, <=, >, >= over a pattern’s value var compile into index range scans

[(< 2020 ?y)]

Function expression

[(op args…) ?out] — binds ?out to the result; op resolves in clojure.core unless namespace-qualified

[(- ?t 6) ?t0]

Constant symbols (e.g. system) are resolved at registration time through the engine opts' :constants map — see Engine opts. A canonical use is the "recently updated" query, where a range bound is computed from a constantly-rewritten clock entity:

'{:find  [?u ?e]
  :where [[?e :status :published]
          [?e :updated ?u]
          [system :current/time ?t]
          [(- ?t 6) ?t0]
          [(< ?t0 ?u)]]}   ;; folds into a dynamic range scan on :updated

Restrictions

Non-recursive — no rules, no view atoms.
No cartesian products: the body’s variables must form one connected graph, and every :find var must appear in the body. Refused at validation time.
Scalar values only in the v slot (ints, strings, keywords, EntityIds).
No anti-joins / negation.

Validation and identity

(api/valid-query query opts)
;; => the canonical query, or an anomaly (janus.anomalies.api)

(api/query-id (api/valid-query query opts))
;; => "0Yc1..." -- content-addressed (SHA-256) id of the canonical form

valid-query specs the query, canonicalises it (>/>= normalisation, range folding), and checks connectedness, range-bound sanity, and that every constant symbol is bound in opts. The query must already be a full extended map (:find and :where required). Always validate before registering: ingest-query drops invalid queries (with a tap>), it does not throw.

query-id is what the engine uses internally to deduplicate query content. Canonicalisation absorbs surface differences (> vs reversed <, relational preds vs :range atoms), but clause order and lvar naming still produce distinct ids.

Engine opts

Every entry point takes the same opts map. It is module-scoped: define it once and pass the same value everywhere (the plan cache keys on it).

(def engine-opts
  {:constants {'system [(ent/->id :janus/system) :neutral]}
   :prefix    'dl
   :pkey-fn   identity})

Key Description

Key	Description
`:constants`	Map of constant symbol → `[value partitioning]`. `value` is the runtime value the symbol resolves to in queries. `partitioning` declares where reads of that entity route: `:neutral` (replicated on every partition — no transition), `:fixed` (one partition, keyed by the value), or `:normal` (broadcast; the default).
`:prefix`	Symbol prefixed to every engine pstate name (see PStates). Must match between `declare-pstates*`, `ingest-query` and the run loops.
`:pkey-fn`	`(fn [external-query-id] partitioning-key)` — routes each top-k bucket to its home partition. Required by `<<run-incremental` and `<<pre-fill`.

:constants

Map of constant symbol → [value partitioning]. value is the runtime value the symbol resolves to in queries. partitioning declares where reads of that entity route: :neutral (replicated on every partition — no transition), :fixed (one partition, keyed by the value), or :normal (broadcast; the default).

:prefix

Symbol prefixed to every engine pstate name (see PStates). Must match between declare-pstates*, ingest-query and the run loops.

:pkey-fn

(fn [external-query-id] partitioning-key) — routes each top-k bucket to its home partition. Required by <<run-incremental and <<pre-fill.

Wiring a module

The high-level loop is <<run-incremental: the engine cascaded into top-k window maintenance. Each registered query feeds the bucket initialized under the same external id; per microbatch, each bucket whose window changed emits its whole new window once.

(ns my.module
  (:require
   [com.rpl.rama :refer :all]
   [janus.datalog.api :as api]
   [janus.datalog.incremental-wco2 :as incremental]))

(def engine-opts
  {:constants {'system [(ent/->id :janus/system) :neutral]}
   :prefix    'dl
   :pkey-fn   identity})

;; query events: [external-id query weight bucket-def]
;; datom events: [[e a v] w], partitioned on e
(defmodule MyModule
  [setup topologies]
  (declare-depot setup *queries :random)
  (declare-depot setup *datoms (hash-by ffirst))
  (let [mb (microbatch-topology topologies "datalog")]
    (api/declare-pstates* mb engine-opts)
    (<<sources mb
      ;; -- query lifecycle ------------------------------------------
      (source> *queries {:start-from :beginning} :> %queries)
      (%queries :> [*external-id *query *weight *bucket-def])
      (|all)
      (api/ingest-query *external-id *query *weight engine-opts)
      ;; `(|all)` above ensures ingest-query (and the bucket lifecycle
      ;; below) run on every task.  ingest-query itself no longer
      ;; performs the partition.
      (<<if (pos? *weight)
        (assoc *bucket-def :id *external-id :> *bdef)
        (api/init-bucket *bdef engine-opts)
       (else>)
        (api/remove-bucket *external-id engine-opts))

      ;; -- the engine -----------------------------------------------
      (source> *datoms {:start-from :beginning} :> %datoms)
      (api/<<run-incremental %datoms engine-opts [*ext-id *topk]
        ;; *ext-id  -- the external query id whose window changed
        ;; *topk    -- the new window, best-first: [[sort-key id] ...]
        (tap> [*ext-id *topk])))))

Notes:

Carrying the bucket-def on the query event (as above) is a caller convention, not an engine requirement — it just makes the bucket lifecycle ride the query lifecycle. A nil bucket-def registers query content whose matches are silently dropped until a bucket appears.
The body of <<run-incremental runs in the post-aggregation phase of a batch block — no partitioners allowed inside it.
Retraction events may carry a nil query body — the engine resolves the external id through its stored mapping. To change the query behind an external id, retract first, then register; a registration for an id mapped to different content is rejected.

Row contract for top-k

The bucket element is the emitted row read as [sort-key id]: the query’s :find clause must list the sort key first and the element id second:

'{:find [?n ?e] :where [[?e :n ?n]]}   ;; top entities by ?n

Bucket definitions

A bucket-def is a map (the :id is assoc’d in at init time, as above):

Key Description

Key	Description
`:janus.datalog.topk/type`	`:janus.datalog.topk/general` or `:janus.datalog.topk/monotonic` (see below).
`:k`	Window size.
`:desc`	Sort direction — `true` for descending (largest sort keys first).
`:reserve`	Monotonic only, optional: retraction headroom kept past the window (default `(min k 64)`).

:janus.datalog.topk/type

:janus.datalog.topk/general or :janus.datalog.topk/monotonic (see below).

:k

Window size.

:desc

Sort direction — true for descending (largest sort keys first).

:reserve

Monotonic only, optional: retraction headroom kept past the window (default (min k 64)).

general buckets keep a full z-set per bucket and accept arbitrary sort keys. monotonic buckets keep only the window itself — a large state saving — but require that every element’s first insertion carries a sort key strictly greater than all sort keys previously inserted into that bucket (e.g. a timestamp). Re-derivations and retractions may then arrive for any element; retractions deeper than the window’s :reserve may not restore evicted elements.

Raw delta rows: `<<run-wco`

When you want the per-row Z-set deltas instead of maintained windows, use <<run-wco:

(source> *datoms {:start-from :beginning} :> %datoms)
(api/<<run-wco %datoms 'dl engine-opts [*query-id *row *w]
  ;; *query-id -- external query id
  ;; *row      -- vector of the :find vars' values
  ;; *w        -- +n rows gained / -n rows lost
  (tap> [*query-id *row *w]))

Emissions are deltas: a row enters the result with positive *w and leaves it with negative *w; multiplicities accumulate. For custom aggregation cascades over these rows, the building blocks ingest-datoms, run-wco, +agg-stage and +agg-stage-windows are also exported — see their docstrings.

Pull queries: one-shot snapshot reads

Everything above is the push engine — it maintains query results incrementally as datoms arrive. The complement is a pull: run a query body once against the current contents of the indices and return the full result set, with no subscription and no delta. Same query language, same planner, same WCO walk — it just reads a fixed snapshot instead of reacting to a stream.

Use a pull when you want the answer now (a request/response read) rather than a maintained, continuously-updated window.

(require '[com.rpl.rama :refer :all]
         '[com.rpl.rama.ops :as ops]
         '[com.rpl.rama.aggs :as aggs]
         '[janus.datalog.api :as api]
         '[janus.datalog.incremental-wco2 :as incremental])

;; Pull requests arrive on their own depot as [req-id query-spec].
;; `query-spec` is either a registered external query id or an ad-hoc
;; query map -- ad-hoc bodies are validated inline and never touch the
;; registry. `req-id` is the caller's request token.
(<<sources mb
  (source> *pulls {:start-from :beginning} :> %pulls)
  (<<batch
    (%pulls :> [*req-id *query-spec])
    (ops/current-microbatch-id :> *t-snap)            ;; snapshot cut
    (api/resolve-pull-query *query-spec engine-opts $$dl-queries $$dl-query-ids
      :> *query-id *canonical)
    (api/run-pull *t-snap *req-id *query-id *canonical #{} engine-opts
                  (incremental/pstates 'dl true)
      :> *tag *row *w)
    ;; one walk emission per matching row; the same row may arrive more
    ;; than once (per partition / set-member), so sum per [tag row] and
    ;; keep only net-positive rows -- the materialised Z-set result.
    (|hash *tag)
    (+compound {*tag {*row (aggs/+sum *w)}} :> *by-tag)
    (ops/explode-map *by-tag :> *req-id *summed)
    (ops/explode-map *summed :> *row *w-net)
    (filter> (pos? *w-net))
    ;; ... write/return per (*req-id, *row); already on *tag's partition.
    ))

How it differs from the push path:

Snapshot, not stream: run-pull reads the index as of a committed microbatch t-snap (every cell with t <= t-snap). Pass a t-snap you know is fully processed for a consistent cross-partition snapshot; obtaining a safe watermark is the caller’s responsibility (the engine does not yet expose one).
Materialised set, not delta: it emits the result rows, not changes. The same row can be produced more than once (across partitions or set-value members), so the caller sums the per-row weights and drops net-zero/negative rows — exactly the Z-set semantics the push path uses internally.
A request tag, not a subscription: run-pull carries an opaque *tag (your req-id) through the walk and emits it with every row. The walk hops partitions, scattering rows away from the origin task, so the tag is how you re-associate each emitted row with its request — it is the group key for the +compound above.

resolve-pull-query always emits [query-id canonical] — canonical is nil for an unknown registered id or an invalid ad-hoc body, and run-pull yields no rows for a nil query (it never crashes the topology).

resolve-pull-query and run-pull are the primitives; they are invoked directly in the caller’s batch (a defgenerator wrapper that runs its own batch← does not compose with the enclosing aggregation). A higher-level returnable surface (a query topology) is a thin layer to add on top.

a pull with no pre-bound vars bootstraps by scanning the cheapest atom’s index — for a single free pattern that is a whole-attribute scan. Constrain the query (a constant entity or value) to make it selective.

PStates

api/declare-pstates* declares everything under :prefix (here 'dl):

PState Contents

PState	Contents
dl-eavt`, `dl-avet	The datom indices.
dl-queries`, `dl-query-ids, `$$dl-subscriptions`	Query registry: content-addressed rows, external-id mapping, wake-up subscriptions.
dl-topk-bucket-defs`, `dl-topk-z-set, `$$dl-topk-windows`	Top-k bucket config and state.

dl-eavt`, `dl-avet

The datom indices.

dl-queries`, `dl-query-ids, $$dl-subscriptions

Query registry: content-addressed rows, external-id mapping, wake-up subscriptions.

dl-topk-bucket-defs`, `dl-topk-z-set, $$dl-topk-windows

Top-k bucket config and state.

These are engine internals — read them for debugging and tests (e.g. asserting a query row exists), but do not write to them directly.

API summary

Var Description

Var	Description
`valid-query`	`(valid-query query opts)` — spec check (extended query only) + canonicalisation + connectedness/constants validation. Returns the canonical query or an anomaly.
`query-id`	Content-addressed id (SHA-256, base64) of a canonical query.
`ingest-query`	Dataflow op `(ingest-query external-id query weight opts)`. `opts` must include `:prefix` (same as `declare-pstates`); registry pstate names are resolved internally. Ref-counts external ids per query content. The caller* must `(
all)` the events before the op (or use `query-source-path`) so that registry rows are written to every partition.	`declare-pstates*`
Declares every engine pstate on a topology under `(:prefix opts)`.	`<<run-incremental`
The engine cascaded into top-k maintenance. Binds `[external-id topk]` per whole-window change, at most once per bucket per microbatch.	`<<run-wco`
The engine alone. Binds `[external-id row w]` per emitted Z-set delta row.	`resolve-pull-query` / `run-pull`
One-shot pull queries — resolve a registered id or ad-hoc body, then run it once against the index snapshot at a committed `t-snap`, emitting `[tag row w]` per match for the caller to sum.	`init-bucket` / `remove-bucket`
Top-k bucket lifecycle. Must run on all tasks — place `(	all)` immediately before `ingest-query` (or use the helper) and keep the bucket forms downstream of it.
`ingest-datoms`, `run-wco`, `+agg-stage`, `+agg-stage-windows`	Lower-level building blocks for custom cascades over the engine’s output.

valid-query

(valid-query query opts) — spec check (extended query only) + canonicalisation + connectedness/constants validation. Returns the canonical query or an anomaly.

query-id

Content-addressed id (SHA-256, base64) of a canonical query.

ingest-query

Dataflow op (ingest-query external-id query weight opts). opts must include :prefix (same as declare-pstates*); registry pstate names are resolved internally. Ref-counts external ids per query content. The caller must `(

all)` the events before the op (or use query-source-path) so that registry rows are written to every partition.

declare-pstates*

Declares every engine pstate on a topology under (:prefix opts).

<<run-incremental

The engine cascaded into top-k maintenance. Binds [external-id topk] per whole-window change, at most once per bucket per microbatch.

<<run-wco

The engine alone. Binds [external-id row w] per emitted Z-set delta row.

resolve-pull-query / run-pull

One-shot pull queries — resolve a registered id or ad-hoc body, then run it once against the index snapshot at a committed t-snap, emitting [tag row w] per match for the caller to sum.

init-bucket / remove-bucket

Top-k bucket lifecycle. Must run on all tasks — place `(

all)` immediately before ingest-query (or use the helper) and keep the bucket forms downstream of it.

ingest-datoms, run-wco, +agg-stage, +agg-stage-windows

Lower-level building blocks for custom cascades over the engine’s output.