   [{: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)

                              (throw (Exception. "not implemented"))))
        flowdef (let [pipe-opts (get-in flowdef [:cogroup-opts ancestor])]
                  (if pipe-opts
                    (add-val flowdef [:cogroup-opts] id pipe-opts)
                    flowdef))
        flowdef (reduce (fn [def cmd] (command->flowdef
                                        (assoc cmd :pipe id
                                               :field-projections field-projections)
