 
 "
   ([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"

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

(defn field-names [parsed-schema]
 (->> (.getFields parsed-schema)
       (map (fn [field]
              (let [type (.getType (.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 %) Schema$Type/RECORD))
                                                                      (map (fn [rec] (prefixed-names rec (.name field)))))]
                                               (if (empty? record-fields) [(.name field)]
                                                   record-fields))

            [pigpen.raw :as raw]
            [pigpen.core :as pig])
  (:import [org.apache.avro
            Schema
            Schema$Field
            Schema$Type
            Schema$Parser
            SchemaNormalization]
           [org.apache.avro.file DataFileReader]
           [org.apache.avro.specific SpecificDatumReader]
           [org.apache.avro.generic GenericData$Record]
           [pigpen.local PigPenLocalLoader]))

(set! *warn-on-reflection* true)
