diff --git a/src/dbval/db.clj b/src/dbval/db.clj index f50476f..014bf4a 100644 --- a/src/dbval/db.clj +++ b/src/dbval/db.clj @@ -328,6 +328,91 @@ ;; ---------------------------------------------------------------------------- +;; Deref value types: attributes flagged with {:dbval/deref true} keep only a +;; SHA-256 content hash of their value in the index keys; the value's pr-str +;; bytes are stored once, content-addressed, in the store's blob area (see +;; `dbval.store`). Reads return a `BlobRef` and the value is only fetched and +;; parsed on `deref`. This keeps large values (which would otherwise blow up +;; key sizes and scan costs) out of the indexes, while equality — datom +;; equality, transact no-op detection, upserts and datalog joins — keeps +;; working by comparing hashes: same content <=> same hash. + +(declare deref-attr? db-get-blob) + +(defn sha-256 + "SHA-256 digest of `bytes`." + ^bytes [^bytes bytes] + (.digest (java.security.MessageDigest/getInstance "SHA-256") bytes)) + +(def ^:private blob-unrealized + "Sentinel marking a BlobRef whose value has not been parsed yet." + (Object.)) + +;; `inline-str` is only set on refs deserialized from legacy inline datoms +;; (written before their attribute was flagged as deref): such refs +;; re-serialize to the original inline form, so retraction keys stay adjacent +;; to the assertion keys they cancel out. +(deftype BlobRef [db ^bytes hash ^bytes bytes inline-str + ^:unsynchronized-mutable value] + clojure.lang.IDeref + (deref [this] + (locking this + (when (identical? blob-unrealized value) + (let [^bytes bs (or bytes (db-get-blob db hash))] + (when (nil? bs) + (util/raise "No blob found for deref value" + {:error :blob/not-found})) + (set! value (edn/read-string (String. bs java.nio.charset.StandardCharsets/UTF_8))))) + value)) + + clojure.lang.IPending + (isRealized [this] + (locking this + (not (identical? blob-unrealized value)))) + + clojure.lang.IHashEq + (hasheq [this] (java.util.Arrays/hashCode hash)) + + Object + (hashCode [this] (java.util.Arrays/hashCode hash)) + (equals [this other] + (or (identical? this other) + (and (instance? BlobRef other) + (java.util.Arrays/equals hash ^bytes (.-hash ^BlobRef other)))))) + +(defn- bytes->hex + ;; java.util.HexFormat needs JDK 17+, but dbval still supports JDK 11 + ^String [^bytes bytes] + (let [sb (StringBuilder. (* 2 (alength bytes)))] + (dotimes [i (alength bytes)] + (let [b (bit-and (aget bytes i) 0xff)] + (when (< b 0x10) + (.append sb \0)) + (.append sb (Integer/toHexString b)))) + (str sb))) + +(defmethod print-method BlobRef [^BlobRef blob-ref ^java.io.Writer w] + ;; prints the content hash, never the value: printing (logs, REPL) must not + ;; fetch the blob + (.write w "#dbval/blob-ref \"") + (.write w (bytes->hex (.-hash blob-ref))) + (.write w "\"")) + +(defn blob-ref? [x] + (instance? BlobRef x)) + +(defn ^BlobRef value->blob-ref + "Wraps `v` — a value of a deref attribute — into a [[BlobRef]] carrying the + SHA-256 of its pr-str bytes. Passes an existing BlobRef through unchanged, + so values copied from query results are never re-serialized or fetched." + [db v] + (if (blob-ref? v) + v + (let [^bytes bs (.getBytes (pr-str v) java.nio.charset.StandardCharsets/UTF_8)] + (BlobRef. db (sha-256 bs) bs nil v)))) + +;; ---------------------------------------------------------------------------- + (defn ^com.apple.foundationdb.tuple.Tuple tuple "Turns the `components` into a `com.apple.foundationdb.tuple.Tuple`." [& components] @@ -353,15 +438,47 @@ :else x)) +(def max-inline-value-bytes + "Maximum size of a serialized value inside an index key. Values are stored + in every index key of their datom, and SlateDB caps keys at 64 KiB, so + larger values must live in the blob area via a {:dbval/deref true} + attribute." + 60000) + +(defn- validate-inline-size + ^String [attr ^String s] + ;; a String of n chars is at least n and at most 3n UTF-8 bytes (surrogate + ;; pairs of 4-byte code points are 2 chars); only compute the exact byte + ;; count when the cheap char-count bound cannot rule out an overflow + (when (and (> (* 3 (.length s)) max-inline-value-bytes) + (> (alength (.getBytes s java.nio.charset.StandardCharsets/UTF_8)) + max-inline-value-bytes)) + (util/raise "Value of attribute " attr " serializes to more than " + max-inline-value-bytes " bytes and cannot be stored inside " + "index keys. Flag the attribute with {:dbval/deref true} to " + "store its values in the blob area instead." + {:error :transact/value-too-large + :attribute attr + :length (.length s)})) + s) + (defn serialize-value [db attr v] (cond + (blob-ref? v) + (or (.-inline-str ^BlobRef v) + (.-hash ^BlobRef v)) + + ;; nil marks an unbound search-pattern component and must stay nil + (and (some? v) (deref-attr? db attr)) + (.-hash (value->blob-ref db v)) + (or (map? v) (keyword? v) (symbol? v) (string? v) (instance? java.util.Date v)) - (pr-str v) + (validate-inline-size attr (pr-str v)) (sequential? v) (serialize-tuple v) @@ -419,10 +536,14 @@ (list (name order) t e (attr-sort-key a) (serialize-value db a v) added) )) (catch Exception e - (throw (ex-info "tuple-list failed" - {:order order - :datom datom} - e))))) + (if (= :transact/value-too-large (:error (ex-data e))) + ;; pass through unchanged: wrapping would put the oversized datom + ;; into ex-data and thereby into every log of the error + (throw e) + (throw (ex-info "tuple-list failed" + {:order order + :datom datom} + e)))))) (defn tuple-range "Turns the `components` into a `com.apple.foundationdb.tuple.Tuple` and returns @@ -463,13 +584,22 @@ (defn deserialize-value [db attr v] - (if (string? v) - (edn/read-string v) - (if (tuple? db attr) - (deserialize-tuple v) - (if (-> (-schema db) (get attr) :db/tupleAttrs) + (if (deref-attr? db attr) + (if (string? v) + ;; legacy datom written before `attr` was flagged as deref: the + ;; serialized value still lives inline in the index key. Hashing it + ;; here keeps equality consistent with blob-backed datoms of the same + ;; value, since both hash the same pr-str bytes. + (let [^bytes bs (.getBytes ^String v java.nio.charset.StandardCharsets/UTF_8)] + (BlobRef. db (sha-256 bs) bs v blob-unrealized)) + (BlobRef. db v nil nil blob-unrealized)) + (if (string? v) + (edn/read-string v) + (if (tuple? db attr) (deserialize-tuple v) - v)))) + (if (-> (-schema db) (get attr) :db/tupleAttrs) + (deserialize-tuple v) + v))))) (defn datom-from-tuple "Reads back a datom that was stored as `com.apple.foundationdb.tuple.Tuple`." @@ -747,7 +877,7 @@ ;; `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 - store pending as-of-tx since-tx history?] + store pending pending-blobs as-of-tx since-tx history?] IDB @@ -757,6 +887,9 @@ ISearch (-search [db pattern] (let [[e a v tx] pattern + v (if (and (some? a) (some? v) (deref-attr? db a)) + (value->blob-ref db v) + v) index (pattern->order db pattern) [begin end] (apply tuple-range @@ -959,13 +1092,30 @@ [db] (.-pending (unfiltered-db db))) +(defn- db-pending-blobs + "The pending blob overlay of this db value: a Map of content hash bytes -> + value bytes staged by the transaction that produced it, or nil outside of + a transaction." + [db] + (.-pending-blobs (unfiltered-db db))) + +(defn ^:no-doc db-get-blob + "Returns the blob bytes stored under the content `hash` as visible to this + db value: the pending transaction overlay first, then the committed + store. Used by BlobRef deref, so a transaction function can deref a value + asserted earlier in the same transaction." + ^bytes [db ^bytes hash] + (or (when-some [^java.util.Map blobs (db-pending-blobs db)] + (.get blobs hash)) + (store/get-blob (db-store db) hash))) + (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) - (.-store db) (.-pending db) + (.-store db) (.-pending db) (.-pending-blobs db) (.-as-of-tx db) (.-since-tx db) (.-history? db))) (declare with-pending) @@ -990,7 +1140,7 @@ (let [tx (coerce-tx t)] (DB. (.-schema db) tx (.-rschema db) (.-pull-patterns db) (.-pull-attrs db) - (.-store db) (.-pending db) + (.-store db) (.-pending db) (.-pending-blobs db) tx (.-since-tx db) (.-history? db)))) (defn as-of-t @@ -1006,7 +1156,7 @@ {:pre [(instance? DB db)]} (DB. (.-schema db) (.-max-tx db) (.-rschema db) (.-pull-patterns db) (.-pull-attrs db) - (.-store db) (.-pending db) + (.-store db) (.-pending db) (.-pending-blobs db) (.-as-of-tx db) (coerce-tx t) (.-history? db))) (defn since-t @@ -1023,7 +1173,7 @@ {:pre [(instance? DB db)]} (DB. (.-schema db) (.-max-tx db) (.-rschema db) (.-pull-patterns db) (.-pull-attrs db) - (.-store db) (.-pending db) + (.-store db) (.-pending db) (.-pending-blobs db) (.-as-of-tx db) (.-since-tx db) true)) (defn temporal-view? @@ -1036,11 +1186,11 @@ (.-history? db))))) (defn- ^DB with-pending - "Copy of `db` with a different pending overlay (nil to clear)." - [^DB db pending] + "Copy of `db` with different pending key and blob overlays (nil to clear)." + [^DB db pending pending-blobs] (DB. (.-schema db) (.-max-tx db) (.-rschema db) (.-pull-patterns db) (.-pull-attrs db) - (.-store db) pending + (.-store db) pending pending-blobs (.-as-of-tx db) (.-since-tx db) (.-history? db))) ;; ---------------------------------------------------------------------------- @@ -1054,6 +1204,7 @@ (cond (and (= :db/isComponent k) (true? v)) [:db/isComponent] (and (= :db/index k) (true? v)) [:db/index] + (and (= :dbval/deref k) (true? v)) [:dbval/deref] (= :db/tupleAttrs k) [:db.type/tuple :db/index] :else []))) @@ -1118,6 +1269,15 @@ (validate-schema-key a :db/valueType (:db/valueType kv) #{:db.type/ref :db.type/tuple}) (validate-schema-key a :db/cardinality (:db/cardinality kv) #{:db.cardinality/one :db.cardinality/many}) + ;; deref: value lives in the blob area, only its content hash is indexed + (validate-schema-key a :dbval/deref (:dbval/deref kv) #{true false}) + (when (and (:dbval/deref kv) + (or (:db/valueType kv) (:db/tupleAttrs kv))) + (util/raise "Bad attribute specification for " a ": {:dbval/deref true} cannot be combined with :db/valueType or :db/tupleAttrs" + {:error :schema/validation + :attribute a + :key :dbval/deref})) + ;; tuple should have tupleAttrs (when (and (= :db.type/tuple (:db/valueType kv)) (not (contains? kv :db/tupleAttrs))) @@ -1181,7 +1341,7 @@ (lru/cache 100) (lru/cache 100) store - nil nil nil nil)] + nil nil nil nil nil)] (with-max-tx db (q-max-tx db)))) (defrecord TxReport [db-before db-after tx-data tempids tx-meta]) @@ -1225,6 +1385,7 @@ (lru/cache 100) (.-store db) (.-pending db) + (.-pending-blobs db) (.-as-of-tx db) (.-since-tx db) (.-history? db))) @@ -1252,15 +1413,24 @@ (declare ref?) +(defn resolve-pattern-v + "Resolves the value component of a search pattern: entity ids for ref + attributes, BlobRefs for deref attributes (so both the scan range and the + `datom=` post-filter compare content hashes)." + [db a v] + (cond + (not (some? v)) v + (ref? db a) (entid-strict db v) + (deref-attr? db a) (value->blob-ref db v) + :else v)) + (defn resolve-datom [db e a v t default-e default-tx] (when (some? a) (validate-attr a (list 'resolve-datom 'db e a v t))) (datom (if (some? e) (entid-strict db e) default-e) a - (if (and (some? v) (ref? db a)) - (entid-strict db v) - v) + (resolve-pattern-v db a v) (if (some? t) (entid-strict db t) default-tx))) (defn components->pattern [db index c0 c1 c2 c3 default-e default-tx] @@ -1274,9 +1444,7 @@ (validate-attr a (list 'resolve-datom 'db e a v t))) [(when (some? e) (entid-strict db e)) a - (if (and (some? v) (ref? db a)) - (entid-strict db v) - v) + (resolve-pattern-v db a v) (when (some? t) (entid-strict db t))]) (defn components->pattern* [db index c0 c1 c2 c3] @@ -1316,6 +1484,12 @@ (defn tuple? [db attr] (is-attr? db attr :db.type/tuple)) +(defn deref-attr? + "True if `attr` is flagged with {:dbval/deref true}: its values are stored + content-addressed in the blob area and surface as BlobRefs." + [db attr] + (is-attr? db attr :dbval/deref)) + (defn tuple-source? [db attr] (is-attr? db attr :db/attrTuples)) @@ -1795,6 +1969,37 @@ (eduction (map (comp vec tuple-from-bytes)) (slice {:db db :begin begin :end end})))) +(defn- find-exact-datom + "Finds the current datom with exactly [e a v]. For deref attributes this + also finds legacy inline datoms (written before the attribute was flagged + as deref), which the hash-ranged search cannot see." + ^Datom [db e a v] + (or (fsearch db [e a v]) + (when (deref-attr? db a) + (some (fn [^Datom d] (when (= (.-v d) v) d)) + (-search db [e a]))))) + +(defn- stage-blob! + "Stages the blob behind `blob-ref` into the transaction's pending blob + overlay, so it is committed atomically with its datom's keys." + [db ^BlobRef blob-ref] + (let [^java.util.Map blobs (db-pending-blobs db) + ^bytes hash (.-hash blob-ref)] + (when (nil? blobs) + (util/raise "stage-blob! outside of a transaction" + {:error :transact/no-pending})) + (when-not (.containsKey blobs hash) + (if-some [^bytes bs (.-bytes blob-ref)] + (.put blobs hash bs) + ;; a BlobRef without bytes came from a query, so its blob normally + ;; already lives in this store; fetch it from the ref's origin db + ;; only when it does not (a value copied from another database) + (when (nil? (store/get-blob (db-store db) hash)) + (if-some [^bytes bs (db-get-blob (.-db blob-ref) hash)] + (.put blobs hash bs) + (util/raise "No blob found for deref value" + {:error :blob/not-found}))))))) + (defn with-datom [db ^Datom datom] (validate-datom db datom) (let [^java.util.NavigableSet pending (db-pending db) @@ -1803,12 +2008,18 @@ {: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)]) + (do + ;; legacy refs (inline-str) serialize back into the key itself and + ;; need no blob + (when (and (blob-ref? (.-v datom)) + (nil? (.-inline-str ^BlobRef (.-v datom)))) + (stage-blob! db (.-v 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-> (find-exact-datom db (.-e datom) (.-a datom) (.-v datom)) (retract-datom (:tx datom)))] (-> db (set-add! pending (datom-tuple db :eavt removing)) @@ -2005,11 +2216,14 @@ (let [tx (or tx (current-tx report)) db (:db-after report) e (entid-strict db e) - v (if (ref? db a) (entid-strict db v) v) + v (cond + (ref? db a) (entid-strict db v) + (deref-attr? db a) (value->blob-ref db v) + :else v) new-datom (datom e a v tx) multival? (multival? db a) old-datom ^Datom (if multival? - (fsearch db [e a v]) + (find-exact-datom db e a v) (fsearch db [e a]))] (cond (nil? old-datom) @@ -2180,7 +2394,11 @@ (let [[_ e a ov nv] entity e (entid-strict db e) _ (validate-attr a entity) - ov (if (ref? db a) (entid-strict db ov) ov) + ov (cond + (ref? db a) (entid-strict db ov) + (and (some? ov) + (deref-attr? db a)) (value->blob-ref db ov) + :else ov) nv (if (ref? db a) (entid-strict db nv) nv) _ (validate-val nv entity) datoms (vec (-search db [e a]))] @@ -2266,10 +2484,13 @@ (and (= op :db/retract) (some? v)) (if-some [e (entid db e)] - (let [v (if (ref? db a) (entid-strict db v) v)] + (let [v (cond + (ref? db a) (entid-strict db v) + (deref-attr? db a) (value->blob-ref db v) + :else v)] (validate-attr a entity) (validate-val v entity) - (if-some [old-datom (fsearch db [e a v])] + (if-some [old-datom (find-exact-datom db e a v)] (recur (transact-retract-datom report old-datom) entities) (recur report entities))) (recur report entities)) @@ -2329,6 +2550,11 @@ ;; 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) + ;; The blobs of deref values staged by this transaction (content hash + ;; -> value bytes); committed atomically with the pending keys and + ;; overlaid over the store's blob reads in the meantime (see + ;; `db-get-blob`). + pending-blobs (java.util.TreeMap. ^java.util.Comparator store/byte-array-comparator) ;; Carry over speculative datoms when chaining a dry-run on a dry-run ;; db-after, so they stay visible in the new speculative view. A real ;; transact must not inherit them: they were never committed and would @@ -2338,12 +2564,15 @@ (.addAll ^java.util.TreeSet pending prev) (util/raise "Cannot transact against a speculative (dry-run) database value" {:error :transact/speculative-view}))) + _ (when dry-run? + (when-some [^java.util.Map prev (db-pending-blobs (:db-after report))] + (.putAll pending-blobs prev))) 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-pending pending)) + (update :db-after with-pending pending pending-blobs)) {: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) @@ -2355,8 +2584,8 @@ (dissoc ::dry-run) (assoc :tx tx-id)) (do - (store/commit! (db-store (:db-after result)) (seq pending)) + (store/commit! (db-store (:db-after result)) (seq pending) pending-blobs) (-> result - (update :db-after with-pending nil) + (update :db-after with-pending nil nil) ;; Add :tx field with the transaction UUID (assoc :tx tx-id)))))) diff --git a/src/dbval/store.clj b/src/dbval/store.clj index ba77e87..ce1ba85 100644 --- a/src/dbval/store.clj +++ b/src/dbval/store.clj @@ -3,17 +3,19 @@ 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. + atomic batch commits. Conceptually the store is a sorted set, mimicking a + transactional ordered key-value store like FoundationDB. + + Datoms of deref attributes (see `dbval.db`) keep only a content hash in + their keys; the value bytes live in a separate content-addressed blob + area: `-commit!` takes the batch's blobs alongside its keys and + `-get-blob` reads one back by hash. Blobs are immutable — the same hash + always maps to the same bytes, so re-writing an existing blob is a no-op. 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) + transaction's pending keys and blobs over `-scan`/`-get-blob`. A store + implementation therefore only has to provide committed data. Implementations: `dbval.store.sqlite` (default), `dbval.store.memory`.") @@ -22,10 +24,14 @@ "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.") + (-commit! [store keys blobs] + "Atomically adds the byte[] `keys` and the `blobs` (a java.util.Map of + byte[] content hash -> byte[] value) to the store: after `-commit!` + returns, either all keys and blobs are durably visible or — if it + throws — none are. Keys and blobs that already exist are ignored.") + (-get-blob [store hash] + "Returns the committed byte[] blob stored under the byte[] content + `hash`, or nil if there is none.") (-close! [store] "Releases the store's resources.")) @@ -36,8 +42,13 @@ (defn commit! "See [[ITupleStore]]." - [store keys] - (-commit! store keys)) + [store keys blobs] + (-commit! store keys blobs)) + +(defn get-blob + "See [[ITupleStore]]." + [store hash] + (-get-blob store hash)) (defn close! "See [[ITupleStore]]." diff --git a/src/dbval/store/memory.clj b/src/dbval/store/memory.clj index d33eafa..90420df 100644 --- a/src/dbval/store/memory.clj +++ b/src/dbval/store/memory.clj @@ -9,9 +9,10 @@ (:require [dbval.store :as store]) (:import - [java.util.concurrent ConcurrentSkipListSet])) + [java.util.concurrent ConcurrentSkipListMap ConcurrentSkipListSet])) -(deftype MemoryStore [^ConcurrentSkipListSet keyset] +(deftype MemoryStore [^ConcurrentSkipListSet keyset + ^ConcurrentSkipListMap blobs] store/ITupleStore (-scan [_ begin end reverse?] (let [sub (.subSet keyset begin true end false)] @@ -19,18 +20,24 @@ (.descendingSet ^java.util.NavigableSet sub) sub))) - (-commit! [this keys] + (-commit! [this keys new-blobs] ;; 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 h ^bytes v] new-blobs] + (.putIfAbsent blobs h v)) (doseq [^bytes k keys] (.add keyset k)))) + (-get-blob [_ hash] + (.get blobs hash)) + (-close! [_] nil)) (defn store "Creates an empty in-memory tuple store." ^dbval.store.memory.MemoryStore [] - (MemoryStore. (ConcurrentSkipListSet. ^java.util.Comparator store/byte-array-comparator))) + (MemoryStore. (ConcurrentSkipListSet. ^java.util.Comparator store/byte-array-comparator) + (ConcurrentSkipListMap. ^java.util.Comparator store/byte-array-comparator))) diff --git a/src/dbval/store/sqlite.clj b/src/dbval/store/sqlite.clj index 14a21c8..a384284 100644 --- a/src/dbval/store/sqlite.clj +++ b/src/dbval/store/sqlite.clj @@ -1,11 +1,14 @@ (ns dbval.store.sqlite - "SQLite-backed tuple store: one table holding the sorted keys. + "SQLite-backed tuple store: one table holding the sorted keys and one + holding the content-addressed blobs of deref attributes. create table dbval (k blob not null, primary key(k)) WITHOUT ROWID; + create table dbval_blob (h blob not null, v blob not null, + primary key(h)) 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." + snapshot); `-commit!` wraps its batch inserts in a single transaction." (:require [clojure.java.io :as io] [clojure.string :as str] @@ -54,7 +57,9 @@ (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;"))) + "create table if not exists dbval (k blob not null, primary key(k)) WITHOUT ROWID;") + (.execute ^java.sql.Statement stmt + "create table if not exists dbval_blob (h blob not null, v blob not null, primary key(h)) WITHOUT ROWID;"))) (defn- scan-iterator ^java.util.Iterator [^java.sql.Connection conn ^bytes begin ^bytes end reverse?] @@ -106,16 +111,24 @@ (iterator [_] (scan-iterator conn begin end (boolean reverse?))))) - (-commit! [this keys] - (when (seq keys) + (-commit! [this keys blobs] + (when (or (seq keys) (seq blobs)) (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)) + (when (seq blobs) + (with-open [stmt (.prepareStatement conn "INSERT OR IGNORE INTO dbval_blob (h, v) VALUES (?, ?)")] + (doseq [[^bytes h ^bytes v] blobs] + (.setBytes stmt 1 h) + (.setBytes stmt 2 v) + (.addBatch stmt)) + (.executeBatch stmt))) + (when (seq keys) + (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 _)) @@ -123,6 +136,13 @@ (finally (.setAutoCommit conn true)))))) + (-get-blob [_ hash] + (with-open [stmt (.prepareStatement conn "select v from dbval_blob where h = ?")] + (.setBytes stmt 1 ^bytes hash) + (with-open [rs (.executeQuery stmt)] + (when (.next rs) + (.getBytes rs "v"))))) + (-close! [_] (.close conn))) diff --git a/store-slatedb/src/dbval/store/slatedb.clj b/store-slatedb/src/dbval/store/slatedb.clj index 015c06b..74e61f3 100644 --- a/store-slatedb/src/dbval/store/slatedb.clj +++ b/store-slatedb/src/dbval/store/slatedb.clj @@ -2,9 +2,13 @@ "SlateDB-backed tuple store: an embedded ordered key-value store built on object storage (S3, GCS, or the local filesystem during development). - dbval only needs the key portion, so every key is stored with an empty - value. `-commit!` writes the batch through a SlateDB WriteBatch, which is - atomic; scans use SlateDB's native ascending/descending iteration. + Datom keys are stored with an empty value. Blobs of deref attributes are + stored under keys with the tuple prefix (\"blob\", ) and carry the + value bytes in the SlateDB value (keys are capped at 64 KiB, values at + 4 GiB). Datom scans are always prefixed with an index name (\"eavt\" etc.), + so the blob prefix never overlaps them. `-commit!` writes keys and blobs + through one SlateDB WriteBatch, which is atomic; scans use SlateDB's + native ascending/descending iteration. This namespace lives in the store-slatedb module because io.slatedb/slatedb-uniffi is a heavy native dependency and this namespace @@ -23,6 +27,12 @@ "Empty byte array used as value for key-only puts in SlateDB." (byte-array 0)) +(defn- blob-key + "SlateDB key for the blob with the given content hash." + ^bytes [^bytes hash] + (.pack (com.apple.foundationdb.tuple.Tuple/from + (into-array Object ["blob" hash])))) + (defonce ^:private native-lib-loaded ;; The slatedb-uniffi jar bundles the native library in JNA resource layout ;; (e.g. linux-x86-64/libslatedb_uniffi.so), but its generated loader only @@ -89,11 +99,13 @@ (iterator [_] (scan-iterator db begin end (boolean reverse?))))) - (-commit! [this keys] - (when (seq keys) + (-commit! [this keys blobs] + (when (or (seq keys) (seq blobs)) (locking this (let [batch (WriteBatch.)] (try + (doseq [[^bytes h ^bytes v] blobs] + (.put batch (blob-key h) v)) (doseq [^bytes k keys] (.put batch k EMPTY_VALUE)) (await-future (.write db batch)) @@ -102,6 +114,9 @@ ;; on error the batch is simply discarded (try (.close batch) (catch Throwable _)))))))) + (-get-blob [_ hash] + (await-future (.get db (blob-key hash)))) + (-close! [_] (.close db))) diff --git a/store-slatedb/test/dbval/store/slatedb_test.clj b/store-slatedb/test/dbval/store/slatedb_test.clj index b725cb7..77b13c1 100644 --- a/store-slatedb/test/dbval/store/slatedb_test.clj +++ b/store-slatedb/test/dbval/store/slatedb_test.clj @@ -51,6 +51,31 @@ (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-slatedb-store-deref-values + ;; SlateDB caps keys at 64 KiB; deref attributes keep only a content hash + ;; in the keys and put the value bytes into the SlateDB value of a + ;; ("blob", ) key — so values far beyond the key cap round-trip + (let [conn (d/conn-from-db + (empty-slatedb-db {:doc/name {:db/unique :db.unique/identity} + :doc/model {:dbval/deref true}})) + ;; ~300 KiB of EDN, far over the 64 KiB key cap + model {:objects (mapv (fn [i] {:id i :content (apply str (repeat 100 "x"))}) + (range 2500))}] + (d/transact! conn [{:doc/name "big" :doc/model model}]) + (let [v (:doc/model (d/entity @conn [:doc/name "big"]))] + (is (not (realized? v))) + (is (= model @v))) + (testing "re-assert is a no-op" + (is (empty? (:tx-data (d/transact! conn [{:doc/name "big" :doc/model model}]))))) + (testing "blob keys do not leak into index scans" + (is (= #{:doc/name :doc/model} + (into #{} (map :a) (d/datoms @conn :eavt))))) + (testing "update and retract" + (d/transact! conn [{:doc/name "big" :doc/model (assoc model :v 2)}]) + (is (= (assoc model :v 2) @(:doc/model (d/entity @conn [:doc/name "big"])))) + (d/transact! conn [[:db/retract [:doc/name "big"] :doc/model (assoc model :v 2)]]) + (is (nil? (:doc/model (d/entity @conn [:doc/name "big"]))))))) + (deftest test-slatedb-store-transaction-isolation ;; a failing transaction must leave the store untouched: nothing is ;; written until the pending overlay commits atomically diff --git a/test/dbval/test/deref_value.clj b/test/dbval/test/deref_value.clj new file mode 100644 index 0000000..1546714 --- /dev/null +++ b/test/dbval/test/deref_value.clj @@ -0,0 +1,216 @@ +(ns dbval.test.deref-value + "Tests for deref value types: attributes flagged with {:dbval/deref true} + store only a content hash in the index keys, the value bytes live in the + store's blob area and reads surface BlobRefs (see `dbval.db/BlobRef`)." + (:require + [clojure.test :as t :refer [is deftest testing]] + [dbval.core :as d] + [dbval.db :as db] + [dbval.store :as store] + [dbval.store.memory :as memory])) + +(def schema + {:doc/name {:db/unique :db.unique/identity} + :doc/model {:dbval/deref true} + :doc/tags {:dbval/deref true + :db/cardinality :db.cardinality/many}}) + +(defn- conn [] + (d/conn-from-db (d/empty-db schema {:store (memory/store)}))) + +(def model-v1 + {:objects (mapv (fn [i] {:id i :content (apply str (repeat 50 "x"))}) + (range 100))}) + +(def model-v2 + (assoc model-v1 :version 2)) + +(deftest test-roundtrip + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (let [v (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn)] + (is (db/blob-ref? v)) + (is (not (realized? v))) + (is (= model-v1 @v)) + (is (realized? v))) + (testing "entity api" + (is (= model-v1 @(:doc/model (d/entity @conn [:doc/name "a"]))))) + (testing "pull api" + (is (= model-v1 @(:doc/model (d/pull @conn [:doc/model] [:doc/name "a"]))))) + (testing "printing never fetches" + (let [v (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn)] + (is (re-matches #"#dbval/blob-ref \"[0-9a-f]{64}\"" (pr-str v))) + (is (not (realized? v))))))) + +(deftest test-no-op-and-update + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (testing "re-asserting the same value is a no-op" + (let [report (d/transact! conn [{:doc/name "a" :doc/model model-v1}])] + (is (empty? (:tx-data report))))) + (testing "a changed value retracts the old datom and asserts the new one" + (let [report (d/transact! conn [{:doc/name "a" :doc/model model-v2}]) + model-datoms (filter #(= :doc/model (:a %)) (:tx-data report))] + (is (= [false true] (mapv :added model-datoms))) + (is (= model-v2 @(d/q '[:find ?v . :where [_ :doc/model ?v]] @conn))))) + (testing "history keeps both versions" + (is (= #{[model-v1 true] [model-v1 false] [model-v2 true]} + (into #{} + (map (fn [datom] [@(:v datom) (:added datom)])) + (filter #(= :doc/model (:a %)) + (d/datoms (d/history @conn) :eavt)))))))) + +(deftest test-retract + (testing "retract by plain value" + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (d/transact! conn [[:db/retract [:doc/name "a"] :doc/model model-v1]]) + (is (nil? (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn))))) + (testing "retract by BlobRef" + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (let [v (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn)] + (d/transact! conn [[:db/retract [:doc/name "a"] :doc/model v]]) + (is (not (realized? v))) + (is (nil? (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn)))))) + (testing "retract-entity retracts deref datoms" + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (d/transact! conn [[:db/retractEntity [:doc/name "a"]]]) + (is (nil? (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn)))))) + +(deftest test-copy-without-fetch + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (let [v (d/q '[:find ?v . :where [_ :doc/model ?v]] @conn)] + (d/transact! conn [{:doc/name "b" :doc/model v}]) + (is (not (realized? v))) + (is (= model-v1 @(:doc/model (d/entity @conn [:doc/name "b"]))))))) + +(deftest test-equality-join + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1} + {:doc/name "b" :doc/model model-v1} + {:doc/name "c" :doc/model model-v2}]) + (is (= #{["a" "b"] ["b" "a"]} + (d/q '[:find ?n1 ?n2 + :where + [?e1 :doc/model ?v] + [?e2 :doc/model ?v] + [(not= ?e1 ?e2)] + [?e1 :doc/name ?n1] + [?e2 :doc/name ?n2]] + @conn))))) + +(deftest test-unique-upsert + (let [schema {:doc/id {:db/unique :db.unique/identity + :dbval/deref true} + :doc/count {}} + conn (d/conn-from-db (d/empty-db schema {:store (memory/store)}))] + (d/transact! conn [{:doc/id model-v1 :doc/count 1}]) + (testing "upsert by deref identity value resolves to the same entity" + (d/transact! conn [{:doc/id model-v1 :doc/count 2}]) + (is (= [2] (d/q '[:find [?c ...] :where [_ :doc/count ?c]] @conn)))) + (testing "different value creates a new entity" + (d/transact! conn [{:doc/id model-v2 :doc/count 3}]) + (is (= #{2 3} (set (d/q '[:find [?c ...] :where [_ :doc/count ?c]] @conn))))) + (testing "lookup ref by deref value" + (is (= 2 (:doc/count (d/entity @conn [:doc/id model-v1]))))))) + +(deftest test-cardinality-many + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/tags [model-v1 model-v2]}]) + (is (= #{model-v1 model-v2} + (into #{} (map deref) (d/q '[:find [?v ...] :where [_ :doc/tags ?v]] @conn)))) + (testing "re-asserting an existing value is a no-op" + (let [report (d/transact! conn [[:db/add [:doc/name "a"] :doc/tags model-v1]])] + (is (empty? (:tx-data report))))) + (testing "retracting one value keeps the other" + (d/transact! conn [[:db/retract [:doc/name "a"] :doc/tags model-v1]]) + (is (= #{model-v2} + (into #{} (map deref) (d/q '[:find [?v ...] :where [_ :doc/tags ?v]] @conn))))))) + +(deftest test-cas + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (let [e (:db/id (d/entity @conn [:doc/name "a"]))] + (d/transact! conn [[:db/cas e :doc/model model-v1 model-v2]]) + (is (= model-v2 @(:doc/model (d/entity @conn [:doc/name "a"])))) + (is (thrown? clojure.lang.ExceptionInfo + (d/transact! conn [[:db/cas e :doc/model model-v1 model-v2]])))))) + +(deftest test-tx-fn-read-your-writes + (let [conn (conn) + tx-fn (fn [db] + ;; derefs a value asserted earlier in the same transaction: + ;; must be served from the pending blob overlay + (let [v (:doc/model (d/entity db [:doc/name "a"]))] + [{:doc/name "copy" :doc/model (assoc @v :copied true)}]))] + (d/transact! conn [{:doc/name "a" :doc/model model-v1} + [:db.fn/call tx-fn]]) + (is (= (assoc model-v1 :copied true) + @(:doc/model (d/entity @conn [:doc/name "copy"])))))) + +(deftest test-dry-run + (let [conn (conn)] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (let [report (d/with-dry-run @conn [{:doc/name "b" :doc/model model-v2}]) + report' (d/with-dry-run (:db-after report) [{:doc/name "c" :doc/model model-v1}])] + (testing "speculative deref values are readable without a commit" + (is (= model-v2 @(:doc/model (d/entity (:db-after report) [:doc/name "b"])))) + (is (= model-v1 @(:doc/model (d/entity (:db-after report') [:doc/name "c"]))))) + (testing "nothing was committed" + (is (nil? (d/entity @conn [:doc/name "b"]))))))) + +(deftest test-legacy-inline-datoms + ;; datoms written before their attribute was flagged as deref keep the + ;; serialized value inline in the index keys; flipping the schema flag must + ;; not require a data migration + (let [store (memory/store) + inline (d/conn-from-db (d/empty-db {:doc/name {:db/unique :db.unique/identity}} + {:store store})) + _ (d/transact! inline [{:doc/name "a" :doc/model model-v1}]) + conn (d/conn-from-db (d/empty-db schema {:store store}))] + (testing "legacy datom reads as a realized-on-demand BlobRef" + (let [v (:doc/model (d/entity @conn [:doc/name "a"]))] + (is (db/blob-ref? v)) + (is (= model-v1 @v)))) + (testing "re-asserting the same value is a no-op across representations" + (let [report (d/transact! conn [{:doc/name "a" :doc/model model-v1}])] + (is (empty? (:tx-data report))))) + (testing "updating a legacy datom retracts it" + (d/transact! conn [{:doc/name "a" :doc/model model-v2}]) + (is (= model-v2 @(:doc/model (d/entity @conn [:doc/name "a"])))) + (is (= 1 (count (vec (d/datoms @conn :aevt :doc/model)))))) + (testing "retract by value finds a legacy datom" + (let [store (memory/store) + inline (d/conn-from-db (d/empty-db {:doc/name {:db/unique :db.unique/identity}} + {:store store})) + _ (d/transact! inline [{:doc/name "b" :doc/model model-v1}]) + conn (d/conn-from-db (d/empty-db schema {:store store}))] + (d/transact! conn [[:db/retract [:doc/name "b"] :doc/model model-v1]]) + (is (nil? (:doc/model (d/entity @conn [:doc/name "b"])))))))) + +(deftest test-oversize-guard + (let [conn (conn) + huge (apply str (repeat 70000 "x"))] + (testing "a large value on a non-deref attribute throws a descriptive error" + (is (thrown? clojure.lang.ExceptionInfo + (d/transact! conn [{:doc/name "a" :doc/plain huge}]))) + (try + (d/transact! conn [{:doc/name "a" :doc/plain huge}]) + (catch clojure.lang.ExceptionInfo e + (is (= :transact/value-too-large (:error (ex-data e))))))) + (testing "the same value on a deref attribute works" + (d/transact! conn [{:doc/name "a" :doc/model huge}]) + (is (= huge @(:doc/model (d/entity @conn [:doc/name "a"]))))))) + +(deftest test-sqlite-persistence + (let [db-file (str (System/getProperty "java.io.tmpdir") + "/dbval-deref-test-" (random-uuid) ".db")] + (let [conn (d/conn-from-db (d/empty-db schema {:db-file db-file}))] + (d/transact! conn [{:doc/name "a" :doc/model model-v1}]) + (store/close! (db/db-store @conn))) + (let [conn (d/conn-from-db (d/empty-db schema {:db-file db-file}))] + (is (= model-v1 @(:doc/model (d/entity @conn [:doc/name "a"])))) + (store/close! (db/db-store @conn)))))