 
 "
   ([location schema]
-     (let [parsed-schema (parse-schema schema)
+     (let [^Schema parsed-schema (parse-schema schema)
            storage (raw/storage$
                     ["piggybank.jar"]
                     "org.apache.pig.piggybank.storage.avro.AvroStorage"

         {:args (clojure.core/mapcat (juxt str identity) field-symbols)
          :field-type-in :native })))))

(defn attrs-lookup [record attr-names]
  (cond
   (nil? record) nil
   (empty? attr-names) record
   :else (attrs-lookup (.get record (first attr-names)) (rest attr-names))))

(defmethod pigpen.local/load "org.apache.pig.piggybank.storage.avro.AvroStorage"
  [{:keys [location fields] }]
  (reify PigPenLocalLoader
    (locations [_] (pigpen.local/load-list location))
    (init-reader [_ filename]
      (-> filename (java.io.File.) (DataFileReader. (SpecificDatumReader.))))
    (read [_ reader]
      (for [datum reader]
        (zipmap fields (map #(attrs-lookup datum (str/split (name %) #"\.")) fields))))
    (close-reader [_ reader] (.close reader))))

  (for [subfield (field-names record-schema)]
    (str prefix "." subfield)))

(defn field-names [^Schema parsed-schema]
 (->> (.getFields parsed-schema)
       (map (fn [^Schema$Field field]
              (let [type (.getType ^{:tag Schema} (.schema field))]
                (cond
                 (= type Schema$Type/RECORD) (prefixed-names (.schema field) (.name field))
                 ;; pig only supports union of null, <type>.
                 (= type Schema$Type/UNION)  (let [record-fields (->> (.getTypes (.schema field))
                                                                      (filter #(= (.getType ^{:tag Schema} %) Schema$Type/RECORD))
                                                                      (map (fn [rec] (prefixed-names rec (.name field)))))]
                                               (if (empty? record-fields) [(.name field)]
                                                   record-fields))
