Changelog and atomicity for queries

docs/developers-guide/driver-changelog.md

@@ -9,9 +9,15 @@ title: Driver interface changelog
 - The Metabase `metabase.mbql.*` namespaces have been moved to `metabase.legacy-mbql.*`. You probably didn't need to
   use these namespaces in your driver, but if you did, please update them.
 
-- The multimethod `metabase.driver/truncate!` has been added. This method is used to delete a table's rows in the most
-  efficient way possible. This is currently only required for drivers that support the `:uploads` feature, and has
-  a default implementation for JDBC-based drivers.
+- The multimethod `metabase.driver/delete!` has been added. This method is used to delete rows from a table.
+  This is currently only required for drivers that support the `:uploads` feature, and has a default implementation for
+  JDBC-based drivers. Implementations should support transactions if relevant for the given driver.
+
+- The multimethod `metabase.driver/with-transaction*` has been added. This method is used to create a scope within
+  which querying and mutating the contents of tables will be done atomically.
+  This is currently only relevant to drivers that support the `:uploads` feature, and has a default implementation for
+  JDBC-based drivers. The method is optional, but will improve the user experience when dealing with failing or
+  concurrent operations on uploaded tables.
 
 ## Metabase 0.49.1
 

src/metabase/driver.clj

@@ -940,13 +940,26 @@
   dispatch-on-initialized-driver
   :hierarchy #'hierarchy)
 
-(defmulti truncate!
+(defmulti with-transaction*
+  "Executes the given thunk within a transaction, if the given driver supports transactions."
+  {:added "0.50.0", :arglists '([driver db-id thunk])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmethod with-transaction* ::driver [_ _ thunk]
+  (thunk))
+
+(defmacro with-transaction
+  "Executes the given body within a transaction, if the given driver supports transactions."
+  {:added "0.50.0", :arglists '([driver db-id & body])}
+  [driver db-id & body]
+  `(with-transaction* ~driver ~db-id (fn [] ~@body)))
+
+(defmulti delete!
   "Delete the current contents of `table-name`.
-  If something like a SQL TRUNCATE statement is supported, we use that, but may otherwise fall back to explicitly
-  deleting rows, or dropping and recreating the table.
-  Depending on the driver, the semantics can vary on whether triggers are fired, AUTO_INCREMENT is reset etc.
-  The application assumes that the implementation can be rolled back if inside a transaction."
-  {:added "0.50.0", :arglists '([driver db-id table-name])}
+  Currently only supports fully truncating the query, but may introduce support for restricting the scope in future.
+  If the driver supports transactions, then it must support this operation being run within a transaction."
+  {:added "0.50.0", :arglists '([driver db-id table-name & args])}
   dispatch-on-initialized-driver
   :hierarchy #'hierarchy)
 

src/metabase/driver/sql_jdbc.clj

@@ -27,11 +27,20 @@
 ;;; |                                                  Run a Query                                                   |
 ;;; +----------------------------------------------------------------------------------------------------------------+
 
+(def ^:private ^:dynamic *tx-connections* {})
+
+(defn- local-conn
+  "If called with a transaction for the given database in dynamic scope, return the corresponding connection.
+  Otherwise, return a non-transactional connection."
+  [db-id]
+  (or (get *tx-connections* db-id)
+      (sql-jdbc.conn/db->pooled-connection-spec db-id)))
+
 ;; TODO - Seems like this is only used in a handful of places, consider moving to util namespace
 (defn query
   "Execute a `honeysql-form` query against `database`, `driver`, and optionally `table`."
   ([driver database honeysql-form]
-   (jdbc/query (sql-jdbc.conn/db->pooled-connection-spec database)
+   (jdbc/query (local-conn database)
                (sql.qp/format-honeysql driver honeysql-form)))
 
   ([driver database table honeysql-form]
@@ -148,14 +157,20 @@
                                :dialect (sql.qp/quote-style driver)))]
     (qp.writeback/execute-write-sql! db-id sql)))
 
-(defmethod driver/truncate! :sql-jdbc
-  [driver db-id table-name]
+(defmethod driver/with-transaction* :sql-jdbc
+  [_ db-id thunk]
+  (jdbc/with-db-transaction [conn (local-conn db-id)]
+    (binding [*tx-connections* (assoc *tx-connections* db-id conn)]
+      (thunk))))
+
+(defmethod driver/delete! :sql-jdbc
+  [driver db-id table-name & _args]
   (let [table-name (keyword table-name)
-        sql        (sql/format {:truncate table-name}
+        sql        (sql/format {:delete-from table-name}
                                :quoted true
                                :dialect (sql.qp/quote-style driver))]
-    (jdbc/with-db-transaction [conn (sql-jdbc.conn/db->pooled-connection-spec db-id)]
-      (jdbc/execute! conn sql))))
+    (driver/with-transaction driver db-id
+      (jdbc/execute! (local-conn db-id) sql))))
 
 (defmethod driver/insert-into! :sql-jdbc
   [driver db-id table-name column-names values]
@@ -176,9 +191,9 @@
                                      :quoted true
                                      :dialect (sql.qp/quote-style driver))
                         chunks)]
-    (jdbc/with-db-transaction [conn (sql-jdbc.conn/db->pooled-connection-spec db-id)]
+    (driver/with-transaction driver db-id
       (doseq [sql sqls]
-        (jdbc/execute! conn sql)))))
+        (jdbc/execute! (local-conn db-id) sql)))))
 
 (defmethod driver/add-columns! :sql-jdbc
   [driver db-id table-name column-definitions & {:keys [primary-key]}]

src/metabase/upload.clj

@@ -807,12 +807,15 @@
                                   :num-columns       (count new-types)
                                   :generated-columns (if create-auto-pk? 1 0)
                                   :size-mb           (file-size-mb file)
-                                  :upload-seconds    (since-ms timer)}]
+                                  :upload-seconds    (since-ms timer)}
+              db-id              (:id database)
+              table-name         (table-identifier table)]
 
           (try
-            (when replace-rows?
-              (driver/truncate! driver (:id database) (table-identifier table)))
-            (driver/insert-into! driver (:id database) (table-identifier table) normed-header parsed-rows)
+            (driver/with-transaction driver db-id
+              (when replace-rows?
+                (driver/delete! driver db-id table-name))
+              (driver/insert-into! driver db-id table-name normed-header parsed-rows))
             (catch Throwable e
               (throw (ex-info (ex-message e) {:status-code 422}))))
 

test/metabase/upload_test.clj

@@ -1809,6 +1809,35 @@
                 (is (= 1 (count (rows-for-table table)))))
               (io/delete-file file))))))))
 
+(deftest replacement-failure-test
+  (mt/test-drivers (filter (fn [driver]
+                             ;; use of varchar(255) is not universal for all drivers, so only test drivers that
+                             ;; have different database types for varchar(255) and text
+                             (apply not= (map (partial driver/upload-type->database-type driver) [::upload/varchar-255 ::upload/text])))
+                           (mt/normal-drivers-with-feature :uploads))
+    (with-mysql-local-infile-off
+     ;; TODO we are abusing this detail in order to trigger a failure, and should decouple these tests in future
+     (testing "Replace fails if the CSV file contains string values that are too long for the column"
+       ;; for drivers that insert rows in chunks, we change the chunk size to 1 so that we can test that the
+       ;; inserted rows are rolled back
+       (binding [driver/*insert-chunk-rows* 1]
+         (with-upload-table!
+           [table (create-upload-table! {:col->upload-type (ordered-map/ordered-map
+                                                            upload/auto-pk-column-keyword ::upload/auto-incrementing-int-pk
+                                                            :test_column ::upload/varchar-255)
+                                         :rows             [["valid"]]})]
+           (let [csv-rows `["test_column" ~@(repeat 50 "valid too") ~(apply str (repeat 256 "x"))]
+                 file  (csv-file-with csv-rows (mt/random-name))]
+             (testing "\nShould return an appropriate error message"
+               (is (=? {;; the error message is different for different drivers, but postgres and mysql have "too long" in the message
+                        :message #"[\s\S]*too long[\s\S]*"
+                        :data    {:status-code 422}}
+                       (catch-ex-info (replace-csv! {:file     file
+                                                    :table-id (:id table)})))))
+             (testing "\nCheck the previous data is still present in the table"
+               (is (= 1 (count (rows-for-table table)))))
+             (io/delete-file file))))))))
+
 (deftest append-too-long-for-varchar-255-mysql-local-infile-test
   (mt/test-driver :mysql
     (with-mysql-local-infile-on