      [^PCollection pcoll]
      (let [schema (:schema options)
            base-coll pcoll]
-       (->> base-coll
-            (write-bq-table-raw to safe-opts))))))
+       (write-bq-table-raw to safe-opts base-coll)))))
 
 (defn write-bq-table
   ([to options ^PCollection pcoll]

           (cond-> mode (.setMode mode))
           (cond-> nested-fields (.setFields (clj->TableFieldSchema nested-fields transform-keys))))))

(defn ->schema
  ^TableSchema
  ([defs transform-keys]
   (if (instance? TableSchema defs)
     defs
     (let [fields (clj->TableFieldSchema defs transform-keys)]
       (-> (TableSchema.) (.setFields fields)))))
  ([defs] (->schema defs (fn [k] (name k)))))

(defn get-bq-table-schema
  "Beware, uses bq util to get the schema!"
  [table-spec]
  (let [{:keys [exit out] :as return} (sh "bq" "--format=json" "show" (name table-spec))]
    (if (= 0 exit)
      (-> (json/decode out true) (:schema) (:fields))
      (throw (ex-info (str "Could not get bq table schema for table " table-spec)
                      {:table table-spec

                                     x))
                           m)))

(defn ^TableRow clj->table-row
  [hmap]
  (let [^TableRow row (TableRow.)]
    (doseq [[k v] hmap]
      (.set row (clean-name k) (coerce-by-bq-val v)))
    row))

(defn ^TableRow clj-nested->table-row
  [hmap]
  (let [clean-map (->> hmap
                       (clojure.walk/prewalk coerce-by-bq-val)
