         :desc (int (- r))))))
 
 (defmethod graph->local :order [{:keys [sort-keys]} data]
-  (let [^Observable data (first data)]
-    (-> data
-      (.toSortedList (rx-interop/fn* (partial pig-compare sort-keys)))
-      (.flatMap (rx-interop/fn* rx/seq->o)))))
+  (->> (first data)
+    (rx/sort (partial pig-compare sort-keys))))
 
 (defmethod graph->local :rank [{:keys [sort-keys]} data]
-  (let [^Observable data (first data)]
-    (if (not-empty sort-keys)
-      (-> data
-        (.toSortedList (rx-interop/fn* (partial pig-compare sort-keys)))
-        (.flatMap (rx-interop/fn* #(rx/seq->o (map-indexed (fn [i v] (assoc v '$0 i)) %)))))
-      (let [i (atom -1)]
-        (->> data
-             (rx/map (fn [v] (assoc v '$0 (swap! i inc)))))))))
+  (cond->> (first data)
+    (not-empty sort-keys) (rx/sort (partial pig-compare sort-keys))
+    true                  (rx/map-indexed (fn [i v] (assoc v '$0 i)))))
 
 ;; ********** Filter **********
 
 (defmethod graph->local :filter [{:keys [code]} data]
-  (let [^Observable data (first data)]
-    (rx/filter (fn [values]
-                 (eval-code code values))
-               data)))
+  (rx/filter (partial eval-code code) (first data)))
 
 (defmethod graph->local :filter-native [{:keys [fields expr]} data]
-  (if-not expr (first data)
-    (let [^Observable data (first data)
-          f (eval `(fn [{:syms ~fields}] ~expr))]
-      (rx/filter (fn [values] (f values))
-                 data))))
+  (if-not expr
+    (first data)
+    (let [f (eval `(fn [{:syms ~fields}] ~expr))]
+      (rx/filter f (first data)))))
 
 (defmethod graph->local :distinct [_ data]
-  (let [^Observable data (first data)
-        seen (atom [false #{}])]
-    (rx/filter (fn [values]
-                 (let [[c _] (swap! seen (fn [[_ s]] [(contains? s values)
-                                                      (conj s values)]))]
-                   (not c)))
-               data)))
+  (rx/distinct (first data)))
 
 (defmethod graph->local :limit [{:keys [n]} data]
-  (let [^Observable data (first data)]
-    (rx/take n data)))
+  (rx/take n (first data)))
 
 (defmethod graph->local :sample [{:keys [p]} data]
-  (let [^Observable data (first data)]
-    (rx/filter (fn [values] (< (rand) p))
-               data)))
+  (rx/filter (fn [_] (< (rand) p)) (first data)))
 
 ;; ********** Join **********
 
 (defmethod graph->local :union [_ ^List data]
   (apply rx/merge data))
 
+;; TODO break this up. This is way too big
 (defn ^:private graph->local-group [{:keys [ancestors keys join-types fields]} data]
-  (->
-    ^List
-    (mapv (fn [a k ^Observable d j]
-            (.flatMap d (rx-interop/fn [values]
-                          ;; This selects all of the fields that are in this relation
-                          (let [^Iterable result
-                                (for [[[r] v :as f] (next fields)
-                                      :when (= r a)]
-                                  {:values (pig/tuple (values v))
-                                   ;; This is to emulate the way pig handles nils
-                                   ;; This changes a nil values into a relation specific nil value
-                                   :key (mapv #(or (values %) (keyword (name a) "nil")) k)
-                                   :relation f
-                                   :required (= j :required)})]
-                            (rx/seq->o result)))))
-         ancestors keys data join-types)
-    (Observable/merge)
-    (.groupBy (rx-interop/fn* :key))
-    (.flatMap (rx-interop/fn [^Observable o]
-                (-> ^Observable o
-                  (.groupBy (rx-interop/fn* :relation))
-                  (.flatMap (rx-interop/fn [%]
-                              (-> ^Observable %
-                                   (.toList)
-                                   (.map (rx-interop/fn [v]
-                                             [(if (keyword? (.getKey ^GroupedObservable %))
-                                                nil
-                                                (.getKey ^GroupedObservable %))
-                                              (apply pig/bag (mapv :values v))])))))
-                  ;; Start with the group key. If it's a single value, flatten it.
-                  ;; Keywords are the fake nils we put in earlier
-                  (.reduce {(first fields) (let [k (.getKey ^GroupedObservable o)
-                                                 k (if (= 1 (count k)) (first k) k)
-                                                 k (if (keyword? k) nil k)]
-                                             k)}
-                    (rx-interop/fn [values [k v]]
-                      (assoc values k v))))))
+  (->>
+    (zipv [a ancestors
+           k keys
+           d data
+           j join-types]
+      (->> d
+        (rx/flatmap
+          (fn [values]
+            ;; This selects all of the fields that are in this relation
+            (rx/seq->o
+              (for [[[r] v :as f] (next fields)
+                    :when (= r a)]
+                {:values (pig/tuple (values v))
+                 ;; This is to emulate the way pig handles nils
+                 ;; This changes a nil values into a relation specific nil value
+                 :key (mapv #(or (values %) (keyword (name a) "nil")) k)
+                 :relation f
+                 :required (= j :required)}))))))
+    (apply rx/merge)
+    (rx/group-by :key)
+    (rx/flatmap (fn [[k' o]]
+                  (->> o
+                    (rx/group-by :relation)
+                    (rx/flatmap (fn [[r o']]
+                                  (->> o'
+                                    (rx/into [])
+                                    (rx/map (fn [v]
+                                              [(if (keyword? r) nil r)
+                                               (apply pig/bag (mapv :values v))])))))
+                    (rx/reduce
+                      (fn [values [k v]]
+                        (assoc values k v))
+                      ;; Start with the group key. If it's a single value, flatten it.
+                      ;; Keywords are the fake nils we put in earlier
+                      {(first fields) (let [k k'
+                                            k (if (= 1 (count k)) (first k) k)
+                                            k (if (keyword? k) nil k)]
+                                        k)}))))
     ;; TODO This is a bad way to do inner groupings
-    (.filter (rx-interop/fn [g]
-               (every? identity
-                 (map (fn [a [k] j]
-                        (or (= j :optional)
-                            (contains? g [[a] k])))
-                      ancestors
-                      keys
-                      join-types))))))
+    (rx/filter (fn [g]
+                 (every? identity
+                         (zipv [a ancestors
+                                [k] keys
+                                j join-types]
+                           (or (= j :optional)
+                               (contains? g [[a] k]))))))))
 
 (defn ^:private graph->local-group-all [{:keys [fields]} data]
-  (->
-    ^List
-    (mapv (fn [[r v] ^Observable d]
-            (.map d (rx-interop/fn [values]
-                      ;; TODO clean up pig dereferencing
-                      (let [v' (v values)]
-                        {:values (if (instance? Tuple v') v' (pig/tuple v'))
-                         :relation [r v]}))))
-         (next fields) data)
-    (Observable/merge)
-    (.groupBy (rx-interop/fn* :relation))
-    (.flatMap (rx-interop/fn [^Observable %]
-                (-> %
-                    (.toList)
-                    (.map (rx-interop/fn [v]
-                            [(.getKey ^GroupedObservable %)
-                             (apply pig/bag (mapv :values v))])))))
-    (.reduce {(first fields) nil}
-      (rx-interop/fn [values [k v]]
-        (assoc values k v)))
-    (.filter (rx-interop/fn [v]
-               (< 1 (count v))))))
+  (->>
+    (zipv [[r v] (next fields)
+           d data]
+      (->> d
+        (rx/map (fn [values]
+                  ;; TODO clean up pig dereferencing
+                  (let [v' (v values)]
+                    {:values (if (instance? Tuple v') v' (pig/tuple v'))
+                     :relation [r v]})))))
+    (apply rx/merge)
+    (rx/group-by :relation)
+    (rx/flatmap (fn [[r o]]
+                  (->> o
+                    (rx/into [])
+                    (rx/map (fn [v]
+                              [r (apply pig/bag (mapv :values v))])))))
+    (rx/reduce (fn [values [k v]]
+                 (assoc values k v))
+               {(first fields) nil})
+    (rx/filter (fn [v] (< 1 (count v))))))
 
 (defmethod graph->local :group [{:keys [keys] :as command} data]
   (if (= keys [:pigpen.raw/group-all])

(defn observable->data
  ^Observable [^Observable o]
  (->> o
       (dereference-all)
       (rx/map (comp 'value pig/thaw-values))
       observable->clj))

(defn observable->raw-data
  ^Observable [^Observable o]
  (->> o
       (dereference-all)
       (rx/map pig/thaw-anything)
       observable->clj))

;; ********** Util **********


  (write [this writer value])
  (close-writer [this writer]))

(defmethod graph->local :load
  [{:keys [location], :as command} _]
  (let [local-loader (load command)
        ^Observable o (->> (rx/observable*
                             (fn [^Subscriber s]
                               (let [cancel (atom false)
                                     read-count (atom 0)]
                                 (.add s
                                   (reify Subscription
                                     (unsubscribe [this]
                                       (reset! cancel true))))
                                 (future
                                   (try
                                     (println "Start reading from " location)
                                     (doseq [file (locations local-loader)
                                             :while (not @cancel)]
                                       (let [reader (init-reader local-loader file)]
                                         (doseq [value (read local-loader reader)
                                                 :while (not @cancel)]
                                           (swap! read-count inc)
                                           (rx/on-next s value))
                                         (close-reader local-loader reader)))
                                     (rx/on-completed s)
                                     (catch Exception e (rx/on-error s e)))))))
                        (rx/finally
                          (println "Stop reading from " location)))]
    (.observeOn o (Schedulers/io))))

(defmethod graph->local :store
