Export to and import from CBOR files using clj-cbor, for byte array s…

…upport. Tests to follow.

deps.edn

@@ -10,6 +10,7 @@
         io.lambdaforge/wanderung                    {:mvn/version "0.2.54"}
         io.replikativ/zufall                        {:mvn/version "0.1.0"}
         junit/junit                                 {:mvn/version "4.13.2"}
+        mvxcvi/clj-cbor                             {:mvn/version "1.1.0"}
         scicloj/tablecloth                          {:mvn/version "6.031"}}
 
  :paths ["src" "target/classes"]

src/datahike/migrate.clj

@@ -2,12 +2,11 @@
   (:require [datahike.api :as api]
             [datahike.datom :as d]
             [datahike.db :as db]
+            [clj-cbor.core :as cbor]
             [clojure.java.io :as io]
             [wanderung.datahike :as wd]
             [tablecloth.api :as tc]))
 
-;; TODO Fressian > Nippy
-
 (defn export-db
   "Export the database in a flat-file of datoms at path."
   [db path]
@@ -23,16 +22,14 @@
   (with-open [f (io/output-stream path)
               w (io/writer f)]
     (binding [*out* w]
-      (doseq [d (let [txs (sort-by first (api/q wd/find-tx-datoms db (java.util.Date. 70)))
-                      query {:query wd/find-datoms-in-tx
-                             :args [(api/history db)]}]
-                  (mapcat (fn [[tid tinst]]
-                            (->> (api/q (update-in query [:args] conj tid))
-                                 (sort-by first)
-                                 (map #(apply d/datom %))
-                                 (into [(d/datom tid :db/txInstant tinst tid true)])))
-                          txs))]
-        (prn d)))))
+      (let [txs (sort-by first (api/q wd/find-tx-datoms db (java.util.Date. 70)))
+            query {:query wd/find-datoms-in-tx
+                   :args [(api/history db)]}]
+        (cbor/spit-all path (mapcat (fn [[tid tinst]]
+                                      (->> (api/q (update-in query [:args] conj tid))
+                                           (sort-by first)
+                                           (into [[tid :db/txInstant tinst tid true]])))
+                                    txs))))))
 
 (defn export-db-tc
   "Export the database in a flat-file of datoms at path."
@@ -44,40 +41,38 @@
             datoms-table (tc/dataset (map seq datoms)
                                      {:layout :as-rows
                                       :column-names [:eid :attr :val :txid :assert]})]
-        (doseq [d (tc/rows (tc/order-by datoms-table [:txid :eid]) :as-seq)]
-          (prn (apply d/datom d)))))))
+        (cbor/spit-all path (tc/rows (tc/order-by datoms-table [:txid :eid]) :as-seq))))))
 
 (defn export-db-clj
   "Export the database in a flat-file of datoms at path."
   [db path]
   (with-open [f (io/output-stream path)
               w (io/writer f)]
     (binding [*out* w]
-      (doseq [d (sort-by (juxt d/datom-tx :e) (api/datoms (api/history db) :eavt))]
-        (prn d)))))
+      (cbor/spit-all path (map seq (sort-by (juxt d/datom-tx :e) (api/datoms (api/history db) :eavt)))))))
 
-(defn update-max-tx-from-file
-  "Find bigest tx in file and update max-tx of db.
+(defn update-max-tx
+  "Find bigest tx in datoms and update max-tx of db.
   Note: the last tx might not be the biggest if the db
   has been imported before."
-  [db file]
-  (let [max-tx (->> (line-seq (io/reader file))
-                    (map read-string)
-                    (reduce #(max %1 (nth %2 3)) 0))]
-    (assoc db :max-tx max-tx)))
+  [db datoms]
+  (assoc db :max-tx (reduce #(max %1 (nth %2 3)) 0 datoms)))
+
+(defn- instance-to-date [v]
+  (if (instance? java.time.Instant v) (java.util.Date/from v) v))
 
 (defn import-db
   "Import a flat-file of datoms at path into your database."
   [conn path]
   (println "Preparing import of" path "in batches of 1000")
-  (swap! conn update-max-tx-from-file path)
-  (print "Importing ")
-  (time
-   (doseq  [datoms (->> (line-seq (io/reader path))
-                        (map read-string)
-                        (partition 1000 1000 nil))]
-     (print ".")
-     (api/transact conn (vec datoms)))))
+  (let [datoms (->> (cbor/slurp-all path)
+                    (map #(-> (apply datom/datom %) (update :v instance-to-date))))]
+    (swap! conn update-max-tx datoms)
+    (print "Importing ")
+    (time
+     (doseq [partitions (partition 1000 1000 nil datoms)]
+       (print ".")
+       (api/transact conn (vec datoms))))))
 
 (comment
   (require '[datahike.api :as api]