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/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..c67c5fd 100644 --- a/src/dbval/db.clj +++ b/src/dbval/db.clj @@ -3,10 +3,9 @@ [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.util :as util] [dbval.arrays :as arrays] [com.yetanalytics.squuid :as squuid]) @@ -271,7 +270,7 @@ ;; ---------------------------------------------------------------------------- -(declare db-conn) +(declare db-store db-pending) (declare indexing?) @@ -479,17 +478,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 +620,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 +878,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 +1011,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") @@ -1076,25 +1024,26 @@ (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) - (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) + (default-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 +1081,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 +1635,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 +1645,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 +2164,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..14a21c8 --- /dev/null +++ b/src/dbval/store/sqlite.clj @@ -0,0 +1,146 @@ +(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- 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] + (load-driver!) + (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))))))