Query middleware (#566)

* allow local kaocha test config

* add hashp for dev environment

* add query middleware

* fix formatting

* move to list of middlewares

* rename middleware, add q db check

* Fix formatting, again

* add historcial db middleware tests

* moved and adjusted middleware tests

* moved middleware tests into test folder
* used print instead of logging for timed-q
* fixed formatting
* removed :middleware test id

* add newline to gitignore

.gitignore

@@ -28,3 +28,4 @@ out/
 /.settings
 .classpath
 /.clj-kondo
+tests.user.edn

bin/run-all-tests

@@ -2,7 +2,11 @@
 
 set -o errexit
 set -o pipefail
-
+echo "Recompiling Java"
+clj -T:build clean
+clj -T:build compile
+echo "Fix formatting"
+clj -M:ffix
 echo "Running unit tests"
 bin/run-unittests --reporter kaocha.report/dots
 echo "Running integration tests"

deps.edn

@@ -29,7 +29,9 @@
                               clj-http/clj-http           {:mvn/version "3.12.3"}
                               org.clojure/tools.cli       {:mvn/version "1.0.206"}
                               incanter/incanter-core      {:mvn/version "1.9.3"}
-                              incanter/incanter-charts    {:mvn/version "1.9.3"}}}
+                              incanter/incanter-charts    {:mvn/version "1.9.3"}
+                              hashp/hashp {:mvn/version "0.2.2"}}
+                 :main-opts ["-e" "(require 'hashp.core)"]}
 
            :test {:extra-paths ["test"]
                   :extra-deps {lambdaisland/kaocha         {:mvn/version "1.70.1086"}

dev/user.clj

@@ -0,0 +1,79 @@
+(ns user
+  (:require [datahike.api :as d]))
+
+(comment
+
+  (def schema [{:db/ident       :name
+                :db/cardinality :db.cardinality/one
+                :db/index       true
+                :db/unique      :db.unique/identity
+                :db/valueType   :db.type/string}
+               {:db/ident       :sibling
+                :db/cardinality :db.cardinality/many
+                :db/valueType   :db.type/ref}
+               {:db/ident       :age
+                :db/cardinality :db.cardinality/one
+                :db/valueType   :db.type/long}])
+
+  (def cfg {:store              {:backend :mem :id "sandbox"}
+            :keep-history?      true
+            :schema-flexibility :write
+            :middleware         {:query ['datahike.middleware.query/timed]}
+            :attribute-refs?    true})
+
+  (def conn (do
+              (d/delete-database cfg)
+              (d/create-database cfg)
+              (d/connect cfg)))
+
+  (d/transact conn schema)
+
+  (d/datoms @conn :avet)
+  (d/datoms @conn :aevt)
+  (d/datoms @conn :eavt)
+
+  (:max-eid @conn)
+
+  (d/transact conn [{:name "Alice"
+                     :age  25}])
+
+  (d/transact conn [{:name "Bob"
+                     :age  25}])
+
+  (d/transact conn [{:name    "Charlie"
+                     :age     45
+                     :sibling [[:name "Alice"] [:name "Bob"]]}])
+
+  (d/q '[:find ?e ?a ?v ?t
+         :in $ ?a
+         :where
+         [?e :name ?v ?t]
+         [?e :age ?a]]
+       @conn
+       25)
+
+  (d/q '[:find ?e ?at ?v
+         :where
+         [?e ?a ?v]
+         [?a :db/ident ?at]]
+       @conn)
+
+
+  (d/q '[:find ?e :where [?e :name "Alice"]] @conn)
+
+  (do (d/transact conn (vec (repeatedly 5000 (fn [] {:age  (long (rand-int 1000))
+                                                     :name (str (rand-int 1000))}))))
+      true)
+
+  (d/q {:query '[:find ?e ?v
+                 :in $
+                 :where [?e :name ?v]]
+        :args [@conn]
+        :offset 0
+        :limit 10})
+
+  (do (d/q {:query '[:find ?v1 ?v2
+                     :in $
+                     :where [?e1 :name ?v1] [?e2 :name ?v2]]
+            :args [@conn]})
+      nil))

src/datahike/config.cljc

@@ -20,6 +20,9 @@
 (s/def ::index-b-factor long)
 (s/def ::index-log-size long)
 (s/def ::index-data-node-size long)
+(s/def :datahike.middleware/fn symbol?)
+(s/def :datahike.middleware/query (s/coll-of :datahike.middleware/fn))
+(s/def ::middleware (s/keys :opt-un [:datahike.middleware/query]))
 
 (s/def ::store map?)
 
@@ -30,7 +33,8 @@
                                          ::schema-flexibility
                                          ::attribute-refs?
                                          ::initial-tx
-                                         ::name]))
+                                         ::name
+                                         ::middleware]))
 
 (s/def :deprecated/schema-on-read boolean?)
 (s/def :deprecated/temporal-index boolean?)

src/datahike/middleware/query.cljc

@@ -0,0 +1,14 @@
+(ns datahike.middleware.query
+  (:require [clojure.pprint :as pprint]))
+
+(defn timed-query [query-handler]
+  (fn [query & inputs]
+    (let [start (. System (nanoTime))
+          result (apply query-handler query inputs)
+          t (/ (double (- (. System (nanoTime)) start)) 1000000.0)]
+      (println "Query time:")
+      (pprint/pprint {:t      t
+                      :q      (update query :args str)
+                      :inputs (str inputs)})
+      result)))
+

src/datahike/middleware/utils.cljc

@@ -0,0 +1,12 @@
+(ns datahike.middleware.utils)
+
+(defn apply-middlewares
+  "Combines a list of middleware functions into one."
+  [middlewares handler]
+  (reduce
+   (fn [acc f-sym]
+     (if-let [f (resolve f-sym)]
+       (f acc)
+       (throw (ex-info "Invalid middleware.😱" {:fn f-sym}))))
+   handler
+   middlewares))

src/datahike/query.cljc

@@ -8,8 +8,10 @@
    [datahike.db.utils :as dbu]
    [datahike.impl.entity :as de]
    [datahike.lru]
+   [datahike.middleware.query]
    [datahike.pull-api :as dpa]
    [datahike.tools #?(:cljs :refer-macros :clj :refer) [raise]]
+   [datahike.middleware.utils :as middleware-utils]
    [datalog.parser :refer [parse]]
    [datalog.parser.impl :as dpi]
    [datalog.parser.impl.proto :as dpip]
@@ -1139,7 +1141,7 @@
 (defmethod q clojure.lang.PersistentList [query & args]
   (q {:query query :args args}))
 
-(defmethod q clojure.lang.PersistentArrayMap [query-map & inputs]
+(defn raw-q [query-map & inputs]
   (let [query         (if (contains? query-map :query) (:query query-map) query-map)
         query         (if (string? query) (edn/read-string query) query)
         query         (if (= 'quote (first query)) (second query) query)
@@ -1170,3 +1172,10 @@
       (some #(instance? Pull %) find-elements)      (pull find-elements context)
       true                                          (-post-process find)
       returnmaps                                    (convert-to-return-maps returnmaps))))
+
+(defmethod q clojure.lang.PersistentArrayMap [{:keys [args] :as query-map} & inputs]
+  (if-let [middleware (when (dbu/db? (first args))
+                        (get-in (dbi/-config (first args)) [:middleware :query]))]
+    (let [q-with-middleware (middleware-utils/apply-middlewares middleware raw-q)]
+      (q-with-middleware query-map inputs))
+    (apply raw-q query-map inputs)))

test/datahike/test.cljc

@@ -41,7 +41,9 @@
    datahike.test.attribute-refs.pull-api-test
    datahike.test.attribute-refs.query-test
    datahike.test.attribute-refs.transact-test
-   datahike.test.attribute-refs.utils))
+   datahike.test.attribute-refs.utils
+   datahike.test.middleware.query-test
+   datahike.test.middleware.utils-test))
 
 (defn ^:export test-clj []
   (datahike.test.core/wrap-res #(t/run-all-tests #"datahike\..*")))

test/datahike/test/middleware/query_test.cljc

@@ -0,0 +1,90 @@
+(ns datahike.test.middleware.query-test
+  (:require
+   [clojure.test :refer [deftest is]]
+   [datahike.api :as d]
+   [datahike.middleware.query]
+   [datahike.test.utils :as utils])
+  (:import
+   [clojure.lang ExceptionInfo]
+   [java.util Date]))
+
+(deftest timed-query-should-log-time-for-query-to-run
+  (let [cfg {:store {:backend :mem
+                     :id "query-middleware"}
+             :keep-history? false
+             :schema-flexibility :read
+             :middleware {:query ['datahike.middleware.query/timed-query]}}
+        conn (utils/setup-db cfg)]
+    (d/transact conn {:tx-data [{:name "Anna"}
+                                {:name "Boris"}]})
+    (let [out-str (with-out-str (d/q '[:find ?e :where [?e :name "Anna"]] @conn))]
+      (is (re-find #"Query time" out-str))
+      (is (re-find #":query" out-str))
+      (is (re-find #":args" out-str))
+      (is (re-find #"DB" out-str))
+      (is (re-find #"[:find ?e :where [?e :name \"Anna\"]]" out-str))
+      (is (re-find #":t" out-str)))))
+
+(deftest invalid-middleware-should-be-caught-on-connection
+  (let [cfg {:store {:backend :mem
+                     :id "query-middleware"}
+             :keep-history? false
+             :schema-flexibility :read
+             :middleware {:query "this is neither a function nor a vector!"}}]
+    (is (thrown-with-msg? ExceptionInfo #"Invalid Datahike configuration." (utils/setup-db cfg)))))
+
+(deftest middleware-should-work-with-as-of-db
+  (let [cfg {:store {:backend :mem
+                     :id "query-middleware"}
+             :keep-history? true
+             :schema-flexibility :read
+             :middleware {:query ['datahike.middleware.query/timed-query]}}
+        conn (utils/setup-db cfg)]
+    (d/transact conn {:tx-data [{:name "Anna"}
+                                {:name "Boris"}]})
+    (let [before (Date.)
+          _ (d/transact conn {:tx-data [{:name "Charlize"}]})
+          out-str (with-out-str (d/q '[:find ?e :where [?e :name "Anna"]] (d/as-of @conn before)))]
+      (is (re-find #"Query time" out-str))
+      (is (re-find #":query" out-str))
+      (is (re-find #":args" out-str))
+      (is (re-find #"AsOfDB" out-str))
+      (is (re-find #"[:find ?e :where [?e :name \"Anna\"]]" out-str))
+      (is (re-find #":t" out-str)))))
+
+(deftest middleware-should-work-with-since
+  (let [cfg {:store {:backend :mem
+                     :id "query-middleware"}
+             :keep-history? true
+             :schema-flexibility :read
+             :middleware {:query ['datahike.middleware.query/timed-query]}}
+        conn (utils/setup-db cfg)]
+    (d/transact conn {:tx-data [{:name "Anna"}
+                                {:name "Boris"}]})
+    (let [before (Date.)
+          _ (d/transact conn {:tx-data [{:name "Charlize"}]})
+          out-str (with-out-str (d/q '[:find ?e :where [?e :name "Anna"]] (d/since @conn before)))]
+      (is (re-find #"Query time" out-str))
+      (is (re-find #":query" out-str))
+      (is (re-find #":args" out-str))
+      (is (re-find #"SinceDB" out-str))
+      (is (re-find #"[:find ?e :where [?e :name \"Anna\"]]" out-str))
+      (is (re-find #":t" out-str)))))
+
+(deftest middleware-should-work-with-history
+  (let [cfg {:store {:backend :mem
+                     :id "query-middleware"}
+             :keep-history? true
+             :schema-flexibility :read
+             :middleware {:query ['datahike.middleware.query/timed-query]}}
+        conn (utils/setup-db cfg)]
+    (d/transact conn {:tx-data [{:name "Anna"}
+                                {:name "Boris"}]})
+    (let [_ (d/transact conn {:tx-data [{:name "Charlize"}]})
+          out-str (with-out-str (d/q '[:find ?e :where [?e :name "Anna"]] (d/history @conn)))]
+      (is (re-find #"Query time" out-str))
+      (is (re-find #":query" out-str))
+      (is (re-find #":args" out-str))
+      (is (re-find #"HistoricalDB" out-str))
+      (is (re-find #"[:find ?e :where [?e :name \"Anna\"]]" out-str))
+      (is (re-find #":t" out-str)))))

test/datahike/test/middleware/utils_test.cljc

@@ -0,0 +1,19 @@
+(ns datahike.test.middleware.utils-test
+  (:require [clojure.test :refer [deftest is testing]]
+            [datahike.middleware.utils :as utils]))
+
+(defn test-handler-inc-first [handler] (fn [a b] (handler (+ a 5) b)))
+(defn test-handler-inc-second [handler] (fn [a b] (handler a (+ b 10))))
+
+(deftest apply-middlewares-should-combine-symbols-to-new-function
+  (let [combined (utils/apply-middlewares ['datahike.test.middleware.utils-test/test-handler-inc-first
+                                           'datahike.test.middleware.utils-test/test-handler-inc-second]
+                                          (fn [a b] (+ a b)))]
+    (is (= 15
+           (combined 0 0)))))
+
+(deftest apply-middlewares-should-throw-exception-on-non-resolved-symbols
+  (is (thrown-with-msg? clojure.lang.ExceptionInfo #"Invalid middleware.😱"
+                        (utils/apply-middlewares ['is-not-there
+                                                  'datahike.middleware.utils-test/test-handler-inc-second]
+                                                 (fn [a b] (+ a b))))))

tests.edn

@@ -1,14 +1,19 @@
-#kaocha/v1 {:tests [{:id :clj
-                     :ns-patterns ["datahike.test."]}
-                    #_{:id :cljs
-                       :type :kaocha.type/cljs
-                       :ns-patterns ["datahike.test."]}
-                    {:id :quick
-                     :ns-patterns ["datahike.test."]
-                     :bindings {datahike.test.config/*user-filter-map* [{:index :datahike.index/persistent-set
-                                                                         :backend :mem
-                                                                         :schema-flexibility :read
-                                                                         }]}}
-                    {:id :integration
-                     :test-paths ["test/datahike/integration_test"]}]
-            :reporter kaocha.report/documentation}
+#kaocha/v1
+        #meta-merge [{:tests    [{:id          :clj
+                                  :focus-meta [:focused]
+                                  :ns-patterns ["datahike.test."]}
+                                 #_{:id          :cljs
+                                    :type        :kaocha.type/cljs
+                                    :ns-patterns ["datahike.test."]}
+                                 {:id          :quick
+                                  :focus-meta [:focused]
+                                  :ns-patterns ["datahike.test."]
+                                  :bindings    {datahike.test.config/*user-filter-map* [{:index              :datahike.index/persistent-set
+                                                                                         :backend            :mem
+                                                                                         :schema-flexibility :read
+                                                                                         }]}}
+                                 {:id         :integration
+                                  :focus-meta [:focused]
+                                  :test-paths ["test/datahike/integration_test"]}]
+                      :reporter kaocha.report/documentation}
+                     #include "tests.user.edn"]