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

  {:pre [expr args]}
  (let [{:keys [init func]} expr
        fields (if field-projections
                 (Fields. (into-array (map #(cascading-field (:alias %)) field-projections)))
                 Fields/UNKNOWN)
        get-group-info #(cond (instance? CoGroup %) {:num-streams (alength (.getPrevious %))
                                                     :type        (if (.startsWith (.getName %) "join") :join :group)}

(defmethod command->flowdef :projection-flat
  [{:keys [code alias pipe field-projections]} flowdef]
  {:pre [code alias]}
  (command->flowdef (assoc code :pipe pipe :field-projections (or field-projections [alias])) flowdef))

(defmethod command->flowdef :generate
  [{:keys [id ancestors projections field-projections opts]} flowdef]
