   [from options]
   (let [safe-opts (dissoc options :name)]
     (ptransform
-     (:name options)
+     :read-bq-table-to-clj
      [p]
      (->> p
-          (read-bq-table-raw from options)
-          (dmap table-row->clj options)))))
+          (read-bq-table-raw safe-opts)
+          (dmap table-row->clj safe-opts)))))
 
 (defn read-bq-table
-  ([from options ^Pipeline p]
-   (let [opts (assoc options :label :read-bq-table)]
-     (-> p
-         (.apply (with-opts base-schema opts
-                   (read-bq-table-clj-transform from opts))))))
-  ([from p] (read-bq-table from {} p)))
+  [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

(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))))))
  ([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)
     [^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)]
     (apply-transform pcoll (write-bq-table-clj-transform to opts) write-bq-table-schema opts)))
  ([to pcoll] (write-bq-table to {} pcoll)))
