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

                                           :fetch-size Integer/MAX_VALUE}))]
                           (when rows
                             (doseq [row rows]
                               (let [time (:time row)
                                     metric-raw (:data row)
                                     metric (if (> (count metric-raw) 1)
                                              (agg-fn metric-raw)
                                              (first metric-raw))]
                                 (aset points (to-ndx time) metric)))
                             (swap! series (fn [m] (assoc m % (vec points))))))
                         (catch  Exception e
                           (info e "Fetching exception"))))

      (throw (ex-info "Too long!" {})))
    result))

(defmacro time-to-ndx
  [from rollup time]
  `(/ (- ~time ~from) ~rollup))

(defn par-fetch
  "Fetch data in parallel fashion."
  [session fetch! paths tenant rollup period from to]
  (let [rollup (long rollup)
        from (long from)
        to (long to)
        asize (inc (time-to-ndx from rollup to))
        series (atom {})
        futures
        (doall (map #(future
