 (defn write-bq-table
   ([to options ^PCollection pcoll]
    (let [opts (assoc options :label :write-bq-table)]
-     (-> pcoll
-         (.apply (with-opts base-schema opts
-                   (write-bq-table-clj-transform to opts))))))
+     (apply-transform pcoll (write-bq-table-clj-transform to opts) write-bq-table-schema opts)))
   ([to pcoll] (write-bq-table to {} pcoll)))

           [com.google.cloud.dataflow.sdk.coders TableRowJsonCoder]))

(defn read-bq-table-raw
  ([from options p]
   (let [opts (assoc options :label :read-bq-table-raw)]
     (-> p
         (cond-> (instance? Pipeline p) (PBegin/in))
         (.apply (with-opts base-schema opts
                   (BigQueryIO$Read/from from))))))
  ([from p] (read-bq-table-raw from {} p)))

(defn table-row->clj
  [^TableRow row]

(defn write-bq-table-raw
  ([to options ^PCollection pcoll]
   (let [opts (assoc options :label :write-bq-table-raw)]
     (apply-transform pcoll (BigQueryIO$Write/to to) write-bq-table-schema opts)))
  ([to pcoll] (write-bq-table-raw to {} pcoll)))

(defn- write-bq-table-clj-transform
  [to options]
  (let [safe-opts (dissoc options :name)]
    (ptransform
     :write-bq-table-from-clj
     [^PCollection pcoll]
     (let [schema (:schema options)
           base-coll (if schema
