Implement transactions.
This commit is contained in:
Nick Alexander
parent
0c51cb6236
commit
baec3815b0
136
src/datomish/datom.cljc
Normal file
136
src/datomish/datom.cljc
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
|
||||
;; Purloined from DataScript.
|
||||
|
||||
(ns datomish.datom)
|
||||
|
||||
(declare hash-datom equiv-datom seq-datom val-at-datom nth-datom assoc-datom)
|
||||
|
||||
(deftype Datom [e a v tx added]
|
||||
#?@(:cljs
|
||||
[IHash
|
||||
(-hash [d] (or (.-__hash d)
|
||||
(set! (.-__hash d) (hash-datom d))))
|
||||
IEquiv
|
||||
(-equiv [d o] (and (instance? Datom o) (equiv-datom d o)))
|
||||
|
||||
ISeqable
|
||||
(-seq [d] (seq-datom d))
|
||||
|
||||
ILookup
|
||||
(-lookup [d k] (val-at-datom d k nil))
|
||||
(-lookup [d k nf] (val-at-datom d k nf))
|
||||
|
||||
IIndexed
|
||||
(-nth [this i] (nth-datom this i))
|
||||
(-nth [this i not-found] (nth-datom this i not-found))
|
||||
|
||||
IAssociative
|
||||
(-assoc [d k v] (assoc-datom d k v))
|
||||
|
||||
IPrintWithWriter
|
||||
(-pr-writer [d writer opts]
|
||||
(pr-sequential-writer writer pr-writer
|
||||
"#datascript/Datom [" " " "]"
|
||||
opts [(.-e d) (.-a d) (.-v d) (.-tx d) (.-added d)]))
|
||||
]
|
||||
:clj
|
||||
[Object
|
||||
(hashCode [d] (hash-datom d))
|
||||
|
||||
clojure.lang.IHashEq
|
||||
(hasheq [d] (hash-datom d))
|
||||
|
||||
clojure.lang.Seqable
|
||||
(seq [d] (seq-datom d))
|
||||
|
||||
clojure.lang.IPersistentCollection
|
||||
(equiv [d o] (and (instance? Datom o) (equiv-datom d o)))
|
||||
(empty [d] (throw (UnsupportedOperationException. "empty is not supported on Datom")))
|
||||
(count [d] 5)
|
||||
(cons [d [k v]] (assoc-datom d k v))
|
||||
|
||||
clojure.lang.Indexed
|
||||
(nth [this i] (nth-datom this i))
|
||||
(nth [this i not-found] (nth-datom this i not-found))
|
||||
|
||||
clojure.lang.ILookup
|
||||
(valAt [d k] (val-at-datom d k nil))
|
||||
(valAt [d k nf] (val-at-datom d k nf))
|
||||
|
||||
clojure.lang.Associative
|
||||
(entryAt [d k] (some->> (val-at-datom d k nil) (clojure.lang.MapEntry k)))
|
||||
(containsKey [e k] (#{:e :a :v :tx :added} k))
|
||||
(assoc [d k v] (assoc-datom d k v))
|
||||
]))
|
||||
|
||||
(defn ^Datom datom
|
||||
([e a v tx] (Datom. e a v tx true))
|
||||
([e a v tx added] (Datom. e a v tx added)))
|
||||
|
||||
(defn datom? [x] (instance? Datom x))
|
||||
|
||||
(defn- hash-datom [^Datom d]
|
||||
(-> (hash (.-e d))
|
||||
(hash-combine (hash (.-a d)))
|
||||
(hash-combine (hash (.-v d)))))
|
||||
|
||||
(defn- equiv-datom [^Datom d ^Datom o]
|
||||
(and (= (.-e d) (.-e o))
|
||||
(= (.-a d) (.-a o))
|
||||
(= (.-v d) (.-v o))))
|
||||
|
||||
(defn- seq-datom [^Datom d]
|
||||
(list (.-e d) (.-a d) (.-v d) (.-tx d) (.-added d)))
|
||||
|
||||
;; keep it fast by duplicating for both keyword and string cases
|
||||
;; instead of using sets or some other matching func
|
||||
(defn- val-at-datom [^Datom d k not-found]
|
||||
(case k
|
||||
:e (.-e d) "e" (.-e d)
|
||||
:a (.-a d) "a" (.-a d)
|
||||
:v (.-v d) "v" (.-v d)
|
||||
:tx (.-tx d) "tx" (.-tx d)
|
||||
:added (.-added d) "added" (.-added d)
|
||||
not-found))
|
||||
|
||||
(defn- nth-datom
|
||||
([^Datom d ^long i]
|
||||
(case i
|
||||
0 (.-e d)
|
||||
1 (.-a d)
|
||||
2 (.-v d)
|
||||
3 (.-tx d)
|
||||
4 (.-added d)
|
||||
#?(:clj (throw (IndexOutOfBoundsException.))
|
||||
:cljs (throw (js/Error. (str "Datom/-nth: Index out of bounds: " i))))))
|
||||
([^Datom d ^long i not-found]
|
||||
(case i
|
||||
0 (.-e d)
|
||||
1 (.-a d)
|
||||
2 (.-v d)
|
||||
3 (.-tx d)
|
||||
4 (.-added d)
|
||||
not-found)))
|
||||
|
||||
(defn- ^Datom assoc-datom [^Datom d k v]
|
||||
(case k
|
||||
:e (Datom. v (.-a d) (.-v d) (.-tx d) (.-added d))
|
||||
:a (Datom. (.-e d) v (.-v d) (.-tx d) (.-added d))
|
||||
:v (Datom. (.-e d) (.-a d) v (.-tx d) (.-added d))
|
||||
:tx (Datom. (.-e d) (.-a d) (.-v d) v (.-added d))
|
||||
:added (Datom. (.-e d) (.-a d) (.-v d) (.-tx d) v)
|
||||
(throw (IllegalArgumentException. (str "invalid key for #datascript/Datom: " k)))))
|
||||
|
||||
;; printing and reading
|
||||
|
||||
(defn ^Datom datom-from-reader [vec]
|
||||
(apply datom vec))
|
||||
|
||||
#?(:clj
|
||||
(defmethod print-method Datom [^Datom d, ^java.io.Writer w]
|
||||
(.write w (str "#datascript/Datom "))
|
||||
(binding [*out* w]
|
||||
(pr [(.-e d) (.-a d) (.-v d) (.-tx d) (.-added d)]))))
|
||||
|
|
@ -12,31 +12,308 @@
|
|||
[datomish.query.projection :as projection]
|
||||
[datomish.query.source :as source]
|
||||
[datomish.query :as query]
|
||||
[honeysql.core :as sql]
|
||||
[datomish.datom :as dd :refer [datom datom? #?@(:cljs [Datom])]]
|
||||
[datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise raise-str cond-let]]
|
||||
[datomish.schema :as ds]
|
||||
[datomish.sqlite :as s]
|
||||
[datomish.sqlite-schema :as sqlite-schema]
|
||||
[datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise raise-str]]
|
||||
#?@(:clj [[datomish.pair-chan :refer [go-pair <?]]
|
||||
[clojure.core.async :as a :refer [chan go <! >!]]])
|
||||
#?@(:cljs [[datomish.pair-chan]
|
||||
[cljs.core.async :as a :refer [chan <! >!]]])))
|
||||
[cljs.core.async :as a :refer [chan <! >!]]]))
|
||||
#?(:clj
|
||||
(:import
|
||||
[datomish.datom Datom])))
|
||||
|
||||
#?(:clj
|
||||
;; From https://stuartsierra.com/2015/05/27/clojure-uncaught-exceptions
|
||||
;; Assuming require [clojure.tools.logging :as log]
|
||||
(Thread/setDefaultUncaughtExceptionHandler
|
||||
(reify Thread$UncaughtExceptionHandler
|
||||
(uncaughtException [_ thread ex]
|
||||
(println ex "Uncaught exception on" (.getName thread))))))
|
||||
|
||||
(defprotocol IDB
|
||||
(query-context
|
||||
[db])
|
||||
(close
|
||||
[db]
|
||||
"Close this database. Returns a pair channel of [nil error]."))
|
||||
|
||||
(defrecord DB [sqlite-connection]
|
||||
(close-db
|
||||
[db]
|
||||
"Close this database. Returns a pair channel of [nil error].")
|
||||
|
||||
(schema
|
||||
[db]
|
||||
"Return the schema of this database.")
|
||||
|
||||
(idents
|
||||
[db]
|
||||
"Return the known idents of this database, as a map from keyword idents to entids.")
|
||||
|
||||
(current-tx
|
||||
[db]
|
||||
"TODO: document this interface.")
|
||||
|
||||
(<eavt
|
||||
[db pattern]
|
||||
"Search for datoms using the EAVT index.")
|
||||
|
||||
(<avet
|
||||
[db pattern]
|
||||
"Search for datoms using the AVET index.")
|
||||
|
||||
(<apply-datoms
|
||||
[db datoms]
|
||||
"Apply datoms to the store.")
|
||||
|
||||
(<advance-tx
|
||||
[db]
|
||||
"TODO: document this interface."))
|
||||
|
||||
;; TODO: handle _?
|
||||
(defn search->sql-clause [pattern]
|
||||
(merge
|
||||
{:select [:*] ;; e :a :v :tx] ;; TODO: generalize columns.
|
||||
:from [:datoms]}
|
||||
(if-not (empty? pattern)
|
||||
{:where (cons :and (map #(vector := %1 (if (keyword? %2) (str %2) %2)) [:e :a :v :tx] pattern))} ;; TODO: use schema to process v.
|
||||
{})))
|
||||
|
||||
(defrecord DB [sqlite-connection schema idents current-tx]
|
||||
;; idents is map {ident -> entid} of known idents. See http://docs.datomic.com/identity.html#idents.
|
||||
IDB
|
||||
(query-context [db] (context/->Context (source/datoms-source db) nil nil))
|
||||
(close [db] (s/close (.-sqlite-connection db))))
|
||||
|
||||
(defn <with-sqlite-connection [sqlite-connection]
|
||||
(schema [db] (.-schema db))
|
||||
|
||||
(idents [db] (.-idents db))
|
||||
|
||||
(current-tx
|
||||
[db]
|
||||
(inc (:current-tx db)))
|
||||
|
||||
;; TODO: use q for searching? Have q use this for searching for a single pattern?
|
||||
(<eavt [db pattern]
|
||||
(go-pair
|
||||
;; TODO: find a better expression of this pattern.
|
||||
(let [rows (<? (->>
|
||||
(search->sql-clause pattern)
|
||||
(sql/format)
|
||||
(s/all-rows (:sqlite-connection db))))]
|
||||
(mapv #(Datom. (:e %) (:a %) (:v %) (:tx %) true) rows)))) ;; TODO: map values according to schema.
|
||||
|
||||
(<avet [db pattern]
|
||||
(go-pair
|
||||
;; TODO: find a better expression of this pattern.
|
||||
(let [[a v] pattern
|
||||
rows (<? (->>
|
||||
{:select [:*] :from [:datoms] :where [:and [:= :a a] [:= :v v]]}
|
||||
(sql/format)
|
||||
(s/all-rows (:sqlite-connection db))))]
|
||||
(mapv #(Datom. (:e %) (:a %) (:v %) (:tx %) true) rows)))) ;; TODO: map values according to schema.
|
||||
|
||||
(<apply-datoms [db datoms]
|
||||
(go-pair
|
||||
(let [exec (partial s/execute! (:sqlite-connection db))]
|
||||
;; TODO: batch insert, batch delete.
|
||||
(doseq [datom datoms]
|
||||
(let [[e a v tx added] datom]
|
||||
;; Append to transaction log.
|
||||
(<? (exec
|
||||
["INSERT INTO transactions VALUES (?, ?, ?, ?, ?)" e a v tx added]))
|
||||
;; Update materialized datom view.
|
||||
(if (.-added datom)
|
||||
(<? (exec
|
||||
;; TODO: use schema to insert correct indexing flags.
|
||||
["INSERT INTO datoms VALUES (?, ?, ?, ?, 0, 0)" e a v tx]))
|
||||
(<? (exec
|
||||
;; TODO: verify this is correct.
|
||||
["DELETE FROM datoms WHERE (e = ? AND a = ? AND v = ?)" e a v]))))))
|
||||
db))
|
||||
|
||||
(<advance-tx [db]
|
||||
(go-pair
|
||||
(let [exec (partial s/execute! (:sqlite-connection db))]
|
||||
;; (let [ret (<? (exec
|
||||
;; ;; TODO: be more clever about UPDATE OR ...?
|
||||
;; ["UPDATE metadata SET current_tx = ? WHERE current_tx = ?" (inc (:current-tx db)) (:current-tx db)]))]
|
||||
|
||||
;; TODO: handle exclusion across transactions here.
|
||||
(update db :current-tx inc))))
|
||||
;; )
|
||||
|
||||
(close-db [db] (s/close (.-sqlite-connection db))))
|
||||
|
||||
(defn db? [x]
|
||||
(and (satisfies? IDB x)))
|
||||
|
||||
(defprotocol IConnection
|
||||
(close
|
||||
[conn]
|
||||
"Close this connection. Returns a pair channel of [nil error].")
|
||||
|
||||
(db
|
||||
[conn]
|
||||
"Get the current DB associated with this connection.")
|
||||
|
||||
(history
|
||||
[conn]
|
||||
"Get the full transaction history DB associated with this connection."))
|
||||
|
||||
(defrecord Connection [current-db]
|
||||
IConnection
|
||||
(close [conn] (close-db @(:current-db conn)))
|
||||
|
||||
(db [conn] @(:current-db conn))
|
||||
|
||||
(history [conn]
|
||||
(raise "Datomic's history is not yet supported." {})))
|
||||
|
||||
(defn conn? [x]
|
||||
(and (satisfies? IConnection x)))
|
||||
|
||||
;; ----------------------------------------------------------------------------
|
||||
;; define data-readers to be made available to EDN readers. in CLJS
|
||||
;; they're magically available. in CLJ, data_readers.clj may or may
|
||||
;; not work, but you can always simply do
|
||||
;;
|
||||
;; (clojure.edn/read-string {:readers datomish/data-readers} "...")
|
||||
;;
|
||||
|
||||
(defonce -id-literal-idx (atom -1000000))
|
||||
|
||||
(defrecord TempId [part idx])
|
||||
|
||||
(defn id-literal
|
||||
([part]
|
||||
(if (sequential? part)
|
||||
(apply id-literal part)
|
||||
(->TempId part (swap! -id-literal-idx dec))))
|
||||
([part idx]
|
||||
(->TempId part idx)))
|
||||
|
||||
(defn id-literal? [x]
|
||||
(and (instance? TempId x)))
|
||||
|
||||
(defn temp-literal? [x]
|
||||
(and (id-literal? x)
|
||||
(= :db.part/temp (:part x))))
|
||||
|
||||
;; (def data-readers {'db/id id-literal})
|
||||
|
||||
;; #?(:cljs
|
||||
;; (doseq [[tag cb] data-readers] (cljs.reader/register-tag-parser! tag cb)))
|
||||
|
||||
;; TODO: implement support for DB parts?
|
||||
(def tx0 0x2000000)
|
||||
|
||||
(defn <idents [sqlite-connection]
|
||||
(go-pair
|
||||
(when-not (= sqlite-schema/current-version (<? (sqlite-schema/<ensure-current-version sqlite-connection)))
|
||||
(raise-str "Could not ensure current SQLite schema version."))
|
||||
(->DB sqlite-connection)))
|
||||
(let [rows (<? (->>
|
||||
{:select [:e :v] :from [:datoms] :where [:= :a ":db/ident"]} ;; TODO: don't stringify?
|
||||
(sql/format)
|
||||
(s/all-rows sqlite-connection)))]
|
||||
(into {} (map #(-> {(keyword (:v %)) (:e %)})) rows))))
|
||||
|
||||
(defn <db-with-sqlite-connection
|
||||
([sqlite-connection]
|
||||
(<db-with-sqlite-connection sqlite-connection {}))
|
||||
|
||||
([sqlite-connection schema]
|
||||
(go-pair
|
||||
(when-not (= sqlite-schema/current-version (<? (sqlite-schema/<ensure-current-version sqlite-connection)))
|
||||
(raise "Could not ensure current SQLite schema version."))
|
||||
(let [idents (into (<? (<idents sqlite-connection)) {:db/txInstant 100 :db/ident 101 :x 102 :y 103})] ;; TODO: pre-populate idents and SQLite tables?
|
||||
(map->DB
|
||||
{:sqlite-connection sqlite-connection
|
||||
:idents idents
|
||||
:schema (ds/schema (into {} (map (fn [[k v]] [(k idents) v]) schema)))
|
||||
:current-tx tx0})))))
|
||||
|
||||
(defn connection-with-db [db]
|
||||
(map->Connection {:current-db (atom db)}))
|
||||
|
||||
(defrecord TxReport [db-before db-after entities tx-data tempids])
|
||||
|
||||
(defn- report? [x]
|
||||
(and (instance? TxReport x)))
|
||||
|
||||
;; ;; TODO: persist max-tx and max-eid in SQLite.
|
||||
|
||||
(defn maybe-datom->entity [entity]
|
||||
(cond
|
||||
(datom? entity)
|
||||
(->
|
||||
(let [[e a v tx added] entity]
|
||||
(if added
|
||||
[:db/add [e a v tx]]
|
||||
[:db/retract [e a v tx]]))
|
||||
(with-meta (get (meta entity) :source)))
|
||||
|
||||
true
|
||||
entity))
|
||||
|
||||
(defn maybe-explode [schema entity] ;; TODO db? schema?
|
||||
(cond
|
||||
(map? entity)
|
||||
;; TODO: reverse refs, lists, nested maps
|
||||
(let [eid (or (:db/id entity)
|
||||
(id-literal :db.part/temp))] ;; Must upsert if no ID given. TODO: check if this fails in Datomic/DS.
|
||||
(for [[a v] (dissoc entity :db/id)]
|
||||
[:db/add eid a v]))
|
||||
|
||||
;; (raise "Map entities are not yet supported, got " entity
|
||||
;; {:error :transact/syntax
|
||||
;; :op entity })
|
||||
|
||||
true
|
||||
[entity]))
|
||||
|
||||
(defn maybe-ident->entid [db [op & entity :as orig]]
|
||||
;; TODO: use something faster than `into` here.
|
||||
(->
|
||||
(into [op] (for [field entity]
|
||||
(get (idents db) field field))) ;; TODO: schema, not db.
|
||||
;; (with-meta (get (meta orig) :source {:source orig}))
|
||||
))
|
||||
|
||||
(defrecord Transaction [db tempids entities])
|
||||
|
||||
(defn- tx-entity [db]
|
||||
(let [tx (current-tx db)]
|
||||
[:db/add tx :db/txInstant 0xdeadbeef tx])) ;; TODO: now.
|
||||
|
||||
(defn maybe-add-current-tx [current-tx entity]
|
||||
(let [[op e a v tx] entity]
|
||||
[op e a v (or tx current-tx)]))
|
||||
|
||||
(defn preprocess [db report]
|
||||
(let [initial-es (conj (or (:entities report) []) (tx-entity db))]
|
||||
(when-not (sequential? initial-es)
|
||||
(raise "Bad transaction data " initial-es ", expected sequential collection"
|
||||
{:error :transact/syntax, :tx-data initial-es}))
|
||||
|
||||
(->>
|
||||
(->
|
||||
(comp
|
||||
;; Track the provenance of each assertion for error reporting.
|
||||
(map #(with-meta % {:source %}))
|
||||
|
||||
;; Normalize Datoms into :db/add or :db/retract vectors.
|
||||
(map maybe-datom->entity)
|
||||
|
||||
;; Explode map shorthand, such as {:db/id e :attr value :_reverse ref},
|
||||
;; to a list of vectors, like
|
||||
;; [[:db/add e :attr value] [:db/add ref :reverse e]].
|
||||
(mapcat (partial maybe-explode (schema db)))
|
||||
|
||||
;; Replace idents with entids where possible.
|
||||
(map (partial maybe-ident->entid db))
|
||||
|
||||
;; Add tx if not given.
|
||||
(map (partial maybe-add-current-tx (current-tx db))))
|
||||
(transduce conj [] initial-es))
|
||||
(assoc-in report [:entities]))))
|
||||
|
||||
(defn <?run
|
||||
"Execute the provided query on the provided DB.
|
||||
|
|
@ -45,8 +322,8 @@
|
|||
[db find args]
|
||||
(let [parsed (query/parse find)
|
||||
context (-> db
|
||||
query-context
|
||||
(query/find-into-context parsed))
|
||||
query-context
|
||||
(query/find-into-context parsed))
|
||||
row-pair-transducer (projection/row-pair-transducer context)
|
||||
sql (query/context->sql-string context args)
|
||||
chan (chan 50 row-pair-transducer)]
|
||||
|
|
@ -67,4 +344,206 @@
|
|||
[db find args]
|
||||
(a/reduce (partial reduce-error-pair conj) [[] nil]
|
||||
(<?run db find args)))
|
||||
|
||||
|
||||
(defonce -eid (atom (- 0x200 1)))
|
||||
|
||||
;; TODO: better here.
|
||||
(defn- next-eid [db]
|
||||
(swap! -eid inc))
|
||||
|
||||
(defn- allocate-eid
|
||||
[report id-literal eid]
|
||||
(assoc-in report [:tempids id-literal] eid))
|
||||
|
||||
(declare <resolve-id-literals)
|
||||
|
||||
(defn <retry-with-tempid [db report es tempid upserted-eid]
|
||||
(if (contains? (:tempids report) tempid)
|
||||
(go-pair
|
||||
(raise "Conflicting upsert: " tempid " resolves"
|
||||
" both to " upserted-eid " and " (get (:tempids report) tempid)
|
||||
{ :error :transact/upsert }))
|
||||
;; try to re-run from the beginning
|
||||
;; but remembering that `old-eid` will resolve to `upserted-eid`
|
||||
(<resolve-id-literals db
|
||||
(->
|
||||
report
|
||||
(assoc-in [:tempids tempid] upserted-eid)
|
||||
(assoc-in [:entities] es)))))
|
||||
|
||||
(defn- transact-entity [report entity]
|
||||
(update-in report [:entities] conj entity))
|
||||
|
||||
(defn <resolve-id-literals
|
||||
"Upsert uniquely identified literals when possible and allocate new entids for all other id literals.
|
||||
|
||||
It's worth noting that some amount of trial and error is probably
|
||||
necessary here, since [[-1 :ref -2] [-2 :ref -1]] is a valid input.
|
||||
It's my belief that no graph algorithm can correctly order the
|
||||
id-literals in quasi-linear time, since that algorithm will need to
|
||||
accept all permutations of the id-literals. Therefore, we simplify
|
||||
by accepting that we may process the input multiple times, and we
|
||||
regain some efficiency by sorting so that upserts happen earlier and
|
||||
we are most likely to find a successful entid allocation without
|
||||
multiple trials.
|
||||
|
||||
Concretely, we sort [-1 a v] < [-1 a -2] < [e a -1] < [e a v]. This
|
||||
means simple upserts will be processed early, followed by entities
|
||||
with multiple id-literals that we hope will reduce to simple upserts
|
||||
based on the earlier upserts. After that, we handle what should be
|
||||
simple allocations."
|
||||
|
||||
[db report]
|
||||
(go-pair
|
||||
(let [keyfn (fn [[op e a v tx]]
|
||||
(if (and (id-literal? e)
|
||||
(not-any? id-literal? [a v tx]))
|
||||
(- 5)
|
||||
(- (count (filter id-literal? [e a v tx])))))
|
||||
initial-report (dissoc report :entities) ;; TODO.
|
||||
initial-entities (sort-by keyfn (:entities report))]
|
||||
(loop [report initial-report
|
||||
es initial-entities]
|
||||
(let [[[op e a v tx :as entity] & entities] es]
|
||||
(cond
|
||||
(nil? entity)
|
||||
;; We can add :db.part/temp id-literals; remove them.
|
||||
(update report :tempids #(into {} (filter (comp not temp-literal? first) %)))
|
||||
|
||||
(and (not= op :db/add)
|
||||
(not (empty? (filter id-literal? [e a v tx]))))
|
||||
(raise "id-literals are resolved for :db/add only"
|
||||
{:error :transact/syntax
|
||||
:op entity })
|
||||
|
||||
;; Upsert!
|
||||
(and (id-literal? e)
|
||||
(ds/unique-identity? (schema db) a) ;; TODO: schema.
|
||||
(not-any? id-literal? [a v tx]))
|
||||
(let [upserted-eid (:e (first (<? (<avet db [a v])))) ;; TODO: define this interface.
|
||||
allocated-eid (get-in report [:tempids e])]
|
||||
(if (and upserted-eid allocated-eid (not= upserted-eid allocated-eid))
|
||||
(do
|
||||
(<? (<retry-with-tempid db initial-report initial-entities e upserted-eid)) ;; TODO: not initial report, just the sorted entities here.
|
||||
)
|
||||
(let [eid (or upserted-eid allocated-eid (next-eid db))]
|
||||
(recur (allocate-eid report e eid) (cons [op eid a v tx] entities)))))
|
||||
|
||||
;; Start allocating and retrying. We try with e last, so as to eventually upsert it.
|
||||
(id-literal? tx)
|
||||
;; TODO: enforce tx part only?
|
||||
(let [eid (or (get-in report [:tempids tx]) (next-eid db))]
|
||||
(recur (allocate-eid report tx eid) (cons [op e a v eid] entities)))
|
||||
|
||||
(id-literal? v)
|
||||
;; We can't fail with unbound literals here, since we could have multiple
|
||||
(let [eid (or (get-in report [:tempids v]) (next-eid db))]
|
||||
(recur (allocate-eid report v eid) (cons [op e a eid tx] entities)))
|
||||
|
||||
(id-literal? a)
|
||||
;; TODO: should we even allow id-literal attributes? Datomic fails in some cases here.
|
||||
(let [eid (or (get-in report [:tempids a]) (next-eid db))]
|
||||
(recur (allocate-eid report a eid) (cons [op e eid v tx] entities)))
|
||||
|
||||
(id-literal? e)
|
||||
(let [eid (or (get-in report [:tempids e]) (next-eid db))]
|
||||
(recur (allocate-eid report e eid) (cons [op eid a v tx] entities)))
|
||||
|
||||
true
|
||||
(recur (transact-entity report entity) entities)
|
||||
))))))
|
||||
|
||||
(defn- transact-report [report datom]
|
||||
(update-in report [:tx-data] conj datom))
|
||||
|
||||
(defn <postprocess [db report]
|
||||
(go-pair
|
||||
(let [initial-report report]
|
||||
(loop [report initial-report
|
||||
es (:entities initial-report)]
|
||||
(let [[[op e a v tx :as entity] & entities] es]
|
||||
(cond
|
||||
(nil? entity)
|
||||
report
|
||||
|
||||
(= op :db/add)
|
||||
(if (ds/multival? (schema db) a)
|
||||
(if (empty? (<? (<eavt db [e a v])))
|
||||
(recur (transact-report report (datom e a v tx true)) entities)
|
||||
(recur report entities))
|
||||
(if-let [^Datom old-datom (first (<? (<eavt db [e a])))]
|
||||
(if (= (.-v old-datom) v)
|
||||
(recur report entities)
|
||||
(recur (-> report
|
||||
(transact-report (datom e a (.-v old-datom) tx false))
|
||||
(transact-report (datom e a v tx true)))
|
||||
entities))
|
||||
(recur (transact-report report (datom e a v tx true)) entities)))
|
||||
|
||||
(= op :db/retract)
|
||||
(if (first (<? (<eavt db [e a v])))
|
||||
(recur (transact-report report (datom e a v tx false)) entities)
|
||||
(recur report entities))
|
||||
|
||||
true
|
||||
(raise "Unknown operation at " entity ", expected :db/add, :db/retract"
|
||||
{:error :transact/syntax, :operation op, :tx-data entity})))))))
|
||||
|
||||
(defn <transact-tx-data
|
||||
[db now initial-report]
|
||||
{:pre [(db? db)]}
|
||||
|
||||
(go-pair
|
||||
(->> initial-report
|
||||
(preprocess db)
|
||||
|
||||
(<resolve-id-literals db)
|
||||
(<?)
|
||||
|
||||
(<postprocess db)
|
||||
(<?))))
|
||||
|
||||
;; Normalize as [op int|id-literal int|id-literal value|id-literal tx|id-literal]. ;; TODO: mention lookup-refs.
|
||||
|
||||
;; Replace lookup-refs with entids where possible.
|
||||
|
||||
;; Upsert or allocate id-literals.
|
||||
|
||||
(defn <with [db tx-data]
|
||||
(go-pair
|
||||
(let [report (<? (<transact-tx-data db 0xdeadbeef ;; TODO
|
||||
(map->TxReport
|
||||
{:db-before db
|
||||
:db-after db
|
||||
;; :current-tx current-tx
|
||||
:entities tx-data
|
||||
:tx-data []
|
||||
:tempids {}})))
|
||||
db-after (->
|
||||
db
|
||||
|
||||
(<apply-datoms (:tx-data report))
|
||||
(<?)
|
||||
|
||||
(<advance-tx)
|
||||
(<?))]
|
||||
|
||||
(-> report
|
||||
(assoc-in [:db-after] db-after)))))
|
||||
|
||||
(defn <db-with [db tx-data]
|
||||
(go-pair
|
||||
(:db-after (<? (<with db tx-data)))))
|
||||
|
||||
(defn <transact!
|
||||
([conn tx-data]
|
||||
(<transact! conn tx-data 0xdeadbeef)) ;; TODO: timestamp!
|
||||
([conn tx-data now]
|
||||
{:pre [(conn? conn)]}
|
||||
(let [db (db conn)] ;; TODO: be careful with swapping atoms.
|
||||
(s/in-transaction!
|
||||
(:sqlite-connection db)
|
||||
#(go-pair
|
||||
(let [report (<? (<with db tx-data))] ;; TODO: timestamp!
|
||||
(reset! (:current-db conn) (:db-after report))
|
||||
report))))))
|
||||
|
|
|
|||
96
src/datomish/schema.cljc
Normal file
96
src/datomish/schema.cljc
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
|
||||
;; Purloined from DataScript.
|
||||
|
||||
(ns datomish.schema)
|
||||
|
||||
(defprotocol ISchema
|
||||
(attrs-by
|
||||
[schema property]
|
||||
"TODO: document this, think more about making this part of the schema."))
|
||||
|
||||
(defn- #?@(:clj [^Boolean is-attr?]
|
||||
:cljs [^boolean is-attr?]) [schema attr property]
|
||||
(contains? (attrs-by schema property) attr))
|
||||
|
||||
(defn #?@(:clj [^Boolean multival?]
|
||||
:cljs [^boolean multival?]) [schema attr]
|
||||
(is-attr? schema attr :db.cardinality/many))
|
||||
|
||||
(defn #?@(:clj [^Boolean ref?]
|
||||
:cljs [^boolean ref?]) [schema attr]
|
||||
(is-attr? schema attr :db.type/ref))
|
||||
|
||||
(defn #?@(:clj [^Boolean component?]
|
||||
:cljs [^boolean component?]) [schema attr]
|
||||
(is-attr? schema attr :db/isComponent))
|
||||
|
||||
(defn #?@(:clj [^Boolean indexing?]
|
||||
:cljs [^boolean indexing?]) [schema attr]
|
||||
(is-attr? schema attr :db/index))
|
||||
|
||||
(defn #?@(:clj [^Boolean unique-identity?]
|
||||
:cljs [^boolean unique-identity?]) [schema attr]
|
||||
(is-attr? schema attr :db.unique/identity))
|
||||
|
||||
(defn #?@(:clj [^Boolean unique-value?]
|
||||
:cljs [^boolean unique-value?]) [schema attr]
|
||||
(is-attr? schema attr :db.unique/value))
|
||||
|
||||
(defrecord Schema [schema rschema]
|
||||
ISchema
|
||||
(attrs-by [schema property]
|
||||
((.-rschema schema) property)))
|
||||
|
||||
(defn- attr->properties [k v]
|
||||
(cond
|
||||
(= [k v] [:db/isComponent true]) [:db/isComponent]
|
||||
(= v :db.type/ref) [:db.type/ref :db/index]
|
||||
(= v :db.cardinality/many) [:db.cardinality/many]
|
||||
(= v :db.unique/identity) [:db/unique :db.unique/identity :db/index]
|
||||
(= v :db.unique/value) [:db/unique :db.unique/value :db/index]
|
||||
(= [k v] [:db/index true]) [:db/index]))
|
||||
|
||||
(defn- multimap [e m]
|
||||
(reduce
|
||||
(fn [acc [k v]]
|
||||
(update-in acc [k] (fnil conj e) v))
|
||||
{} m))
|
||||
|
||||
(defn- rschema [schema]
|
||||
(->>
|
||||
(for [[a kv] schema
|
||||
[k v] kv
|
||||
prop (attr->properties k v)]
|
||||
[prop a])
|
||||
(multimap #{})))
|
||||
|
||||
(defn- validate-schema-key [a k v expected]
|
||||
(when-not (or (nil? v)
|
||||
(contains? expected v))
|
||||
(throw (ex-info (str "Bad attribute specification for " (pr-str {a {k v}}) ", expected one of " expected)
|
||||
{:error :schema/validation
|
||||
:attribute a
|
||||
:key k
|
||||
:value v}))))
|
||||
|
||||
(defn- validate-schema [schema]
|
||||
(doseq [[a kv] schema]
|
||||
(let [comp? (:db/isComponent kv false)]
|
||||
(validate-schema-key a :db/isComponent (:db/isComponent kv) #{true false})
|
||||
(when (and comp? (not= (:db/valueType kv) :db.type/ref))
|
||||
(throw (ex-info (str "Bad attribute specification for " a ": {:db/isComponent true} should also have {:db/valueType :db.type/ref}")
|
||||
{:error :schema/validation
|
||||
:attribute a
|
||||
:key :db/isComponent}))))
|
||||
(validate-schema-key a :db/unique (:db/unique kv) #{:db.unique/value :db.unique/identity})
|
||||
(validate-schema-key a :db/valueType (:db/valueType kv) #{:db.type/ref})
|
||||
(validate-schema-key a :db/cardinality (:db/cardinality kv) #{:db.cardinality/one :db.cardinality/many}))
|
||||
schema)
|
||||
|
||||
(defn schema [schema]
|
||||
{:pre [(or (nil? schema) (map? schema))]}
|
||||
(map->Schema {:schema (validate-schema schema)
|
||||
:rschema (rschema schema)}))
|
||||
149
test/datomish/db_test.cljc
Normal file
149
test/datomish/db_test.cljc
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
;; This Source Code Form is subject to the terms of the Mozilla Public
|
||||
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
|
||||
(ns datomish.db-test
|
||||
#?(:cljs
|
||||
(:require-macros
|
||||
[datomish.pair-chan :refer [go-pair <?]]
|
||||
[datomish.node-tempfile-macros :refer [with-tempfile]]
|
||||
[cljs.core.async.macros :as a :refer [go]]))
|
||||
(:require
|
||||
[datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise cond-let]]
|
||||
[datomish.sqlite :as s]
|
||||
[datomish.datom]
|
||||
|
||||
[datascript.core :as d]
|
||||
[datascript.db :as db]
|
||||
|
||||
[datomish.db :as dm]
|
||||
#?@(:clj [[datomish.pair-chan :refer [go-pair <?]]
|
||||
[tempfile.core :refer [tempfile with-tempfile]]
|
||||
[datomish.test-macros :refer [deftest-async]]
|
||||
[clojure.test :as t :refer [is are deftest testing]]
|
||||
[clojure.core.async :refer [go <! >!]]])
|
||||
#?@(:cljs [[datomish.pair-chan]
|
||||
[datomish.test-macros :refer-macros [deftest-async]]
|
||||
[datomish.node-tempfile :refer [tempfile]]
|
||||
[cljs.test :as t :refer-macros [is are deftest testing async]]
|
||||
[cljs.core.async :as a :refer [<! >!]]]))
|
||||
#?(:clj
|
||||
(:import [clojure.lang ExceptionInfo]))
|
||||
#?(:clj
|
||||
(:import [datascript.db DB])))
|
||||
|
||||
(defn- <datoms [db]
|
||||
(let [entids (zipmap (vals (dm/idents db)) (keys (dm/idents db)))]
|
||||
(go-pair
|
||||
(->>
|
||||
(s/all-rows (:sqlite-connection db) ["SELECT e, a, v, tx FROM datoms"])
|
||||
(<?)
|
||||
(mapv #(vector (:e %) (entids (:a %)) (:v %)))
|
||||
(filter #(not (= :db/txInstant (second %))))
|
||||
(set)))))
|
||||
|
||||
(defn- <transactions [db]
|
||||
(let [entids (zipmap (vals (dm/idents db)) (keys (dm/idents db)))]
|
||||
(go-pair
|
||||
(->>
|
||||
(s/all-rows (:sqlite-connection db) ["SELECT e, a, v, tx, added FROM transactions ORDER BY tx ASC, e, a, v, added"])
|
||||
(<?)
|
||||
(mapv #(vector (:e %) (entids (:a %)) (:v %) (:tx %) (:added %)))))))
|
||||
|
||||
(defn tx [report]
|
||||
(get-in report [:db-after :current-tx]))
|
||||
|
||||
(deftest-async test-add-one
|
||||
(with-tempfile [t (tempfile)]
|
||||
(let [c (<? (s/<sqlite-connection t))
|
||||
db (<? (dm/<db-with-sqlite-connection c))
|
||||
conn (dm/connection-with-db db)
|
||||
now 0xdeadbeef]
|
||||
(try
|
||||
(let [;; TODO: drop now, allow to set :db/txInstant.
|
||||
report (<? (dm/<transact! conn [[:db/add 0 :x "valuex"]] now))
|
||||
tx (tx report)]
|
||||
(is (= (<? (<datoms (dm/db conn)))
|
||||
#{[0 :x "valuex"]}))
|
||||
(is (= (<? (<transactions (dm/db conn)))
|
||||
[[0 :x "valuex" tx 1] ;; TODO: true, not 1.
|
||||
[tx :db/txInstant now tx 1]])))
|
||||
(finally
|
||||
(<? (dm/close-db db)))))))
|
||||
|
||||
(deftest-async test-add-two
|
||||
(with-tempfile [t (tempfile)]
|
||||
(let [c (<? (s/<sqlite-connection t))
|
||||
db (<? (dm/<db-with-sqlite-connection c
|
||||
{:x {:db/unique :db.unique/identity} ;; TODO: :name and :aka.
|
||||
:y {:db/cardinality :db.cardinality/many}}))
|
||||
conn (dm/connection-with-db db)
|
||||
now 0xdeadbeef]
|
||||
(try
|
||||
(let [tx1 (tx (<? (dm/<transact! conn [[:db/add 1 :x "Ivan"]] now)))
|
||||
tx2 (tx (<? (dm/<transact! conn [[:db/add 1 :x "Petr"]] now)))
|
||||
tx3 (tx (<? (dm/<transact! conn [[:db/add 1 :y "Tupen"]] now)))
|
||||
tx4 (tx (<? (dm/<transact! conn [[:db/add 1 :y "Devil"]] now)))]
|
||||
(is (= (<? (<datoms (dm/db conn)))
|
||||
#{[1 :x "Petr"]
|
||||
[1 :y "Tupen"]
|
||||
[1 :y "Devil"]}))
|
||||
|
||||
(is (= (<? (<transactions (dm/db conn)))
|
||||
[[1 :x "Ivan" tx1 1] ;; TODO: true, not 1.
|
||||
[tx1 :db/txInstant now tx1 1]
|
||||
[1 :x "Ivan" tx2 0]
|
||||
[1 :x "Petr" tx2 1]
|
||||
[tx2 :db/txInstant now tx2 1]
|
||||
[1 :y "Tupen" tx3 1]
|
||||
[tx3 :db/txInstant now tx3 1]
|
||||
[1 :y "Devil" tx4 1]
|
||||
[tx4 :db/txInstant now tx4 1]])))
|
||||
(finally
|
||||
(<? (dm/close-db db)))))))
|
||||
|
||||
;; TODO: fail multiple :add and :retract of the same datom in the same transaction.
|
||||
(deftest-async test-retract
|
||||
(with-tempfile [t (tempfile)]
|
||||
(let [c (<? (s/<sqlite-connection t))
|
||||
db (<? (dm/<db-with-sqlite-connection c))
|
||||
conn (dm/connection-with-db db)
|
||||
now 0xdeadbeef]
|
||||
(try
|
||||
(let [txa (tx (<? (dm/<transact! conn [[:db/add 0 :x "valuex"]] now)))
|
||||
txb (tx (<? (dm/<transact! conn [[:db/retract 0 :x "valuex"]] now)))]
|
||||
(is (= (<? (<datoms db))
|
||||
#{}))
|
||||
(is (= (<? (<transactions db))
|
||||
[[0 :x "valuex" txa 1] ;; TODO: true, not 1.
|
||||
[txa :db/txInstant now txa 1]
|
||||
[0 :x "valuex" txb 0]
|
||||
[txb :db/txInstant now txb 1]])))
|
||||
(finally
|
||||
(<? (dm/close-db db)))))))
|
||||
|
||||
(deftest-async test-id-literal-1
|
||||
(with-tempfile [t (tempfile)]
|
||||
(let [c (<? (s/<sqlite-connection t))
|
||||
db (<? (dm/<db-with-sqlite-connection c))
|
||||
conn (dm/connection-with-db db)
|
||||
now -1]
|
||||
(try
|
||||
(let [report (<? (dm/<transact! conn [[:db/add (dm/id-literal :db.part/user -1) :x 0]
|
||||
[:db/add (dm/id-literal :db.part/user -1) :y 1]
|
||||
[:db/add (dm/id-literal :db.part/user -2) :y 2]
|
||||
[:db/add (dm/id-literal :db.part/user -2) :y 3]] now))]
|
||||
(is (= (keys (:tempids report)) ;; TODO: include values.
|
||||
[(dm/id-literal :db.part/user -1)
|
||||
(dm/id-literal :db.part/user -2)]))
|
||||
|
||||
(let [eid1 (get-in report [:tempids (dm/id-literal :db.part/user -1)])
|
||||
eid2 (get-in report [:tempids (dm/id-literal :db.part/user -2)])]
|
||||
(is (= (<? (<datoms db))
|
||||
#{[eid1 :x 0]
|
||||
[eid1 :y 1]
|
||||
[eid2 :y 2]
|
||||
[eid2 :y 3]}))))
|
||||
|
||||
(finally
|
||||
(<? (dm/close-db db)))))))
|
||||
|
|
@ -3,6 +3,7 @@
|
|||
[doo.runner :refer-macros [doo-tests doo-all-tests]]
|
||||
[cljs.test :as t :refer-macros [is are deftest testing]]
|
||||
datomish.promise-sqlite-test
|
||||
datomish.db-test
|
||||
datomish.sqlite-user-version-test
|
||||
datomish.test.util
|
||||
datomish.test.transforms
|
||||
|
|
@ -11,6 +12,7 @@
|
|||
|
||||
(doo-tests
|
||||
'datomish.promise-sqlite-test
|
||||
'datomish.db-test
|
||||
'datomish.sqlite-user-version-test
|
||||
'datomish.test.util
|
||||
'datomish.test.transforms
|
||||
|
|
|
|||
Loading…
Reference in a new issue