Drastically optimize queries by using a local index

Author: plexus

src/clojurians_log/application.clj

@@ -4,6 +4,7 @@
             [system.components.endpoint :refer [new-endpoint]]
             [clojurians-log.components.server-info :refer [server-info]]
             [clojurians-log.components.datomic-schema :refer [new-datomic-schema]]
+            [clojurians-log.components.indexer :refer [new-indexer]]
             [system.components.handler :refer [new-handler]]
             [system.components.middleware :refer [new-middleware]]
             [system.components.http-kit :refer [new-web-server]]
@@ -38,7 +39,9 @@
    :server-info (server-info (:port http))
    :datomic (new-datomic-db (:uri datomic))
    :datomic-schema (-> (new-datomic-schema)
-                       (component/using [:datomic]))))
+                       (component/using [:datomic]))
+   :indexer (-> (new-indexer)
+                (component/using [:datomic]))))
 
 (defn -main [& [config-file]]
   (let [conf (if (and config-file (.exists (io/file config-file)))

src/clojurians_log/components/indexer.clj

@@ -0,0 +1,24 @@
+(ns clojurians-log.components.indexer
+  (:require [com.stuartsierra.component :as component]
+            [datomic.api :as d]
+            [clojurians-log.db.queries :as queries]))
+
+(defrecord Indexer [datomic]
+  component/Lifecycle
+  (start [this]
+    (let [thread
+          (Thread.
+           (fn []
+             (queries/build-indexes! (d/db (:conn datomic)))
+             (Thread/sleep 3600)
+             (recur)))]
+      (.start thread)
+      (assoc this :thread thread)))
+
+  (stop [this]
+    (when-let [thread (:thread this)]
+      (.interrupt thread))
+    (dissoc this :thread)))
+
+(defn new-indexer []
+  (map->Indexer {}))

src/clojurians_log/components/server_info.clj

@@ -8,5 +8,6 @@
     component)
   (stop [component]
     component))
+
 (defn server-info [http-port]
   (->ServerInfoPrinter http-port))

src/clojurians_log/db/queries.clj

@@ -2,23 +2,46 @@
   (:require [datomic.api :as d]
             [clojurians-log.time-util :as time-util]))
 
+(defonce !indexes (atom {}))
+
+(defn channels-dates-msgcounts [db]
+  (d/q '[:find ?slack-id ?chan-name ?day (count ?msg)
+         :in $
+         :where
+         [?chan :channel/slack-id ?slack-id]
+         [?chan :channel/name ?chan-name]
+         [?msg :message/channel ?chan]
+         [?msg :message/day ?day]]
+       db))
+
+(defn build-indexes [db]
+  (let [cdm (channels-dates-msgcounts db)]
+    (reduce
+     (fn [acc [slack-id chan-name day msgcount]]
+       (-> acc
+           (assoc-in [:chan-day-cnt slack-id day] msgcount)
+           (assoc-in [:day-chan-cnt day slack-id] msgcount)
+           (assoc-in [:chan-id->name slack-id] chan-name)
+           (assoc-in [:chan-name->id chan-name] slack-id)))
+     {}
+     cdm)))
+
+(defn build-indexes! [db]
+  (reset! !indexes (build-indexes db)))
+
 (defn channel-list
   ([db]
-   (->> (d/q '[:find [(pull ?chan [:channel/slack-id :channel/name]) ...]
-               :in $
-               :where
-               [?msg :message/channel ?chan]]
-             db)
+   (->> (map (fn [[id name]]
+               {:channel/slack-id id
+                :channel/name name})
+             (:chan-id->name @!indexes))
         (sort-by :channel/name)))
   ([db day]
-   (->> (d/q '[:find (pull ?chan [:channel/slack-id :channel/name]) (count ?msg)
-               :in $ ?day
-               :where
-               [?msg :message/day ?day]
-               [?msg :message/channel ?chan]]
-             db
-             day)
-        (map #(assoc (first %) :channel/message-count (last %))))))
+   (let [{:keys [day-chan-cnt chan-id->name]} @!indexes]
+     (for [[ch-id cnt] (get day-chan-cnt day)]
+       #:channel{:slack-id ch-id
+                 :name (chan-id->name ch-id)
+                 :message-count cnt}))))
 
 (defn- assoc-inst [message]
   (assoc message :message/inst (time-util/ts->inst (:message/ts message))))
@@ -75,15 +98,12 @@
   (compare y x))
 
 (defn channel-days [db chan-name]
-  (->> (d/q '[:find ?day (count ?msg)
-              :in $ ?chan-name
-              :where
-              [?chan :channel/name ?chan-name]
-              [?msg :message/channel ?chan]
-              [?msg :message/day ?day]]
-            db
-            chan-name)
-       (sort-by first reverse-compare)))
+  (let [{:keys [chan-day-cnt chan-name->id]} @!indexes]
+    (some->> chan-name
+             chan-name->id
+             chan-day-cnt
+             keys
+             (sort reverse-compare))))
 
 (defn channel [db name]
   (d/q '[:find (pull ?chan [*]) .
@@ -94,22 +114,14 @@
        name))
 
 (defn user-names
-  [db names]
+  [db ids]
   (d/q '[:find ?id ?username
          :in $ [?id ...]
          :where
          [?user :user/slack-id ?id]
          [?user :user/name ?username]]
        db
-       names))
-
-(defn message-by-ts [db ts]
-  (d/q '[:find (pull ?msg [*]) .
-         :in $ ?ts
-         :where
-         [?msg :message/ts ?ts]]
-       db
-       ts))
+       ids))
 
 (defn thread-messages
   "Retrieve all child messages for the given parent threads"

src/clojurians_log/db/schema.clj

@@ -30,7 +30,8 @@
    #:db{:ident       :message/thread-ts
         :valueType   :db.type/string
         :cardinality :db.cardinality/one
-        :doc         "Thread parent message timestamp (seconds since epoch up to 6 decimals). Stored as string because it is used by slack as a kind of identifier. Unique per channel."}
+        :doc         "Thread parent message timestamp (seconds since epoch up to 6 decimals). Stored as string because it is used by slack as a kind of identifier. Unique per channel."
+        :index       true}
    #:db{:ident       :message/thread-inst
         :valueType   :db.type/instant
         :cardinality :db.cardinality/one

src/clojurians_log/repl.clj

@@ -173,22 +173,45 @@
   (use 'clojurians-log.repl)
   (load-slack-data!)
   (def result (load-files! (log-files)))
-  @(second result)
   result
 
+  (def result (load-files! (drop 1508 (log-files))))
+
+  (while (not (realized? (second result)))
+    (println (java.util.Date.) "\t" @(first result))
+    (Thread/sleep 5000))
+
   ;; old way (slower)
   (run! load-log-file! (log-files))
 
   ;; incremental
-  (load-from "2016-08-04")
+  (load-from "2019-08-23")
 
 
 
   (load-demo-data! "/home/arne/github/clojurians-log-demo-data")
-
+  (build-indexes! (d/db (conn)))
 
   (do
     (write-edn "users.edn" (slack/users))
     (write-edn "channels.edn" (slack/channels)))
 
-  )
+  (time
+   (do
+     (time (clojurians-log.db.queries/channel-day-messages db "clojurescript" "2018-02-04"))
+     (time (clojurians-log.db.queries/thread-messages db '("1517722327.000023" "1517722363.000043" "1517722613.000012" "1517724278.000043" "1517724340.000044" "1517724770.000024" "1517724836.000023" "1517725105.000054")))
+     (time (clojurians-log.db.queries/channel db "clojurescript"))
+     (time (clojurians-log.db.queries/channel-list db "2018-02-04"))
+     (time (clojurians-log.db.queries/user-names db #{"U2TUBBPNU"}))
+     (time (clojurians-log.db.queries/channel-days db "clojurescript"))
+
+     nil)
+
+   "Elapsed time: 18.166254 msecs"
+   "Elapsed time: 631.458841 msecs"
+   "Elapsed time: 1.568807 msecs"
+   "Elapsed time: 16.425878 msecs"
+   "Elapsed time: 1.126005 msecs"
+   "Elapsed time: 1535.355001 msecs"
+   "Elapsed time: 2205.20762 msecs"
+   ))

src/clojurians_log/views.clj

@@ -95,15 +95,11 @@
   `offset` positions away. Returns nil if the applying the offset goes out of
   bounds."
   [channel-days today offset]
-
   (as-> channel-days $
     (map vector (range) $)
-    (some (fn [[index [a-date msg-count]]] (when (and (= a-date today)
-                                                     (not (zero? msg-count)))
-                                            index)) $)
+    (some (fn [[index a-date]] index) $)
     (+ $ offset)
-    (nth channel-days $ nil)
-    (first $)))
+    (nth channel-days $ nil)))
 
 (defn- log-page-header [{:data/keys [channel date channel-days]}]
   [:div.header