            [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)))
+  [{:keys [query table] :as options} p]
+  (let [opts (assoc options :label :read-bq-table-raw)
+        ptrans (cond
+                 query (BigQueryIO$Read/fromQuery query)
+                 table (BigQueryIO$Read/from table)
+                 :else (throw (ex-info
+                               "Error with options of read-bq-table, should specify one of :table or :query"
+                               {:options options})))]
+    (-> p
+        (cond-> (instance? Pipeline p) (PBegin/in))
+        (apply-transform ptrans named-schema opts))))
 
 (defn table-row->clj
   [^TableRow row]

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

(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)))

(defn ->schema
  ^TableSchema

(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)))
