 (defn write-bq-table-raw
   ([to options ^PCollection pcoll]
    (let [opts (assoc options :label :write-bq-table-raw)]
-     (-> pcoll
-         (.apply (with-opts write-bq-table-schema opts
-                   (BigQueryIO$Write/to to))))))
+     (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
-     (:name options)
+     :write-bq-table-from-clj
      [^PCollection pcoll]
      (let [schema (:schema options)
            base-coll (if schema

(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))))))
  ([to pcoll] (write-bq-table to {} pcoll)))

  [from options]
  (let [safe-opts (dissoc options :name)]
    (ptransform
     :read-bq-table-to-clj
     [p]
     (->> p
          (read-bq-table-raw safe-opts)
          (dmap table-row->clj safe-opts)))))

(defn read-bq-table
  [options ^Pipeline p]
  (let [opts (assoc options :label :read-bq-table)]
    (apply-transform p (read-bq-table-clj-transform opts) :base-schema opts)))

(defn ->schema
  ^TableSchema
