CSV uploads for ClickHouse Cloud

src/metabase/driver/clickhouse.clj

@@ -1,7 +1,8 @@
 (ns metabase.driver.clickhouse
   "Driver for ClickHouse databases"
   #_{:clj-kondo/ignore [:unsorted-required-namespaces]}
-  (:require [clojure.string :as str]
+  (:require [clojure.java.jdbc :as jdbc]
+            [clojure.string :as str]
             [honey.sql :as sql]
             [metabase [config :as config]]
             [metabase.driver :as driver]
@@ -15,10 +16,9 @@
             [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
             [metabase.driver.sql.query-processor :as sql.qp]
             [metabase.driver.sql.util :as sql.u]
-            [metabase.query-processor.writeback :as qp.writeback]
-            [metabase.test.data.sql :as sql.tx]
             [metabase.upload :as upload]
-            [metabase.util.log :as log]))
+            [metabase.util.log :as log])
+  (:import [com.clickhouse.jdbc.internal ClickHouseStatementImpl]))
 
 (set! *warn-on-reflection* true)
 
@@ -142,7 +142,13 @@
   (let [parts (str/split (name s) #"\.")]
     (str/join "." (map #(str "`" % "`") parts))))
 
+(defn- cloud? [conn]
+  (let [sql "SELECT value='1' FROM system.settings WHERE name='cloud_mode'"]
+    (= 1 (val (ffirst (jdbc/query {:connection conn} sql))))))
+
 (defn- create-table!-sql
+  "Creates a ClickHouse table with the given name and column definitions. It assumes the engine is MergeTree,
+   so it only works with Clickhouse Cloud and single node on-premise deployments at the moment."
   [driver table-name column-definitions & {:keys [primary-key]}]
   (str/join "\n"
             [(first (sql/format {:create-table (keyword table-name)
@@ -157,8 +163,20 @@
 
 (defmethod driver/create-table! :clickhouse
   [driver db-id table-name column-definitions & {:keys [primary-key]}]
-  (let [sql (create-table!-sql driver table-name column-definitions :primary-key primary-key)]
-    (qp.writeback/execute-write-sql! db-id sql)))
+  (sql-jdbc.execute/do-with-connection-with-options
+   driver
+   db-id
+   {:write? true}
+   (fn [^java.sql.Connection conn]
+     (if (cloud? conn)
+       (throw (Exception. "Only supported for ClickHouse Cloud"))
+       (with-open [stmt (.createStatement conn)]
+         (let [stmt    (.unwrap stmt ClickHouseStatementImpl)
+               request (.getRequest stmt)]
+           (.set request "wait_end_of_query" 1)
+           (with-open [_response (-> request
+                                     (.query (create-table!-sql driver table-name column-definitions :primary-key primary-key))
+                                     (.executeAndWait))])))))))
 
 (defmethod driver/insert-into! :clickhouse
   [driver db-id table-name column-names values]