           imported-table-name (gen-table-name "imported")
           imported-columns (:columns (imported-dataset-columns-by-dataset-id conn
                                                                              {:dataset-id dataset-id}))
-          {:keys [transformations]} (latest-dataset-version-by-dataset-id conn {:dataset-id dataset-id})]
+          {:keys [transformations] :as dataset-version} (latest-dataset-version-by-dataset-id
+                                                         conn {:dataset-id dataset-id})]
       (with-open [importer (import/dataset-importer (get data-source-spec "source") config)]
         (let [columns (import/columns importer)]
           (if-not (compatible-columns? imported-columns columns)

    (java.sql.Timestamp. (.toEpochMilli value))))

(defn coerce-to-sql [record]

  (reduce-kv
   (fn [result k v]
     (assoc result k (when v (coerce v))))

   of the previous one, except with an updated :version and pointing to the new table-name
   and imported-table-name. We also delete the previous table-name and imported-table-name
   so we don't accumulate unused datasets on each update."
  [conn job-execution-id dataset-id table-name imported-table-name dataset-version]
  (insert-dataset-version conn {:id (str (squuid))
                                :dataset-id dataset-id
                                :job-execution-id job-execution-id
                                :table-name table-name
                                :imported-table-name imported-table-name
                                :version (inc (:version dataset-version))
                                :columns (vec (:columns dataset-version))
                                :transformations (vec (:transformations dataset-version))})
  (drop-table (:imported-table-name dataset-version))
  (drop-table (:table-name dataset-version))
  (update-successful-job-execution conn {:id job-execution-id}))

(defn failed-import [conn job-execution-id reason]
