   [{:keys [id ancestors]} flowdef]
   {:pre [id (= 1 (count ancestors))]}
   (let [pipe ((:pipes flowdef) (first ancestors))]
-    (update-in flowdef [:pipes] (partial merge {id (Unique. pipe Fields/ALL)}))))
+    (add-val flowdef [:pipes] id (Unique. pipe Fields/ALL))))
 
 (defmethod command->flowdef :limit
   [{:keys [id n ancestors]} flowdef]
   {:pre [id n (= 1 (count ancestors))]}
   (let [pipe ((:pipes flowdef) (first ancestors))]
-    (update-in flowdef [:pipes] (partial merge {id (Each. pipe (Limit. n))}))))
+    (add-val flowdef [:pipes] id (Each. pipe (Limit. n)))))
 
 (defmethod command->flowdef :sample
   [{:keys [id p ancestors]} flowdef]
   {:pre [id p (= 1 (count ancestors))]}
   (let [pipe ((:pipes flowdef) (first ancestors))]
-    (update-in flowdef [:pipes] (partial merge {id (Each. pipe (Sample. p))}))))
+    (add-val flowdef [:pipes] id (Each. pipe (Sample. p)))))
 
 (defmethod command->flowdef :script
   [_ flowdef]

          "Converts an individual command into the equivalent Cascading flow definition."
          (fn [{:keys [type]} flowdef] type))

(defmethod command->flowdef :load
  [{:keys [id location storage fields opts]} flowdef]
  {:pre [id location storage fields]}
  (let [pipe (Pipe. (str id))]
    (-> flowdef
        (update-in [:sources] (partial merge {id ((get-tap-fn storage) location opts)}))
        (update-in [:pipes] (partial merge {id pipe})))))

(defmethod command->flowdef :store
  [{:keys [id ancestors location storage opts]} flowdef]
  {:pre [id ancestors location storage]}
  (-> flowdef
      (update-in [:pipe-to-sink] (partial merge {(first ancestors) id}))
      (update-in [:sinks] (partial merge {id ((get-tap-fn storage) location opts)}))))

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

        pipes (map (fn [p k t] (if (= :required t)
                                 (Each. p (Fields. (into-array (map str k))) (FilterNull.))
                                 p))
                   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]
