From a9e44b3a4cff9ee75a39744ba2b4116776ae5ca5 Mon Sep 17 00:00:00 2001 From: max Date: Wed, 15 Jul 2026 15:48:07 +0200 Subject: [PATCH 1/2] Introduce dbval.store: pluggable tuple storage behind a protocol. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine only ever needed an ordered set of byte-array keys with range scans and atomic batch commits — the contract of a transactional ordered key-value store. `dbval.store/ITupleStore` makes that explicit: -scan committed keys in [begin, end), unsigned byte order -commit! atomically add a batch of keys -close! release resources Read-your-writes inside a running transaction moves into the engine: the transaction stages its keys in a pending TreeSet overlay carried by the db value, and `slice` lazily merges the overlay over the store scan. Stores therefore only see committed state, and nothing touches the store until the single atomic commit — a failing transaction just discards the overlay (the JDBC rollback machinery is gone). Two adapters: - `dbval.store.sqlite` (default): the previous storage code. Reads now run with autocommit, so a scan always sees the latest committed state — this also fixes the WAL snapshot pinning that made a second connection to the same file blind to other connections' commits (regression test re-added). Read consistency for snapshots comes from :max-tx filtering, not storage-level read transactions. - `dbval.store.memory`: a ConcurrentSkipListSet, for tests and for running the engine without any storage backend. `empty-db` accepts a :store option; without it the SQLite store is built from :db-file as before. `dbval.test.store` runs representative engine flows (transact, upsert, retract history, query, pull, rseek, index-range, snapshot isolation, failed-transaction atomicity) on the memory store. Co-Authored-By: Claude Fable 5 --- src/dbval/core.clj | 7 +- src/dbval/db.clj | 322 +++++++++++++------------------------ src/dbval/store.clj | 56 +++++++ src/dbval/store/memory.clj | 36 +++++ src/dbval/store/sqlite.clj | 134 +++++++++++++++ test/dbval/test.clj | 1 + test/dbval/test/conn.clj | 11 ++ test/dbval/test/store.clj | 63 ++++++++ 8 files changed, 421 insertions(+), 209 deletions(-) create mode 100644 src/dbval/store.clj create mode 100644 src/dbval/store/memory.clj create mode 100644 src/dbval/store/sqlite.clj create mode 100644 test/dbval/test/store.clj diff --git a/src/dbval/core.clj b/src/dbval/core.clj index 673ecac..577172a 100644 --- a/src/dbval/core.clj +++ b/src/dbval/core.clj @@ -159,8 +159,11 @@ Options are: - :db-file Path to the SQLite file backing this database. - Defaults to a fresh temporary file." + :store The tuple store backing this database. + Defaults to a SQLite store (see dbval.store.sqlite); + dbval.store.memory provides an in-memory store. + :db-file Path to the SQLite file backing this database, when no + :store is given. Defaults to a fresh temporary file." ([] (db/empty-db nil {})) ([schema] diff --git a/src/dbval/db.clj b/src/dbval/db.clj index 6b49c0d..52a77e2 100644 --- a/src/dbval/db.clj +++ b/src/dbval/db.clj @@ -3,10 +3,10 @@ [clojure.data] [clojure.edn :as edn] - [clojure.string :as str] - [clojure.java.io :as io] [dbval.inline :refer [update]] [dbval.lru :as lru] + [dbval.store :as store] + [dbval.store.sqlite :as sqlite-store] [dbval.util :as util] [dbval.arrays :as arrays] [com.yetanalytics.squuid :as squuid]) @@ -271,7 +271,7 @@ ;; ---------------------------------------------------------------------------- -(declare db-conn) +(declare db-store db-pending) (declare indexing?) @@ -479,17 +479,6 @@ (map (bytes-to-datoms-xf db)) byte-tuples)) -(defn byte-array-compare - [^bytes a ^bytes b] - (java.util.Arrays/compareUnsigned - a - b)) - -(def byte-array-comparator - (reify java.util.Comparator - (compare [_ a b] - (byte-array-compare ^bytes a ^bytes b)))) - (defn pack [^com.apple.foundationdb.tuple.Tuple tuple] (try @@ -632,90 +621,50 @@ :aevt) :eavt)))) -(defn- sqlite-jdbc-url - "Builds a full SQLite JDBC URL with optional pragmas." - [db-file {:keys [busy-timeout-ms foreign-keys? wal? read-only? mmap-size sync] - :or {busy-timeout-ms 5000 - foreign-keys? true - wal? true - read-only? false - sync "NORMAL"}}] - (let [params (cond-> {"busy_timeout" busy-timeout-ms - "foreign_keys" (if foreign-keys? "on" "off") - "synchronous" (str sync)} - wal? (assoc "journal_mode" "WAL") - read-only? (assoc "mode" "ro") - mmap-size (assoc "mmap_size" mmap-size)) - qs (->> params - (map (fn [[k v]] (str k "=" v))) - (str/join "&"))] - (str "jdbc:sqlite:" db-file (when (seq qs) (str "?" qs))))) - -(defn ^java.sql.Connection get-sqlite-connection - "Returns a java.sql.Connection for the given SQLite file. - Caller must close the connection manually." - [{:keys [db-file opts]}] - ;; explicitly load the driver class - (java.lang.Class/forName "org.sqlite.JDBC") - (let [^String url (sqlite-jdbc-url db-file opts) - conn (java.sql.DriverManager/getConnection url)] - (.setAutoCommit conn false) - conn)) +(defn- merge-sorted + "Lazily merges two seqs that are sorted by `cmp`, dropping duplicates." + [cmp xs ys] + (lazy-seq + (let [xs (seq xs) + ys (seq ys)] + (cond + (nil? xs) ys + (nil? ys) xs + :else + (let [x (first xs) + y (first ys) + c (cmp x y)] + (cond + (neg? c) (cons x (merge-sorted cmp (rest xs) ys)) + (pos? c) (cons y (merge-sorted cmp xs (rest ys))) + :else (cons x (merge-sorted cmp (rest xs) (rest ys))))))))) (defn slice + "Scans the db's store for keys in [begin, end), in unsigned byte order + (descending when `reverse`). When the db value carries a pending + transaction overlay, its keys are merged into the scan, so reads during + a transaction see the transaction's own uncommitted writes." [{:keys [db ^bytes begin ^bytes end reverse]}] - (reify java.lang.Iterable - (iterator [_] - (let [^java.sql.Connection conn (db-conn db) - ^java.sql.PreparedStatement stmt - (.prepareStatement conn - (str "select k from dbval where k >= ? and k < ?" - (when reverse - " order by k desc")) - java.sql.ResultSet/TYPE_FORWARD_ONLY - java.sql.ResultSet/CONCUR_READ_ONLY) - _ (.setBytes stmt 1 begin) - _ (.setBytes stmt 2 end) - _ (.setFetchSize ^java.sql.Statement stmt 1000) ; hint (SQLite may ignore) - ^java.sql.ResultSet rs (.executeQuery stmt) - - next-val (atom nil) - advanced (atom false) - closed (atom false) - close! (fn [] - (when-not @closed - (reset! closed true) - (try (.close rs) (catch Throwable _)) - (try (.close stmt) (catch Throwable _)) - )) - advance! (fn [] - (when-not @closed - (if (.next rs) - (do (reset! next-val (.getBytes rs "k")) true) - (do (reset! next-val nil) (close!) false))))] - - (reify java.util.Iterator - (hasNext [this] - (or @advanced - (reset! advanced - (boolean (advance!))))) - (next [_] - (let [ok (or @advanced (advance!))] - (when-not ok - (throw (java.util.NoSuchElementException.))) - (let [v @next-val] - (reset! advanced false) - (reset! next-val nil) - v))) - (remove [_] - (throw (UnsupportedOperationException. "remove not supported")))))))) + (let [scan (store/scan (db-store db) begin end (boolean reverse))] + (if-some [^java.util.NavigableSet pending (db-pending db)] + (let [sub (.subSet pending begin true end false) + overlay (if reverse + (.descendingSet ^java.util.NavigableSet sub) + sub) + cmp (if reverse + (fn [a b] (store/byte-array-compare b a)) + (fn [a b] (store/byte-array-compare a b)))] + (merge-sorted cmp + (iterator-seq (.iterator ^java.lang.Iterable scan)) + (seq overlay))) + scan))) ;; An opaque handle to a database value, like Datomic's Db: it hashes and ;; compares by reference identity. To compare two snapshots, compare ;; `basis-tx` (and know which store they came from) — content-based value ;; semantics would have to realize a potentially larger-than-memory database. (deftype DB [schema max-tx rschema pull-patterns pull-attrs - db-file conn] + store pending] IDB @@ -930,18 +879,32 @@ [db] (.-max-tx (unfiltered-db db))) -(defn ^:no-doc ^java.sql.Connection db-conn - "The JDBC connection of the store backing this database value." - [db] - (.-conn (unfiltered-db db))) +(defn ^:no-doc db-store + "The tuple store backing this database value (see `dbval.store`)." + [db] + (.-store (unfiltered-db db))) + +(defn- db-pending + "The pending transaction overlay of this db value: a NavigableSet of the + keys staged by the transaction that produced it, or nil outside of a + transaction." + [db] + (.-pending (unfiltered-db db))) (defn ^:no-doc ^DB with-max-tx "Copy of `db` with a different basis. Low-level; a db value normally gets its basis from the store (see `dbval.conn`) or a transaction." [^DB db max-tx] (DB. (.-schema db) max-tx (.-rschema db) - (.-pull-patterns db) (.-pull-attrs db) - (.-db-file db) (.-conn db))) + (.-pull-patterns db) (.-pull-attrs db) + (.-store db) (.-pending db))) + +(defn- ^DB with-pending + "Copy of `db` with a different pending overlay (nil to clear)." + [^DB db pending] + (DB. (.-schema db) (.-max-tx db) (.-rschema db) + (.-pull-patterns db) (.-pull-attrs db) + (.-store db) pending)) ;; ---------------------------------------------------------------------------- @@ -1049,20 +1012,6 @@ (when (= :db.cardinality/many (:db/cardinality (get schema attr))) (util/raise a " :db/tupleAttrs can't depend on :db.cardinality/many attribute: " attr ex-data)))))))) -(defn execute-sql! - "Executes a single SQL statement using the given java.sql.Connection. - Returns true if the statement returned a ResultSet, or false for an update/count. - Example: - (execute-sql! conn \"create table if not exists users (id integer primary key, name text)\")" - [^java.sql.Connection conn ^String sql] - (with-open [stmt (.createStatement conn)] - (.execute ^java.sql.Statement stmt sql))) - -(defn create-table! - [^java.sql.Connection conn] - ;; Create a table - (execute-sql! conn "create table if not exists dbval (k blob not null, primary key(k)) WITHOUT ROWID;")) - (defn q-max-tx [db] (let [[begin end] (tuple-range "teav") @@ -1079,22 +1028,16 @@ (defn ^DB empty-db [schema opts] {:pre [(or (nil? schema) (map? schema))]} (validate-schema schema) - (let [db-file (or (:db-file opts) - (.getCanonicalPath - (java.io.File/createTempFile (str (random-uuid)) - ".db"))) - _ (io/make-parents db-file) - ;; TODO: consider how to close the connection: - conn ^java.sql.Connection (get-sqlite-connection {:db-file db-file}) - _ (create-table! conn) - _ (.commit conn) - db (DB. schema - tx0 - (rschema (merge implicit-schema schema)) - (lru/cache 100) - (lru/cache 100) - db-file - conn)] + ;; TODO: consider how to close the store: + (let [store (or (:store opts) + (sqlite-store/store opts)) + db (DB. schema + tx0 + (rschema (merge implicit-schema schema)) + (lru/cache 100) + (lru/cache 100) + store + nil)] (with-max-tx db (q-max-tx db)))) (defrecord TxReport [db-before db-after tx-data tempids tx-meta]) @@ -1132,12 +1075,12 @@ (defn ^DB with-schema [^DB db schema] {:pre [(db? db) (or (nil? schema) (map? schema))]} (DB. schema - (.-max-tx db) - (rschema (merge implicit-schema schema)) - (lru/cache 100) - (lru/cache 100) - (.-db-file db) - (.-conn db))) + (.-max-tx db) + (rschema (merge implicit-schema schema)) + (lru/cache 100) + (lru/cache 100) + (.-store db) + (.-pending db))) (do @@ -1686,12 +1629,9 @@ false)) (defn set-add! - [db stmt tuple] + [db ^java.util.NavigableSet pending tuple] (try - (.setBytes ^java.sql.PreparedStatement stmt - 1 - (pack tuple)) - (.addBatch ^java.sql.PreparedStatement stmt) + (.add pending (pack tuple)) (catch Exception e (throw (ex-info "set-add! failed" {:tuple tuple} @@ -1699,45 +1639,36 @@ db) (defn all-tuples - "Returns a reducible of all tuples stored in the dbval table. - Each tuple is decoded from its byte representation. - Useful for debugging and inspecting the raw storage. - Example: (into [] (take 10) (all-tuples db))" + "Returns a reducible of all tuples in the db's store (committed, plus the + pending overlay if this db value carries one). Each tuple is decoded from + its byte representation. Useful for debugging and inspecting the raw + storage. Example: (into [] (take 10) (all-tuples db))" [db] - (reify clojure.lang.IReduceInit - (reduce [_ rf init] - (let [conn ^java.sql.Connection (db-conn db)] - (with-open [stmt (.prepareStatement conn "SELECT k FROM dbval ORDER BY k") - rs (.executeQuery stmt)] - (loop [state init] - (if (or (reduced? state) (not (.next rs))) - (unreduced state) - (recur (rf state (vec (tuple-from-bytes (.getBytes rs "k")))))))))))) + (let [[begin end] (tuple-range)] + (eduction (map (comp vec tuple-from-bytes)) + (slice {:db db :begin begin :end end})))) (defn with-datom [db ^Datom datom] (validate-datom db datom) - (let [conn ^java.sql.Connection (db-conn db) - stmt ^java.sql.PreparedStatement (.prepareStatement - conn - "INSERT OR IGNORE INTO dbval (k) VALUES (?)") - indexing? (indexing? db (.-a datom)) - db (if (datom-added datom) - (-> db - (set-add! stmt (datom-tuple db :eavt datom)) - (set-add! stmt (datom-tuple db :aevt datom)) - (cond-> indexing? (set-add! stmt (datom-tuple db :avet datom))) - (set-add! stmt (datom-tuple db :teav datom))) - (if-some [removing (some-> (fsearch db [(.-e datom) (.-a datom) (.-v datom)]) - (retract-datom (:tx datom)))] - (-> db - (set-add! stmt (datom-tuple db :eavt removing)) - (set-add! stmt (datom-tuple db :aevt removing)) - (cond-> indexing? (set-add! stmt (datom-tuple db :avet removing))) - (set-add! stmt (datom-tuple db :teav removing))) - db))] - (.executeBatch stmt) - (.close stmt) - db)) + (let [^java.util.NavigableSet pending (db-pending db) + _ (when (nil? pending) + (util/raise "with-datom outside of a transaction" + {:error :transact/no-pending})) + indexing? (indexing? db (.-a datom))] + (if (datom-added datom) + (-> db + (set-add! pending (datom-tuple db :eavt datom)) + (set-add! pending (datom-tuple db :aevt datom)) + (cond-> indexing? (set-add! pending (datom-tuple db :avet datom))) + (set-add! pending (datom-tuple db :teav datom))) + (if-some [removing (some-> (fsearch db [(.-e datom) (.-a datom) (.-v datom)]) + (retract-datom (:tx datom)))] + (-> db + (set-add! pending (datom-tuple db :eavt removing)) + (set-add! pending (datom-tuple db :aevt removing)) + (cond-> indexing? (set-add! pending (datom-tuple db :avet removing))) + (set-add! pending (datom-tuple db :teav removing))) + db)))) (defn- queue-tuple [queue tuple idx db e a v] (let [tuple-attrs (-> db (-schema) (get tuple) :db/tupleAttrs) @@ -2227,53 +2158,30 @@ (util/raise "Bad entity type at " entity ", expected map or vector" {:error :transact/syntax, :tx-data entity}))))) -(defmacro with-transaction - "Run BODY within a JDBC transaction on CONN. - - If CONN is already in a transaction (autocommit=false), just join it. - - If autocommit=true, disable it, start a transaction, and commit/rollback at the end." - [^java.sql.Connection conn & body] - `(let [was-auto?# (.getAutoCommit ~conn)] - (if-not was-auto?# - ;; Already inside a transaction — just run the body - (do ~@body) - ;; Start and manage a new transaction - (do - (.setAutoCommit ~conn false) - (try - (let [res# (do ~@body)] - (.commit ~conn) - res#) - (catch Throwable t# - (try (.rollback ~conn) (catch Throwable _#)) - (throw t#)) - (finally - (.setAutoCommit ~conn true))))))) - (defn transact-tx-data [report es] (when-not (or (nil? es) (sequential? es)) (util/raise "Bad transaction data " es ", expected sequential collection" {:error :transact/syntax, :tx-data es})) - (let [tx-id (squuid/generate-squuid) + (let [tx-id (squuid/generate-squuid) + ;; The pending overlay collects this transaction's keys; reads during + ;; the transaction merge it over the store (see `slice`), so nothing + ;; touches the store until the final atomic commit — an exception + ;; while transacting simply discards the overlay. + pending (java.util.TreeSet. ^java.util.Comparator store/byte-array-comparator) report' (-> report (assoc ::tx-id tx-id) ;; Set max-tx to current tx-id so datoms added during this ;; transaction are visible when searching for duplicates - (update :db-after with-max-tx tx-id)) + (update :db-after with-max-tx tx-id) + (update :db-after with-pending pending)) {:keys [tx-data id-map]} (assign-entity-ids (:db-before report') es) ;; Pre-populate tempids with the tempid -> UUID mapping report'' (update report' :tempids merge id-map) - conn ^java.sql.Connection (db-conn (:db-after report''))] - (try - (let [result (with-transaction conn - (transact-tx-data-impl report'' tx-data))] - (.commit conn) + result (transact-tx-data-impl report'' tx-data)] + (store/commit! (db-store (:db-after result)) (seq pending)) + (-> result + (update :db-after with-pending nil) ;; Add :tx field with the transaction UUID - (assoc result :tx tx-id)) - (catch Throwable t - (try - (.rollback conn) - (.setAutoCommit conn false) - (catch Throwable _)) - (throw t))))) + (assoc :tx tx-id)))) diff --git a/src/dbval/store.clj b/src/dbval/store.clj new file mode 100644 index 0000000..ba77e87 --- /dev/null +++ b/src/dbval/store.clj @@ -0,0 +1,56 @@ +(ns dbval.store + "Storage abstraction for dbval. + + A store is an ordered set of byte-array keys (FoundationDB-tuple encoded + datoms, see `dbval.db`) that supports range scans over committed data and + atomic batch commits. dbval only needs the key portion: conceptually the + store is a sorted set, mimicking a transactional ordered key-value store + like FoundationDB. + + Stores never see uncommitted state: read-your-writes inside a running + transaction is handled by the engine (`dbval.db`), which overlays the + transaction's pending keys over `-scan`. A store implementation therefore + only has to provide: + + - `-scan`: committed keys in unsigned byte order + - `-commit!`: atomically add a batch of keys (all or nothing) + + Implementations: `dbval.store.sqlite` (default), `dbval.store.memory`.") + +(defprotocol ITupleStore + (-scan [store begin end reverse?] + "Returns an Iterable/seqable of byte[] keys k with begin <= k < end, + compared in unsigned byte order, ascending — or descending when + `reverse?`. Only committed keys are visible.") + (-commit! [store keys] + "Atomically adds the byte[] `keys` to the store: after `-commit!` + returns, either all keys are durably visible to subsequent scans or — + if it throws — none are. Keys that already exist are ignored.") + (-close! [store] + "Releases the store's resources.")) + +(defn scan + "See [[ITupleStore]]." + [store begin end reverse?] + (-scan store begin end reverse?)) + +(defn commit! + "See [[ITupleStore]]." + [store keys] + (-commit! store keys)) + +(defn close! + "See [[ITupleStore]]." + [store] + (-close! store)) + +(defn byte-array-compare + ^long [^bytes a ^bytes b] + (java.util.Arrays/compareUnsigned a b)) + +(def byte-array-comparator + "Unsigned lexicographic byte[] comparator — the key order every store + must scan in." + (reify java.util.Comparator + (compare [_ a b] + (byte-array-compare ^bytes a ^bytes b)))) diff --git a/src/dbval/store/memory.clj b/src/dbval/store/memory.clj new file mode 100644 index 0000000..d33eafa --- /dev/null +++ b/src/dbval/store/memory.clj @@ -0,0 +1,36 @@ +(ns dbval.store.memory + "In-memory tuple store: a concurrent sorted set of byte-array keys. + + Nothing is persisted — useful for tests and for exercising the engine + without any storage backend. Scans return a live view of the set; that + is safe because the engine filters every datom by the snapshot's + `:max-tx`, so keys committed after a snapshot was taken are invisible + to it regardless of when they appear in a scan." + (:require + [dbval.store :as store]) + (:import + [java.util.concurrent ConcurrentSkipListSet])) + +(deftype MemoryStore [^ConcurrentSkipListSet keyset] + store/ITupleStore + (-scan [_ begin end reverse?] + (let [sub (.subSet keyset begin true end false)] + (if reverse? + (.descendingSet ^java.util.NavigableSet sub) + sub))) + + (-commit! [this keys] + ;; single writer at a time keeps the batch atomic with respect to other + ;; commits; readers may observe a batch mid-insert, but the engine's + ;; :max-tx filtering makes those keys invisible until the transaction's + ;; basis is handed out + (locking this + (doseq [^bytes k keys] + (.add keyset k)))) + + (-close! [_] nil)) + +(defn store + "Creates an empty in-memory tuple store." + ^dbval.store.memory.MemoryStore [] + (MemoryStore. (ConcurrentSkipListSet. ^java.util.Comparator store/byte-array-comparator))) diff --git a/src/dbval/store/sqlite.clj b/src/dbval/store/sqlite.clj new file mode 100644 index 0000000..382f302 --- /dev/null +++ b/src/dbval/store/sqlite.clj @@ -0,0 +1,134 @@ +(ns dbval.store.sqlite + "SQLite-backed tuple store: one table holding the sorted keys. + + create table dbval (k blob not null, primary key(k)) WITHOUT ROWID; + + The JDBC connection runs with autocommit on, so every scan reads the + latest committed state (no lingering WAL read transaction pinning an old + snapshot); `-commit!` wraps its batch insert in a single transaction." + (:require + [clojure.java.io :as io] + [clojure.string :as str] + [dbval.store :as store])) + +(set! *warn-on-reflection* true) + +(defn- sqlite-jdbc-url + "Builds a full SQLite JDBC URL with optional pragmas." + [db-file {:keys [busy-timeout-ms foreign-keys? wal? read-only? mmap-size sync] + :or {busy-timeout-ms 5000 + foreign-keys? true + wal? true + read-only? false + sync "NORMAL"}}] + (let [params (cond-> {"busy_timeout" busy-timeout-ms + "foreign_keys" (if foreign-keys? "on" "off") + "synchronous" (str sync)} + wal? (assoc "journal_mode" "WAL") + read-only? (assoc "mode" "ro") + mmap-size (assoc "mmap_size" mmap-size)) + qs (->> params + (map (fn [[k v]] (str k "=" v))) + (str/join "&"))] + (str "jdbc:sqlite:" db-file (when (seq qs) (str "?" qs))))) + +(defn- ^java.sql.Connection get-connection + [db-file opts] + ;; explicitly load the driver class + (java.lang.Class/forName "org.sqlite.JDBC") + (let [^String url (sqlite-jdbc-url db-file opts)] + (java.sql.DriverManager/getConnection url))) + +(defn- create-table! [^java.sql.Connection conn] + (with-open [stmt (.createStatement conn)] + (.execute ^java.sql.Statement stmt + "create table if not exists dbval (k blob not null, primary key(k)) WITHOUT ROWID;"))) + +(defn- scan-iterator + ^java.util.Iterator [^java.sql.Connection conn ^bytes begin ^bytes end reverse?] + (let [^java.sql.PreparedStatement stmt + (.prepareStatement conn + (str "select k from dbval where k >= ? and k < ?" + (when reverse? + " order by k desc")) + java.sql.ResultSet/TYPE_FORWARD_ONLY + java.sql.ResultSet/CONCUR_READ_ONLY) + _ (.setBytes stmt 1 begin) + _ (.setBytes stmt 2 end) + _ (.setFetchSize ^java.sql.Statement stmt 1000) ; hint (SQLite may ignore) + ^java.sql.ResultSet rs (.executeQuery stmt) + + next-val (atom nil) + advanced (atom false) + closed (atom false) + close! (fn [] + (when-not @closed + (reset! closed true) + (try (.close rs) (catch Throwable _)) + (try (.close stmt) (catch Throwable _)))) + advance! (fn [] + (when-not @closed + (if (.next rs) + (do (reset! next-val (.getBytes rs "k")) true) + (do (reset! next-val nil) (close!) false))))] + (reify java.util.Iterator + (hasNext [this] + (or @advanced + (reset! advanced + (boolean (advance!))))) + (next [_] + (let [ok (or @advanced (advance!))] + (when-not ok + (throw (java.util.NoSuchElementException.))) + (let [v @next-val] + (reset! advanced false) + (reset! next-val nil) + v))) + (remove [_] + (throw (UnsupportedOperationException. "remove not supported")))))) + +(deftype SqliteStore [^java.sql.Connection conn db-file] + store/ITupleStore + (-scan [_ begin end reverse?] + (reify java.lang.Iterable + (iterator [_] + (scan-iterator conn begin end (boolean reverse?))))) + + (-commit! [this keys] + (when (seq keys) + (locking this + (.setAutoCommit conn false) + (try + (with-open [stmt (.prepareStatement conn "INSERT OR IGNORE INTO dbval (k) VALUES (?)")] + (doseq [^bytes k keys] + (.setBytes stmt 1 k) + (.addBatch stmt)) + (.executeBatch stmt)) + (.commit conn) + (catch Throwable t + (try (.rollback conn) (catch Throwable _)) + (throw t)) + (finally + (.setAutoCommit conn true)))))) + + (-close! [_] + (.close conn))) + +(defn store + "Opens (creating if necessary) a SQLite-backed tuple store. + + Options: + + :db-file Path to the SQLite file. Defaults to a fresh + temporary file. + :opts SQLite pragmas: :busy-timeout-ms, :foreign-keys?, + :wal?, :read-only?, :mmap-size, :sync." + ^dbval.store.sqlite.SqliteStore [{:keys [db-file opts]}] + (let [db-file (or db-file + (.getCanonicalPath + (java.io.File/createTempFile (str (random-uuid)) + ".db"))) + _ (io/make-parents db-file) + conn (get-connection db-file opts)] + (create-table! conn) + (SqliteStore. conn db-file))) diff --git a/test/dbval/test.clj b/test/dbval/test.clj index 6d50bb8..5fa33dc 100644 --- a/test/dbval/test.clj +++ b/test/dbval/test.clj @@ -22,6 +22,7 @@ dbval.test.parser-where dbval.test.pull-api dbval.test.pull-parser + dbval.test.store dbval.test.query dbval.test.query-aggregates dbval.test.query-find-specs diff --git a/test/dbval/test/conn.clj b/test/dbval/test/conn.clj index 978a0de..e3f36c6 100644 --- a/test/dbval/test/conn.clj +++ b/test/dbval/test/conn.clj @@ -65,6 +65,17 @@ (is (thrown-msg? "underlying tuple store has already been modified" (d/db-with db1 [{:name "Oleg"}]))))) +(deftest test-deref-sees-other-connections + ;; two connections to the same SQLite file: reads run with autocommit, so + ;; a deref always sees the latest committed transaction instead of a + ;; pinned WAL read snapshot + (let [db-file (str (System/getProperty "java.io.tmpdir") + "/dbval-test-" (random-uuid) ".db") + conn1 (d/create-conn nil {:db-file db-file}) + conn2 (d/create-conn nil {:db-file db-file})] + (d/transact! conn1 [{:name "Ivan"}]) + (is (= ["Ivan"] (mapv :v (d/datoms @conn2 :aevt :name)))))) + (deftest test-transact!-not-repeated-by-concurrent-conn-update ;; regression: `-transact!` used to run the (side-effecting, committing) ;; transaction inside `swap!`; a concurrent update of the conn state diff --git a/test/dbval/test/store.clj b/test/dbval/test/store.clj new file mode 100644 index 0000000..ffc79a7 --- /dev/null +++ b/test/dbval/test/store.clj @@ -0,0 +1,63 @@ +(ns dbval.test.store + "Exercises the engine against a non-default `dbval.store` implementation. + The regular suite covers the SQLite store (the default); this namespace + runs representative flows on the in-memory store to prove the engine is + storage-agnostic." + (:require + [clojure.test :as t :refer [is deftest testing]] + [dbval.core :as d] + [dbval.store.memory :as memory] + [dbval.test.core])) + +(defn- empty-mem-db + ([] (empty-mem-db nil)) + ([schema] (d/empty-db schema {:store (memory/store)}))) + +(deftest test-memory-store-transact-and-query + (let [conn (d/conn-from-db + (empty-mem-db {:name {:db/unique :db.unique/identity} + :aka {:db/cardinality :db.cardinality/many} + :friend {:db/valueType :db.type/ref} + :age {:db/index true}}))] + (d/transact! conn [{:db/id "ivan" :name "Ivan" :age 30 :aka ["I" "Terrible"]} + {:db/id "petr" :name "Petr" :age 44 :friend "ivan"}]) + (let [db @conn + ivan (:e (first (d/datoms db :avet :name "Ivan"))) + petr (:e (first (d/datoms db :avet :name "Petr")))] + (testing "query" + (is (= #{["Ivan" 30] ["Petr" 44]} + (d/q '[:find ?n ?a :where [?e :name ?n] [?e :age ?a]] db)))) + + (testing "pull and entity over refs" + (is (= "Ivan" (get-in (d/pull db [{:friend [:name]}] petr) [:friend :name]))) + (is (= "Ivan" (:name (:friend (d/entity db petr)))))) + + (testing "upsert redirects to the existing entity" + (d/transact! conn [{:name "Ivan" :age 31}]) + (is (= [31] (mapv :v (d/datoms @conn :eavt ivan :age))))) + + (testing "retraction with history stays invisible, forward and reverse" + (d/transact! conn [[:db/retract ivan :aka "Terrible"]]) + (let [db @conn] + (is (= #{"I"} (set (map :v (d/datoms db :eavt ivan :aka))))) + (is (= #{"I"} (set (map :v (filter #(= :aka (:a %)) + (d/rseek-datoms db :eavt ivan)))))))) + + (testing "index-range" + (is (= [31 44] (mapv :v (d/index-range @conn :age 0 100))))) + + (testing "snapshot isolation across stores" + (let [snapshot @conn] + (d/transact! conn [{:name "Oleg" :age 11}]) + (is (= [31 44] (mapv :v (d/index-range snapshot :age 0 100)))) + (is (= [11 31 44] (mapv :v (d/index-range @conn :age 0 100))))))))) + +(deftest test-memory-store-transaction-isolation + ;; a failing transaction must leave the store untouched: nothing is + ;; written until the pending overlay commits atomically + (let [conn (d/conn-from-db (empty-mem-db {:name {:db/unique :db.unique/identity}}))] + (d/transact! conn [{:name "Ivan"}]) + (is (thrown? clojure.lang.ExceptionInfo + (d/transact! conn [{:name "Oleg"} + [:db/add "x" :bad nil]]))) + (is (= ["Ivan"] (mapv :v (d/datoms @conn :aevt :name)))))) From da2437ffb79774b0c088b76223c3857f5a63ab46 Mon Sep 17 00:00:00 2001 From: max Date: Wed, 15 Jul 2026 21:09:40 +0200 Subject: [PATCH 2/2] Ship no storage driver: sqlite-jdbc becomes a consumer-provided dep. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Like next.jdbc, dbval no longer declares a JDBC driver in :deps — a consumer using the SlateDB or memory backend should not download the SQLite driver's bundled native libraries. The SQLite adapter stays in the artifact (it only references java.sql and loads org.sqlite.JDBC reflectively), `empty-db` resolves it lazily via requiring-resolve, and a missing driver now produces an instructive error instead of a bare ClassNotFoundException. The :dev and :bench aliases provide the driver for the test suite and benchmarks. Co-Authored-By: Claude Fable 5 --- README.md | 9 +++++++++ deps.edn | 7 ++++--- src/dbval/db.clj | 10 ++++++++-- src/dbval/store/sqlite.clj | 16 ++++++++++++++-- 4 files changed, 35 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 219e925..c65da4e 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,15 @@ dbval only needs the key portion. Consequently, you are dealing with a sorted set and Datascript's core is a [persistent-sorted-set](https://github.com/tonsky/persistent-sorted-set). +The storage layer is pluggable via the `dbval.store` protocol: a store only +has to provide ordered range scans over committed keys and atomic batch +commits. `dbval.store.sqlite` is the default backend and +`dbval.store.memory` provides an in-memory store for tests. Note that dbval +ships no storage driver — to use the default SQLite store, add +`org.xerial/sqlite-jdbc` to your dependencies (like with +[next.jdbc](https://github.com/seancorfield/next-jdbc), you bring the driver +for the store you pick). + I first assumed that I need to make one part of a datom mutable, so that I can mark it as retracted. Until I discovered that the sorting of `t` in the Datomic indexes `:eavt`, `:aevt`, `:avet` and `:vaet` allows to figure out what the diff --git a/deps.edn b/deps.edn index 569483d..8075246 100644 --- a/deps.edn +++ b/deps.edn @@ -1,6 +1,5 @@ {:deps {org.foundationdb/fdb-java {:mvn/version "7.3.47"} - org.xerial/sqlite-jdbc {:mvn/version "3.47.1.0"} com.yetanalytics/colossal-squuid {:mvn/version "0.1.5" :exclusions [ ;; important to avoid `java.lang.NoSuchMethodError: 'boolean com.google.protobuf.GeneratedMessageV3.isStringEmpty(java.lang.Object)' at com.google.cloud.secretmanager.v1.ListSecretsRequest.getSerializedSize(ListSecretsRequest.java:355)` @@ -30,7 +29,8 @@ :jvm-opts ["-ea" "-Ddbval.debug" "-Dclojure.main.report=stderr" "-Djdk.attach.allowAttachSelf"] :extra-deps {io.github.tonsky/duti {:git/sha "e36d65296a4f9758664309ec35b00887e88c405a"} - com.cognitect/transit-clj {:mvn/version "1.0.324"}}} + com.cognitect/transit-clj {:mvn/version "1.0.324"} + org.xerial/sqlite-jdbc {:mvn/version "3.47.1.0"}}} :bench {:extra-paths ["bench"] @@ -39,4 +39,5 @@ "-Djdk.attach.allowAttachSelf" "-XX:+DebugNonSafepoints"] :extra-deps - {io.github.tonsky/duti {:git/sha "e36d65296a4f9758664309ec35b00887e88c405a"}}}}} + {io.github.tonsky/duti {:git/sha "e36d65296a4f9758664309ec35b00887e88c405a"} + org.xerial/sqlite-jdbc {:mvn/version "3.47.1.0"}}}}} diff --git a/src/dbval/db.clj b/src/dbval/db.clj index 52a77e2..c67c5fd 100644 --- a/src/dbval/db.clj +++ b/src/dbval/db.clj @@ -6,7 +6,6 @@ [dbval.inline :refer [update]] [dbval.lru :as lru] [dbval.store :as store] - [dbval.store.sqlite :as sqlite-store] [dbval.util :as util] [dbval.arrays :as arrays] [com.yetanalytics.squuid :as squuid]) @@ -1025,12 +1024,19 @@ (second)) tx0))) +(defn- default-store + "Builds the default SQLite-backed store. Loaded lazily: dbval ships no + storage driver, so the SQLite adapter (and its driver requirement) is + only touched when no explicit :store is given." + [opts] + ((requiring-resolve 'dbval.store.sqlite/store) opts)) + (defn ^DB empty-db [schema opts] {:pre [(or (nil? schema) (map? schema))]} (validate-schema schema) ;; TODO: consider how to close the store: (let [store (or (:store opts) - (sqlite-store/store opts)) + (default-store opts)) db (DB. schema tx0 (rschema (merge implicit-schema schema)) diff --git a/src/dbval/store/sqlite.clj b/src/dbval/store/sqlite.clj index 382f302..14a21c8 100644 --- a/src/dbval/store/sqlite.clj +++ b/src/dbval/store/sqlite.clj @@ -32,10 +32,22 @@ (str/join "&"))] (str "jdbc:sqlite:" db-file (when (seq qs) (str "?" qs))))) +(defn- load-driver! [] + (try + (java.lang.Class/forName "org.sqlite.JDBC") + (catch ClassNotFoundException e + (throw (ex-info + (str "SQLite JDBC driver not found on the classpath. " + "dbval ships no storage driver: add org.xerial/sqlite-jdbc " + "to your dependencies to use the default SQLite store, " + "or pass an explicit :store to empty-db " + "(e.g. dbval.store.memory).") + {:error :store/missing-driver} + e))))) + (defn- ^java.sql.Connection get-connection [db-file opts] - ;; explicitly load the driver class - (java.lang.Class/forName "org.sqlite.JDBC") + (load-driver!) (let [^String url (sqlite-jdbc-url db-file opts)] (java.sql.DriverManager/getConnection url)))