             [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])
+            [cheshire.core :as json]
+            [qbits.knit :as knit])
   (:import [com.datastax.driver.core
             BatchStatement
-            PreparedStatement]))
+            PreparedStatement]
+           [java.util.concurrent ExecutorService]))
 
 (set! *warn-on-reflection* true)
 


(defn par-fetch
  "Fetch data in parallel fashion."
  [session fetch! 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 #(future
                       (try
                         (let [points (object-array asize)
                               agg-fn (agg-fn-by-path %)

        (let [min-point  (align-time from rollup)
              max-point  (align-time (apply min [to (now)]) rollup)]
          (if-let [series (and (seq paths)
                               (par-fetch session fetchq executor paths tenant
                                          rollup period min-point max-point))]
            (format "{\"from\":%s,\"to\":%s,\"step\":%s,\"series\":%s}"
                    min-point max-point rollup series)
            (json/generate-string {:from from
