Make persisted models use "create table as" (#22640)

* Make persisted models use "create table as"

There are outstanding issues with database type erasure.

If we try to recreate table columns based on the saved
definition we'll get types such as `type/Structured` that
could be `json`, `xml`, etc. Perhaps worse, consider 'citext' that we
see as `type/Text`. If they have queries on the model such as
`where email = 'JOE@example.com'` that would start to return different
results if we created the table with `email text` rather than
`email citext`.

By using `create table ... as select ....`, we avoid this issue as the db types
selected will match the db types queried.

* Remove unused require

* Missed validatiy check needs a query to create table as

src/metabase/driver/ddl/postgres.clj

 (ns metabase.driver.ddl.postgres
   (:require [clojure.java.jdbc :as jdbc]
-            [clojure.string :as str]
             [clojure.tools.logging :as log]
             [metabase.driver.ddl.interface :as ddl.i]
             [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
@@ -20,17 +19,12 @@
     (format "create schema %s"
             (q :table (ddl.i/schema-name database (public-settings/site-uuid))))))
 
-(defn- create-table-sql [{driver :engine :as database} definition]
+(defn- create-table-sql [{driver :engine :as database} definition query]
   (let [q (quote-fn driver)]
-    (format "create table %s.%s (%s);"
+    (format "create table %s.%s as %s"
             (q :table (ddl.i/schema-name database (public-settings/site-uuid)))
             (q :table (:table-name definition))
-            (str/join
-             ", "
-             (for [{:keys [field-name base-type]} (:field-definitions definition)]
-               (format "%s %s"
-                       (q :field field-name)
-                       (ddl.i/field-base-type->sql-type driver base-type)))))))
+            query)))
 
 (defn- drop-table-sql [{driver :engine :as database} table-name]
   (let [q (quote-fn driver)]
@@ -38,26 +32,13 @@
             (q :table (ddl.i/schema-name database (public-settings/site-uuid)))
             (q :table table-name))))
 
-(defn- populate-table-sql [{driver :engine :as database} definition query]
-  (let [q (quote-fn driver)]
-   (format "insert into %s.%s (%s) %s"
-           (q :table (ddl.i/schema-name database (public-settings/site-uuid)))
-           (q :table (:table-name definition))
-           (str/join
-            ", "
-            (for [{:keys [field-name]} (:field-definitions definition)]
-              (q :field field-name)))
-           query)))
-
 (defmethod ddl.i/refresh! :postgres [_driver database definition dataset-query]
   (try
-    (jdbc/with-db-connection [conn (sql-jdbc.conn/db->pooled-connection-spec database)]
-      (jdbc/execute! conn [(drop-table-sql database (:table-name definition))])
-      (jdbc/execute! conn [(create-table-sql database definition)])
-      (jdbc/execute! conn [(populate-table-sql database definition (-> dataset-query
-                                                                       qp/compile
-                                                                       :query))])
-      {:state :success})
+    (let [{:keys [query params]} (qp/compile dataset-query)]
+      (jdbc/with-db-connection [conn (sql-jdbc.conn/db->pooled-connection-spec database)]
+        (jdbc/execute! conn [(drop-table-sql database (:table-name definition))])
+        (jdbc/execute! conn (into [(create-table-sql database definition query)] params))
+        {:state :success}))
     (catch Exception e
       {:state :error :error (ex-message e)})))
 
@@ -87,7 +68,8 @@
                                        (create-table-sql database
                                                          {:table-name table-name
                                                           :field-definitions [{:field-name "field"
-                                                                               :base-type :type/Text}]})))]
+                                                                               :base-type :type/Text}]}
+                                                         "values (1)")))]
                      [:persist.check/read-table
                       (fn read-table [conn]
                         (jdbc/query conn [(format "select * from %s.%s"

src/metabase/query_processor/middleware/fetch_source_query.clj

@@ -25,7 +25,10 @@
             [clojure.string :as str]
             [clojure.tools.logging :as log]
             [medley.core :as m]
+            [metabase.driver :as driver]
             [metabase.driver.ddl.interface :as ddl.i]
+            [metabase.driver.sql.util :as sql.u]
+            [metabase.driver.util :as driver.u]
             [metabase.mbql.normalize :as mbql.normalize]
             [metabase.mbql.schema :as mbql.s]
             [metabase.mbql.util :as mbql.u]
@@ -105,6 +108,23 @@
          (assoc m :field_ref [:field (:name m) {:base-type (:base_type m)}]))
        metadata))
 
+(defn- persisted-info-native-query [card database-id]
+  (let [driver (or driver/*driver* (driver.u/database->driver database-id))]
+    (format "select %s from %s.%s"
+            (str/join ", " (map #(sql.u/quote-name
+                                   driver
+                                   :field
+                                   (:field-name %))
+                                (get-in card [:definition :field-definitions])))
+            (sql.u/quote-name
+              driver
+              :table
+              (ddl.i/schema-name {:id database-id} (public-settings/site-uuid)))
+            (sql.u/quote-name
+              driver
+              :table
+              (:table_name card)))))
+
 (s/defn card-id->source-query-and-metadata :- SourceQueryAndMetadata
   "Return the source query info for Card with `card-id`. Pass true as the optional second arg `log?` to enable
   logging. (The circularity check calls this and will print more than desired)"
@@ -170,10 +190,7 @@
 
      (cond-> {:source-query    (cond-> source-query
                                  ;; This will be applied, if still appropriate, by the peristence middleware
-                                 persisted? (assoc :persisted-info/native (format "select %s from %s.%s"
-                                                                                  (str/join ", " (map :field-name (get-in card [:definition :field-definitions])))
-                                                                                  (ddl.i/schema-name {:id database-id} (public-settings/site-uuid))
-                                                                                  (:table_name card))))
+                                 persisted? (assoc :persisted-info/native (persisted-info-native-query card database-id)))
               :database        database-id
               :source-metadata (cond-> (seq (map mbql.normalize/normalize-source-metadata result-metadata))
                                  persisted? sub-cached-field-refs)}