                                        :tcp-no-delay? false}})
                     (alia/connect keyspace))
         insert! (insertq session)
-        fetch!  (fetchqp session)]
+        fetch!  (fetchqp session)
+        executor (knit/executor :cached)]
     (reify
       Metricstore
       (channel-for [this]

            [clojure.tools.logging       :refer [error info debug]]
            [lamina.core                 :refer [channel receive-all]]
            [clojure.core.async :as async :refer [<! >! go chan]]
            [cheshire.core :as json])
  (:import [com.datastax.driver.core
            BatchStatement
            PreparedStatement]))

(set! *warn-on-reflection* true)



(defn par-fetch
  "Fetch data in parallel fashion."
  [session fetch! executor paths tenant rollup period from to]
  (let [rollup (int rollup)
        from (long from)
        to (long to)
        asize (inc (time-to-ndx from rollup to))
        futures
        (doall (map #(knit/future executor
                       (try
                         (let [points (object-array asize)
                               agg-fn (agg-fn-by-path %)
