           "Converts an individual command into the equivalent Cascading flow definition."
           (fn [{:keys [type]} flowdef] type))
 
-(defn- add-val [flowdef path id val]
-  (update-in flowdef path (partial merge {id val})))
 
 (defmethod command->flowdef :load
   [{:keys [id location storage fields opts]} flowdef]

                                                                                     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))}))))

        keys (if is-group-all [["group_all"]] keys)
        pipes (if is-group-all
                [(Each. ((:pipes flowdef) (first ancestors))
                        (Insert. (cfields ["group_all"]) (into-array [1]))
                        (cfields ["group_all" "value"]))]
                (map (:pipes flowdef) ancestors))
        pipes (map (fn [p k t] (if (= :required t)
                                 (Each. p (cfields k) (FilterNull.))
                                 p))
                   pipes keys join-types)]
    (-> flowdef
        (add-val [:pipes] id (CoGroup. (str id)
                                       (into-array Pipe pipes)
                                       (into-array (map cfields keys))
                                       Fields/NONE
                                       (BufferJoin.)))
        (add-val [:cogroup-opts] id {:group-type  :group
