 (defmethod command->flowdef :load
            [{:keys [id location storage fields opts]} flowdef]
   {:pre [id location storage fields]}
-  (update-in flowdef [:sources] (partial merge {id ((get-tap-fn (:func storage)) location)})))
+  (update-in flowdef [:sources] (partial merge {id ((get-tap-fn storage) location)})))
 
 (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 (:func storage)) location)}))))
+      (update-in [:sinks] (partial merge {id ((get-tap-fn storage) location)}))))
 
 (defn- cascading-field [name-or-number]
   (if (number? name-or-number)


;; TODO: there must be a better way to pass a Tap to a storage definition.
(defn- get-tap-fn [name]
  (let [tap ({"text" (fn [location] (Hfs. (TextLine.) location))}
             name)]
    (if (nil? tap)
      (throw (Exception. (str "Unrecognized tap type: " name)))
      tap)))

(defn load-text [location]
  (-> (raw/load$ location '[offset line] (raw/storage$ [] "text" {}) {})))

(defn load-tsv
  ([location] (load-tsv location "\t"))

  (-> relation
      (raw/bind$ `(rt/map->bind ~f)
                 {:args (:fields relation), :field-type-out :native})
      (raw/store$ location :string {})))

(defn store-tsv
  ([location relation] (store-tsv location "\t" relation))
