     (UserDefinedFunction/inline udf)))
 
 (defn query-job
-  [service query {:keys [create-disposition write-disposition large-results? dry-run? destination-table use-cache? flatten-results? priority udfs]}]
-  (let [{:keys [project-id dataset-id table-id]} destination-table
-        priorities {:batch       (QueryJobConfiguration$Priority/BATCH)
+  [service query {:keys [create-disposition write-disposition large-results? dry-run? destination-table default-dataset use-cache? flatten-results? priority udfs]}]
+  (let [priorities {:batch       (QueryJobConfiguration$Priority/BATCH)
                     :interactive (QueryJobConfiguration$Priority/INTERACTIVE)}
         builder    (QueryJobConfiguration/builder query)]
+    (when default-dataset
+      (let [{:keys [project-id dataset-id]} default-dataset]
+        (.defaultDataset builder (DatasetId/of project-id dataset-id))))
     (.createDisposition builder (create-dispositions (or create-disposition :never)))
     (.writeDisposition builder (write-dispositions (or write-disposition :append)))
     (.allowLargeResults builder large-results?)

(defn job [service {:keys [project-id job-id] :as job}]
  (to-clojure (.getJob service (JobId/of project-id job-id) (into-array BigQuery$JobOption []))))

(defn completed? [job]
  (and (= :done (get-in job [:status :state]))
       (empty? (get-in job [:status :errors]))))

(def create-dispositions {:needed JobInfo$CreateDisposition/CREATE_IF_NEEDED
                          :never  JobInfo$CreateDisposition/CREATE_NEVER})

    (when udfs
      (.userDefinedFunctions builder udfs))
    (when destination-table
      (let [{:keys [project-id dataset-id table-id]} destination-table]
        (.destinationTable builder (TableId/of project-id dataset-id table-id))))
    (when-not (nil? dry-run?)
      (.dryRun builder dry-run?))
    (let [query-config (.build builder)]
