     (graph->local-group command data)))
 
 (defmethod graph->local :join [{:keys [ancestors keys join-types fields]} data]
-  (->
-    ^List
-    (mapv (fn [a k ^Observable d]
-            (.map d (rx-interop/fn [values]
-                      ;; This selects all of the fields that are in this relation
-                      {:values (into {} (for [[[r v] :as f] fields
-                                              :when (= r a)]
-                                          [f (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 a})))
-         ancestors keys data)
-    (Observable/merge)
-    (.groupBy (rx-interop/fn* :key))
-    (.flatMap (rx-interop/fn [^Observable key-grouping]
-                (-> key-grouping
-                  (.groupBy (rx-interop/fn* :relation))
-                  (.flatMap (rx-interop/fn [%]
-                              (-> ^Observable %
-                                  (.toList)
-                                  (.map (rx-interop/fn [v]
-                                          [(.getKey ^GroupedObservable %)
-                                           (map :values v)])))))
-                  (.reduce (->>
-                             ;; This seeds the inner/outer joins, by placing a
-                             ;; defualt empty value for inner joins
-                             (zipmap ancestors join-types)
-                             (filter (fn [[_ j]] (= j :required)))
-                             (map (fn [[a _]] [a []]))
-                             (into {}))
-                           (rx-interop/fn [values [k v]]
-                             (assoc values k v)))
-                  (.flatMap (rx-interop/fn [relation-grouping]
-                              (rx/seq->o (cross-product (vals relation-grouping))))))))))
+  (->>
+    (zipv [a ancestors
+           k keys
+           d data]
+      (->> d
+        (rx/map (fn [values]
+                  ;; This selects all of the fields that are in this relation
+                  {:values (into {} (for [[[r v] :as f] fields
+                                          :when (= r a)]
+                                      [f (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 a}))))
+    (apply rx/merge)
+    (rx/group-by :key)
+    (rx/flatmap (fn [[_ key-grouping]]
+                  (->> key-grouping
+                    (rx/group-by :relation)
+                    (rx/flatmap (fn [[k o]]
+                                  (->> o
+                                    (rx/into [])
+                                    (rx/map (fn [v]
+                                              [k (map :values v)])))))
+                    (rx/reduce (fn [values [k v]]
+                                 (assoc values k v))
+                               (->>
+                                 ;; This seeds the inner/outer joins, by placing a
+                                 ;; defualt empty value for inner joins
+                                 (zipmap ancestors join-types)
+                                 (filter (fn [[_ j]] (= j :required)))
+                                 (map (fn [[a _]] [a []]))
+                                 (into {})))
+                    (rx/flatmap (fn [relation-grouping]
+                                  (rx/seq->o (cross-product (vals relation-grouping))))))))))
 
 ;; ********** Script **********
 
 (defmethod graph->local :script [_ data]
-  (Observable/merge ^List (vec data)))
+  (apply rx/merge (vec data)))

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

;; unsubscribe doesn't seem to work with this version of rx
;; This caps the maximum number of records read from a file/stream
(def ^:dynamic *max-load-records* nil)

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

(defmethod graph->local :store


(defn dereference-all ^Observable [^Observable o]
  (rx/map #(->> %
             (map (fn [[k v]]
                    [k (dereference v)]))
             (into {}))
          o))

(defn observable->clj [^Observable o]
