         pipes (map (fn [p k t] (if (= :required t)
                                  (Each. p (Fields. (into-array (map str k))) (FilterNull.))
                                  p))
-                   pipes keys join-types)
-        flowdef (update-in flowdef [:pipes] (partial merge {id (CoGroup. (str id)
-                                                                         (into-array Pipe pipes)
-                                                                         (into-array (map #(Fields. (into-array (map str %)))
-                                                                                          keys))
-                                                                         Fields/NONE
-                                                                         (BufferJoin.))}))
-        flowdef (update-in flowdef [:cogroup-opts] (partial merge {id {:group-type  :group
-                                                                       :group-all   is-group-all
-                                                                       :num-streams (count pipes)
-                                                                       :inner-join  (= #{:required} (into #{} join-types))}}))]
-    flowdef))
+                   pipes keys join-types)]
+    (-> flowdef
+        (add-val [:pipes] id (CoGroup. (str id)
+                                       (into-array Pipe pipes)
+                                       (into-array (map #(Fields. (into-array (map str %)))
+                                                        keys))
+                                       Fields/NONE
+                                       (BufferJoin.)))
+        (add-val [:cogroup-opts] id {:group-type  :group
+                                     :group-all   is-group-all
+                                     :num-streams (count pipes)
+                                     :inner-join  (= #{:required} (into #{} join-types))}))))
 
 (defmethod command->flowdef :join
   [{:keys [id keys fields join-types ancestors opts]} flowdef]

  {:pre [id (= 1 (count ancestors)) (not-empty projections)]}
  (let [ancestor (first ancestors)
        pipe ((:pipes flowdef) ancestor)
        flowdef (cond (contains? (:pipes flowdef) ancestor) (update-in flowdef [:pipes] (partial merge {id (Pipe. (str id) pipe)}))
                      :else (do
                              (println "flowdef" flowdef)
                              (println "id" id)

                 [:required :required] (InnerJoin.)
                 [:required :optional] (LeftJoin.)
                 [:optional :required] (RightJoin.)
                 [:optional :optional] (OuterJoin.))]
    (-> flowdef
        (add-val [:pipes] id (CoGroup. (str id)
                                       ;; TODO: adding an Each with Identity here is a hack around a possible bug in Cascading involving self-joins.
                                       (into-array Pipe (map-indexed (fn [i a] (let [p ((:pipes flowdef) a)
                                                                                     p (Pipe. (str (nth fields (* i 2))) p)
                                                                                     p (Each. p (Identity.))]
                                                                                 p)) ancestors))
                                       (into-array (map #(let [f (Fields. (into-array (map str %)))]
                                                          (when-not join-nils (.setComparator f (str (first %)) (NullNotEquivalentComparator.)))
                                                          f) keys))
                                       (Fields. (into-array (map str fields)))
                                       joiner))
        (add-val [:cogroup-opts] id {:group-type :join
                                     :all-args   (true? (:all-args opts))}))))

(defmethod command->flowdef :projection-flat
  [{:keys [code alias pipe field-projections]} flowdef]
