datahike.pg.server — org.replikativ/pg-datahike 0.1.40

PostgreSQL wire protocol server for Datahike.

Starts a pgwire-compatible server that accepts SQL queries from standard PostgreSQL clients (psql, DBeaver, JDBC, Python/psycopg2, etc.) and translates them to Datahike Datalog queries.

Usage: (require '[datahike.api :as d]) (require '[datahike.pg.server :as pg])

(def conn (d/connect cfg)) (def server (pg/start-server conn {:port 5432})) ;; ... use any PostgreSQL client ... (pg/stop-server server)

*schema-cache-enabled?* clj

Bind false to bypass the cache. For perf comparisons only; production code should leave this on.

add-database! clj

(add-database! server-result name conn)

Add a Datahike conn to a running pg-datahike server's registry under name. Subsequent client connections with database=name will route to it. Returns the new registry contents.

Symmetric with the SQL CREATE DATABASE path: add-database! is the Clojure-side knob, :on-create-database is the SQL-side knob, and they share the same atom so either source is visible to both.

cache-stats clj

Hit/miss counter for the schema cache. Atom whose identity is stable across reloads so external observers can hold a reference and watch counts. Mutated by every schema-cached call; read for inspection / verification harnesses.

databases clj

(databases server-result)

Return the current set of database names registered with the server.

invalidate-schema-cache! clj

(invalidate-schema-cache!)

Clear the per-schema cache. Called from every DDL exec branch.

make-query-handler clj

(make-query-handler conn
                    &
                    [{:keys [on-query db-name registered-databases
                             initial-branch dispatch-stats on-create-database
                             on-delete-database registry-atom]
                      :as opts}])

Create a PgWireServer.QueryHandler that dispatches SQL to Datahike.

conn: a Datahike connection opts: optional map with :on-query (fn [sql]) invoked on every SQL string :compat :strict (default) | :permissive — named bundle of features to silently accept, see compat-presets. :silently-accept a set of reject-kinds to swallow on top of the preset. Valid kinds: :grant :revoke :policy :rls :create-extension. These return a synthetic success tag (e.g. "GRANT") instead of SQLSTATE 0A000. :db-name string — the database name this handler represents. Returned by current_database(); defaults to "datahike" when omitted. :registered-databases seq of strings — all database names registered at the server. Surfaced in the virtual pg_database catalog so \l and DatabaseMetaData enumerate the server's tenancy. Typically supplied by start-server with the keys of its registry; omit for single-DB / bare-handler use. :dispatch-stats optional atom; when supplied, the handler bumps :fast-path-count or :full-parse-count on each parse-sql invocation depending on whether catalog/system-query? matched. Used by tests and observability tooling to detect when an upgrade silently demotes a probe to the slow path.

Supports temporal session variables: SET datahike.as_of = '2024-01-15T00:00:00Z' SET datahike.since = '2024-01-01T00:00:00Z' SET datahike.history = 'true' RESET datahike.as_of

make-query-handler-factory clj

(make-query-handler-factory registry-or-atom & [opts])

Build a QueryHandlerFactory that routes on the StartupMessage's database parameter.

registry: {name → conn} map of database name to Datahike conn. opts: forwarded to make-query-handler (e.g. :on-query, :compat).

Clients that pass database=X land on (registry "X"). An optional :branch suffix — database=X:feature — pins the connection's session-state to the :feature branch from the first query. Both the current_database() reply and the pg_database catalog still report the base name; the branch is a session-state detail.

Unknown base-names get a handler that errors every query with 3D000 invalid_catalog_name — matching PG's behaviour on a non-existent db.

Intended for callers who want to wrap their own PgWireServer (e.g. custom host/port binding); start-server uses it internally.

nextval! clj

(nextval! conn seq-name)

Atomically advance the named sequence on conn and return the new long. Shared core of SELECT nextval(...) and INSERT-VALUES nextval resolution.

PG semantics: a nextval advance is never rolled back, even by ROLLBACK on the surrounding transaction. We match that by always committing to the live conn, regardless of any session's :in-tx? state.

Atomicity: optimistic CAS-retry. Each iteration reads the current :seq/value, computes the new value, and submits

[[:db/cas seq-eid :__seq__/value curr new]]

to d/transact. Datahike's transactor serialises tx applications per conn, so two concurrent CAS submissions see each other: the first wins, the second's old-val no longer matches the transactor's current value and the CAS raises :transact/cas. We re-read and retry. The loser thread will see the winner's new value, compute the next slot, and commit cleanly.

Why CAS instead of the simpler :db.fn/call pattern: Datahike's writer batches multiple in-flight transactions for a single commit, then writes the SAME :db-after (the batch's final db) into every tx-report in the batch. So reading (:db-after report) to recover the value this call assigned doesn't work — it returns the LAST tx in the batch. CAS sidesteps that: the value we intended is the literal new slot of the op-vec, available without reading :db-after.

Throws ex-info :undefined-sequence if the named sequence does not exist, and :serialization-failure if the contention retry budget is exhausted.

remove-database! clj

(remove-database! server-result name)

Remove a database from a running pg-datahike server's registry. Does NOT release the conn or delete the backing store — that's the operator's call. Returns the new registry contents.

reset-advisory-locks! clj

(reset-advisory-locks!)

Clear the advisory-lock registry. Test-fixture helper; not for handler code. In production, locks release on unlock, COMMIT/ROLLBACK (xact- level), DISCARD ALL, or connection close.

reset-lock-registry! clj

(reset-lock-registry!)

Clear the server-wide row-lock registry. Intended for test fixtures — in production, locks are released on COMMIT/ROLLBACK/DISCARD ALL and when the handler's TCP connection closes. Do not call this from handler code.

start-server clj

(start-server conn-or-registry
              &
              [{:keys [port host on-query default on-create-database
                       on-delete-database database-template]
                :or {port 5432 host "127.0.0.1" default "datahike"}
                :as opts}])

Start a PostgreSQL wire protocol server for one or more Datahike connections.

conn-or-registry is either:

• a single Datahike conn (convenience — treated as {"datahike" conn}), or
• a map {name → conn} of database name to conn. Clients route via the StartupMessage database parameter (e.g. jdbc:postgresql://…/prod lands on the "prod" conn).

Options: :port — Port to listen on (default 5432) :host — Host to bind to (default "127.0.0.1") :on-query — Callback (fn [sql-string]) for logging :default — Database name used when conn-or-registry is a bare conn (default "datahike"). Ignored when a map is supplied. :on-create-database — Hook (fn [db-name parsed-options]) -> conn that runs when a SQL client issues CREATE DATABASE name [WITH …]. Without this hook configured, CREATE DATABASE returns SQLSTATE 0A000 (provisioning from SQL is a deployment policy decision, not a default). See datahike.pg.sql.database/db-from-template for the common template-driven helper. :on-delete-database — Hook (fn [db-name conn parsed-options]) that runs on DROP DATABASE name. Should release the conn and delete the backing store. Symmetric with :on-create-database; without it, DROP DATABASE returns 0A000. :database-template — Convenience shorthand: a partial datahike config template that pg-datahike uses to build both :on-create-database and :on-delete-database via db-from-template / db-delete-from-template. The template can interpolate {{name}} in string values (handy for file backends with per-database paths). Mutually composable with the explicit hooks; if both are given, the explicit hook wins.

Returns a map with :server (PgWireServer), :registry-atom (an atom holding the live {name → conn} map — mutated by SQL CREATE/DROP DATABASE and add-database! / remove-database!), :port, :host.

Examples: ;; single DB, no SQL provisioning (def srv (pg/start-server conn {:port 5433}))

;; multi-DB, static (def srv (pg/start-server {"prod" prod-conn "staging" staging-conn} {:port 5432}))

;; SQL CREATE/DROP DATABASE provisioned in-memory (def srv (pg/start-server {} {:port 5432 :database-template {:store {:backend :memory} :schema-flexibility :write}}))

(pg/stop-server srv)

stop-server clj

(stop-server {:keys [server]})

Stop a running PgWire server.