   ;(let [pig-field (format-field field)
   ;      pig-schema (str " AS " alias)]
   ;  [nil (str pig-field pig-schema)])
-  (println "flowdef" flowdef)
-  ;(update-in flowdef [:pipes pipe :fields] (fn [fields] (assoc fields field alias))))
-  (println "updated" (update-in flowdef [:pipes pipe] #(Each. % (Identity.))))
-  (println "x" x)
-  (println "pipe" (get-in flowdef [:pipes pipe]))
+  ;(println "flowdef" flowdef)
+  ;(println "updated" (update-in flowdef [:pipes pipe] #(Each. % (Identity.))))
+  ;(println "x" x)
+  ;(println "pipe" (get-in flowdef [:pipes pipe]))
   (update-in flowdef [:pipes pipe] #(Each. % (Identity.))))
 
 (defmethod command->flowdef :group

  (println "keys" keys)
  (println "ancestors" ancestors)
  (println "join-types" join-types)
  (update-in flowdef [:pipes] (partial merge {id (Pipe. (str id))})))

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

(defmethod command->flowdef :generate
           [{:keys [id ancestors projections opts]} flowdef]
  {:pre [id ancestors (not-empty projections)]}
  (let [new-flowdef (reduce (fn [def ancestor]
                              (cond (contains? (:sources def) ancestor) (let [pipe (Pipe. (str id))]
                                                                          (-> def
                                                                              (update-in [:pipe-to-source] (partial merge {id ancestor}))
                                                                              (update-in [:pipes] (partial merge {id pipe}))))
                                    (contains? (:pipes def) ancestor) (let [pipe ((:pipes def) ancestor)]
                                                                        (-> def
                                                                            (update-in [:pipes] (partial merge {id (Pipe. (str id) pipe)}))))
                                    :else (do
                                            (println "flowdef" flowdef)
                                            (println "id" id)
                                            (println "ancestors" ancestors)
                                            (throw (Exception. "not implemented")))))
                            flowdef ancestors
                            )]
    (reduce (fn [def cmd] (command->flowdef (assoc cmd :pipe id) def)) new-flowdef projections)))

(defmethod command->flowdef :default
