                                        :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]

  (let [cluster (if (sequential? cluster) cluster [cluster])
        session (-> (alia/cluster
                     {:contact-points cluster
                      ;;:compression :lz4
                      :pooling-options {:max-connections-per-host
                                        {:local 8192
                                         :remote 8192}


(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 %)
