           "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]

      (add-val [:pipe-to-sink] (first ancestors) id)
      (add-val [:sinks] id ((get-tap-fn storage) location opts))))

(defn- cascading-field [name-or-number]
  (if (number? name-or-number)
    (int name-or-number)
    (str name-or-number)))

(defmethod command->flowdef :code
  [{:keys [expr args pipe field-projections]} flowdef]
  {:pre [expr args]}
  (let [{:keys [init func]} expr
        fields (if field-projections
                 (Fields. (into-array (map #(cascading-field (:alias %)) field-projections)))
                 Fields/UNKNOWN)
        cogroup-opts (get-in flowdef [:cogroup-opts pipe])]
    (if-not (nil? cogroup-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
