     (set/subset? (set (map #(select-keys % [:id :type]) imported-columns))
                  (set (map #(select-keys % [:id :type]) columns)))))
 
-(defn do-update [conn config dataset-id data-source-id job-execution-id data-source-spec]
-  (with-open [importer (import/dataset-importer (get data-source-spec "source") config)]
-    (let [initial-dataset-version (initial-dataset-version-to-update-by-dataset-id conn {:dataset-id dataset-id})
-          imported-dataset-columns (vec (:columns initial-dataset-version))
-          importer-columns (p/columns importer)]
-      (if-not (compatible-columns? imported-dataset-columns importer-columns)
-        (failed-update conn job-execution-id "Column mismatch")
-        (let [table-name (util/gen-table-name "ds")
-              imported-table-name (util/gen-table-name "imported")]
-          (postgres/create-dataset-table conn table-name importer-columns)
-          (doseq [record (map postgres/coerce-to-sql (p/records importer))]
-            (jdbc/insert! conn table-name record))
-          (clone-data-table conn {:from-table table-name
-                                  :to-table imported-table-name}
-                            {}
-                            {:transaction? false})
-          (let [dataset-version (latest-dataset-version-by-dataset-id conn {:dataset-id dataset-id})
-                coerce-column-fn (fn [{:keys [title id type key multiple-id multiple-type] :as column}]
-                                   (cond-> {"type" (name type)
-                                            "title" title
-                                            "columnName" (name id)
-                                            "sort" nil
-                                            "direction" nil
-                                            "hidden" false}
-                                     key           (assoc "key" (boolean key))
-                                     multiple-type (assoc "multipleType" multiple-type)
-                                     multiple-id   (assoc "multipleId" multiple-id)))
-                importer-columns (mapv coerce-column-fn importer-columns)]
-            (engine/apply-transformation-log conn
-                                             table-name
-                                             imported-table-name
-                                             importer-columns
-                                             imported-dataset-columns
-                                             dataset-id
-                                             job-execution-id
-                                             dataset-version)
-            (successful-update conn
-                               job-execution-id
-                               dataset-id
-                               table-name
-                               imported-table-name
-                               dataset-version)))))))
+(defn- do-update [tenant-conn config dataset-id data-source-id job-execution-id data-source-spec]
+  (jdbc/with-db-transaction [conn tenant-conn]
+    (with-open [importer (import/dataset-importer (get data-source-spec "source") config)]
+      (let [initial-dataset-version  (initial-dataset-version-to-update-by-dataset-id conn {:dataset-id dataset-id})
+            imported-dataset-columns (vec (:columns initial-dataset-version))
+            importer-columns         (p/columns importer)]
+        (if-not (compatible-columns? imported-dataset-columns importer-columns)
+          (failed-update conn job-execution-id "Column mismatch")
+          (let [table-name          (util/gen-table-name "ds")
+                imported-table-name (util/gen-table-name "imported")]
+            (postgres/create-dataset-table conn table-name importer-columns)
+            (doseq [record (map postgres/coerce-to-sql (p/records importer))]
+              (jdbc/insert! conn table-name record))
+            (clone-data-table conn {:from-table table-name
+                                    :to-table   imported-table-name}
+                              {}
+                              {:transaction? false})
+            (let [dataset-version  (latest-dataset-version-by-dataset-id conn {:dataset-id dataset-id})
+                  coerce-column-fn (fn [{:keys [title id type key multiple-id multiple-type] :as column}]
+                                     (cond-> {"type" (name type)
+                                              "title" title
+                                              "columnName" (name id)
+                                              "sort" nil
+                                              "direction" nil
+                                              "hidden" false}
+                                       key           (assoc "key" (boolean key))
+                                       multiple-type (assoc "multipleType" multiple-type)
+                                       multiple-id   (assoc "multipleId" multiple-id)))
+                  importer-columns (mapv coerce-column-fn importer-columns)]
+              (engine/apply-transformation-log conn
+                                               table-name
+                                               imported-table-name
+                                               importer-columns
+                                               imported-dataset-columns
+                                               dataset-id
+                                               job-execution-id
+                                               dataset-version)
+              (successful-update conn
+                                 job-execution-id
+                                 dataset-id
+                                 table-name
+                                 imported-table-name
+                                 dataset-version))))))))
 
 (defn update-dataset [tenant-conn config error-tracker dataset-id data-source-id data-source-spec]
   (let [job-execution-id (str (util/squuid))]

                                                      :data-source-id data-source-id})
    (future
      (try
        (jdbc/with-db-transaction [tx-conn tenant-conn]
          (do-update tx-conn
                     config
                     dataset-id
                     data-source-id
                     job-execution-id
                     data-source-spec))
        (catch Exception e
          (failed-update tenant-conn job-execution-id (.getMessage e))
          (p/track error-tracker e)

(hugsql/def-db-fns "akvo/lumen/lib/transformation.sql")
(hugsql/def-db-fns "akvo/lumen/lib/dataset.sql")

(defn- successful-update
  "On a successful update we need to create a new dataset-version that
  is similar to the previous one, except with an updated :version and
  pointing to the new table-name, imported-table-name and columns. We
