Sync dynamic tables for snowflake (#36121)

docs/developers-guide/driver-changelog.md

@@ -78,6 +78,9 @@ title: Driver interface changelog
 - The multimethod `metabase.driver.sql-jdbc.execute/inject-remark` has been added. It allows JDBC-based drivers to
   override the default behavior of how SQL query remarks are added to queries (prepending them as a comment).
 
+- The arity of multimethod `metabase.driver.sql-jdbc.sync.interface/fallback-metadata-query` has been updated from 3 to 4,
+  it now takes an additional `db` argument. The new function arguments are: `[driver db-name-or-nil schema table]`.
+
 ## Metabase 0.47.0
 
 - A new driver feature has been added: `:schemas`. This feature signals whether the database organizes tables in

modules/drivers/redshift/src/metabase/driver/redshift.clj

@@ -390,7 +390,7 @@
            (into
             #{}
             (sql-jdbc.describe-table/describe-table-fields-xf driver table)
-            (sql-jdbc.describe-table/fallback-fields-metadata-from-select-query driver conn schema table-name))))))
+            (sql-jdbc.describe-table/fallback-fields-metadata-from-select-query driver conn db-name-or-nil schema table-name))))))
 
 (defmethod sql-jdbc.execute/set-parameter [:redshift java.time.ZonedDateTime]
   [driver ps i t]

modules/drivers/snowflake/src/metabase/driver/snowflake.clj

@@ -18,8 +18,8 @@
    [metabase.driver.sql-jdbc.execute.legacy-impl :as sql-jdbc.legacy]
    [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
    [metabase.driver.sql-jdbc.sync.common :as sql-jdbc.sync.common]
-   [metabase.driver.sql-jdbc.sync.describe-table
-    :as sql-jdbc.describe-table]
+   [metabase.driver.sql-jdbc.sync.describe-database :as sql-jdbc.describe-database]
+   [metabase.driver.sql-jdbc.sync.describe-table :as sql-jdbc.describe-table]
    [metabase.driver.sql.query-processor :as sql.qp]
    [metabase.driver.sql.util :as sql.u]
    [metabase.driver.sql.util.unprepare :as unprepare]
@@ -41,7 +41,8 @@
   (:import
    (java.io File)
    (java.sql Connection DatabaseMetaData ResultSet Types)
-   (java.time OffsetDateTime OffsetTime ZonedDateTime)))
+   (java.time OffsetDateTime OffsetTime ZonedDateTime)
+   (net.snowflake.client.jdbc SnowflakeSQLException)))
 
 (set! *warn-on-reflection* true)
 
@@ -408,31 +409,21 @@
   (let [db-name          (db-name database)
         excluded-schemas (set (sql-jdbc.sync/excluded-schemas driver))]
     (qp.store/with-metadata-provider (u/the-id database)
-      (let [sql             (format "SHOW OBJECTS IN DATABASE \"%s\"" db-name)
-            schema-patterns (driver.s/db-details->schema-filter-patterns "schema-filters" database)
+      (let [schema-patterns (driver.s/db-details->schema-filter-patterns "schema-filters" database)
             [inclusion-patterns exclusion-patterns] schema-patterns]
-        (log/tracef "[Snowflake] %s" sql)
         (sql-jdbc.execute/do-with-connection-with-options
          driver
          database
          nil
          (fn [^Connection conn]
-           {:tables (into
-                     #{}
-                     (comp (filter (fn [{schema :schema_name, table-name :name}]
-                                     (and (not (contains? excluded-schemas schema))
-                                          (driver.s/include-schema? inclusion-patterns
-                                                                    exclusion-patterns
-                                                                    schema)
-                                          (sql-jdbc.sync/have-select-privilege? driver conn schema table-name))))
-                           (map (fn [{schema :schema_name, table-name :name, remark :comment}]
-                                  {:name        table-name
-                                   :schema      schema
-                                   :description (not-empty remark)})))
-                     (try
-                       (jdbc/reducible-query {:connection conn} sql)
-                       (catch Throwable e
-                         (throw (ex-info (trs "Error executing query: {0}" (ex-message e)) {:sql sql} e)))))}))))))
+           {:tables (->> (sql-jdbc.describe-database/db-tables driver (.getMetaData conn) nil db-name)
+                         (filter (fn [{schema :schema table-name :name}]
+                                   (and (not (contains? excluded-schemas schema))
+                                        (driver.s/include-schema? inclusion-patterns
+                                                                  exclusion-patterns
+                                                                  schema)
+                                        (sql-jdbc.sync/have-select-privilege? driver conn schema table-name))))
+                         set)}))))))
 
 (defmethod driver/describe-table :snowflake
   [driver database table]
@@ -462,29 +453,48 @@
 ;; more context.
 (defmethod sql-jdbc.describe-table/get-table-pks :snowflake
   [_driver ^Connection conn db-name-or-nil table]
-  (let [^DatabaseMetaData metadata (.getMetaData conn)]
-    (into [] (sql-jdbc.sync.common/reducible-results
-              #(.getPrimaryKeys metadata db-name-or-nil
-                                (-> table :schema escape-name-for-metadata)
-                                (-> table :name escape-name-for-metadata))
-              (fn [^ResultSet rs] #(.getString rs "COLUMN_NAME"))))))
+  (let [^DatabaseMetaData metadata (.getMetaData conn)
+        schema-name                (-> table :schema escape-name-for-metadata)
+        table-name                 (-> table :name escape-name-for-metadata)]
+    (try
+     (into [] (sql-jdbc.sync.common/reducible-results
+               #(.getPrimaryKeys metadata db-name-or-nil
+                                 schema-name
+                                 table-name)
+               (fn [^ResultSet rs] #(.getString rs "COLUMN_NAME"))))
+     (catch SnowflakeSQLException e
+       ;; dynamic tables doesn't support pks so it's fine to suppress the exception
+       (if (= "DYNAMIC_TABLE"
+              (-> (.getTables metadata db-name-or-nil schema-name table-name nil)
+                  jdbc/result-set-seq first :table_type))
+         []
+         (throw e))))))
 
 (defn- describe-table-fks*
   "Stolen from [[sql-jdbc.describe-table]].
   The only change is that it escapes `schema` and `table-name`."
   [_driver ^Connection conn {^String schema :schema, ^String table-name :name} & [^String db-name-or-nil]]
   ;; Snowflake bug: schema and table name are interpreted as patterns
-  (let [schema (escape-name-for-metadata schema)
-        table-name (escape-name-for-metadata table-name)]
-    (into
-     #{}
-     (sql-jdbc.sync.common/reducible-results #(.getImportedKeys (.getMetaData conn) db-name-or-nil schema table-name)
-                                             (fn [^ResultSet rs]
-                                               (fn []
-                                                 {:fk-column-name   (.getString rs "FKCOLUMN_NAME")
-                                                  :dest-table       {:name   (.getString rs "PKTABLE_NAME")
-                                                                     :schema (.getString rs "PKTABLE_SCHEM")}
-                                                  :dest-column-name (.getString rs "PKCOLUMN_NAME")}))))))
+  (let [metadata    (.getMetaData conn)
+        schema-name (escape-name-for-metadata schema)
+        table-name  (escape-name-for-metadata table-name)]
+    (try
+     (into
+      #{}
+      (sql-jdbc.sync.common/reducible-results #(.getImportedKeys metadata db-name-or-nil schema-name table-name)
+                                              (fn [^ResultSet rs]
+                                                (fn []
+                                                  {:fk-column-name   (.getString rs "FKCOLUMN_NAME")
+                                                   :dest-table       {:name   (.getString rs "PKTABLE_NAME")
+                                                                      :schema (.getString rs "PKTABLE_SCHEM")}
+                                                   :dest-column-name (.getString rs "PKCOLUMN_NAME")}))))
+     (catch SnowflakeSQLException e
+       ;; dynamic tables doesn't support kks so it's fine to suppress the exception
+       (if (= "DYNAMIC_TABLE"
+              (-> (.getTables metadata db-name-or-nil schema-name table-name nil)
+                  jdbc/result-set-seq first :table_type))
+         #{}
+         (throw e))))))
 
 (defn- describe-table-fks
   "Stolen from [[sql-jdbc.describe-table]].

modules/drivers/snowflake/test/metabase/driver/snowflake_test.clj

@@ -88,8 +88,6 @@
                (ddl/create-db-tables-ddl-statements :snowflake (-> (mt/get-dataset-definition defs/test-data)
                                                                    (update :database-name #(str "v3_" %))))))))))
 
-;; TODO -- disabled because these are randomly failing, will figure out when I'm back from vacation. I think it's a
-;; bug in the JDBC driver -- Cam
 (deftest describe-database-test
   (mt/test-driver :snowflake
     (testing "describe-database"
@@ -135,6 +133,29 @@
                    (map (partial into {})
                         (t2/select [Table :name] :db_id (u/the-id database)))))))))))
 
+(deftest sync-dynamic-tables-test
+  (testing "Should be able to sync dynamic tables"
+    (mt/test-driver :snowflake
+      (mt/dataset (mt/dataset-definition "dynamic-table"
+                    ["metabase_users"
+                     [{:field-name "name" :base-type :type/Text}]
+                     [["mb_qnkhuat"]]])
+        (let [db-id   (:id (mt/db))
+              details (:details (mt/db))
+              spec    (sql-jdbc.conn/connection-details->spec driver/*driver* details)]
+          (jdbc/execute! spec [(format "CREATE OR REPLACE DYNAMIC TABLE \"%s\".\"PUBLIC\".\"metabase_fan\" target_lag = '1 minute' warehouse = 'COMPUTE_WH' AS
+                                       SELECT * FROM \"%s\".\"PUBLIC\".\"metabase_users\" WHERE \"%s\".\"PUBLIC\".\"metabase_users\".\"name\" LIKE 'MB_%%';"
+                                       (:db details) (:db details) (:db details))])
+          (sync/sync-database! (t2/select-one :model/Database db-id))
+          (testing "both base tables and dynamic tables should be synced"
+            (is (= #{"metabase_fan" "metabase_users"}
+                   (t2/select-fn-set :name :model/Table :db_id db-id)))
+            (testing "the fields for dynamic tables are synced correctly"
+              (is (= #{{:name "name" :base_type :type/Text}
+                       {:name "id" :base_type :type/Number}}
+                     (set (t2/select [:model/Field :name :base_type]
+                                     :table_id (t2/select-one-pk :model/Table :name "metabase_fan" :db_id db-id))))))))))))
+
 (deftest describe-table-test
   (mt/test-driver :snowflake
     (testing "make sure describe-table uses the NAME FROM DETAILS too"

modules/drivers/snowflake/test/metabase/test/data/snowflake.clj

@@ -42,7 +42,7 @@
   because it relies on [[public-settings/site-uuid]]."
   #'sql.tu.unique-prefix/unique-prefix)
 
-(defn- qualified-db-name
+(defn qualified-db-name
   "Prepend `database-name` with the [[*database-prefix-fn*]] so we don't stomp on any other jobs running at the same
   time."
   [database-name]

modules/drivers/sqlite/src/metabase/driver/sqlite.clj

@@ -109,9 +109,9 @@
 ;; The normal SELECT * FROM table WHERE 1 <> 1 LIMIT 0 query doesn't return any information for SQLite views -- it
 ;; seems to be the case that the query has to return at least one row
 (defmethod sql-jdbc.sync/fallback-metadata-query :sqlite
-  [driver schema table]
+  [driver _db-name-or-nil _schema-name table-name]
   (sql.qp/format-honeysql driver {:select [:*]
-                                  :from   [[(h2x/identifier :table schema table)]]
+                                  :from   [[(h2x/identifier :table table-name)]]
                                   :limit  1}))
 
 (defn- ->date [& args]

src/metabase/driver/sql_jdbc/sync.clj

@@ -11,7 +11,7 @@
 
 (comment sql-jdbc.dbms-version/keep-me sql-jdbc.sync.interface/keep-me sql-jdbc.describe-database/keep-me sql-jdbc.describe-table/keep-me)
 
- #_{:clj-kondo/ignore [:deprecated-var]}
+#_{:clj-kondo/ignore [:deprecated-var]}
 (p/import-vars
  [sql-jdbc.sync.interface
   active-tables

src/metabase/driver/sql_jdbc/sync/describe_database.clj

@@ -99,13 +99,13 @@
           (.rollback conn))
         false))))
 
-(defn- db-tables
+(defn db-tables
   "Fetch a JDBC Metadata ResultSet of tables in the DB, optionally limited to ones belonging to a given
   schema. Returns a reducible sequence of results."
   [driver ^DatabaseMetaData metadata ^String schema-or-nil ^String db-name-or-nil]
   (with-open [rset (.getTables metadata db-name-or-nil (some->> schema-or-nil (driver/escape-entity-name-for-metadata driver)) "%"
                                (into-array String ["TABLE" "PARTITIONED TABLE" "VIEW" "FOREIGN TABLE" "MATERIALIZED VIEW"
-                                                   "EXTERNAL TABLE"]))]
+                                                   "EXTERNAL TABLE" "DYNAMIC_TABLE"]))]
     (loop [acc []]
       (if-not (.next rset)
         acc
@@ -127,7 +127,7 @@
     (eduction
      (comp (mapcat (fn [schema]
                      (db-tables driver metadata schema db-name-or-nil)))
-           (filter (fn [{table-schema :schema, table-name :name}]
+           (filter (fn [{table-schema :schema table-name :name}]
                      (sql-jdbc.sync.interface/have-select-privilege? driver conn table-schema table-name))))
      (sql-jdbc.sync.interface/filtered-syncable-schemas driver conn metadata
                                                         schema-inclusion-filters schema-exclusion-filters))))

src/metabase/driver/sql_jdbc/sync/describe_table.clj

@@ -67,12 +67,12 @@
     semantic-type))
 
 (defmethod sql-jdbc.sync.interface/fallback-metadata-query :sql-jdbc
-  [driver schema table]
-  {:pre [(string? table)]}
+  [driver db-name-or-nil schema-name table-name]
+  {:pre [(string? table-name)]}
   ;; Using our SQL compiler here to get portable LIMIT (e.g. `SELECT TOP n ...` for SQL Server/Oracle)
   (sql.qp/with-driver-honey-sql-version driver
     (let [honeysql {:select [:*]
-                    :from   [(sql.qp/maybe-wrap-unaliased-expr (sql.qp/->honeysql driver (hx/identifier :table schema table)))]
+                    :from   [(sql.qp/maybe-wrap-unaliased-expr (sql.qp/->honeysql driver (hx/identifier :table db-name-or-nil schema-name table-name)))]
                     :where  [:not= (sql.qp/inline-num 1) (sql.qp/inline-num 1)]}
           honeysql (sql.qp/apply-top-level-clause driver :limit honeysql {:limit 0})]
       (sql.qp/format-honeysql driver honeysql))))
@@ -80,9 +80,9 @@
 (defn fallback-fields-metadata-from-select-query
   "In some rare cases `:column_name` is blank (eg. SQLite's views with group by) fallback to sniffing the type from a
   SELECT * query."
-  [driver ^Connection conn table-schema table-name]
+  [driver ^Connection conn db-name-or-nil schema table]
   ;; some DBs (:sqlite) don't actually return the correct metadata for LIMIT 0 queries
-  (let [[sql & params] (sql-jdbc.sync.interface/fallback-metadata-query driver table-schema table-name)]
+  (let [[sql & params] (sql-jdbc.sync.interface/fallback-metadata-query driver db-name-or-nil schema table)]
     (reify clojure.lang.IReduceInit
       (reduce [_ rf init]
         (with-open [stmt (sql-jdbc.sync.common/prepare-statement driver conn sql params)
@@ -90,8 +90,11 @@
           (let [metadata (.getMetaData rs)]
             (reduce
              ((map (fn [^Integer i]
-                     {:name          (.getColumnName metadata i)
-                      :database-type (.getColumnTypeName metadata i)})) rf)
+                     ;; TODO: missing :database-required column as ResultSetMetadata does not have information about
+                     ;; the default value of a column, so we can't make sure whether a column is required or not
+                     {:name                       (.getColumnName metadata i)
+                      :database-type              (.getColumnTypeName metadata i)
+                      :database-is-auto-increment (.isAutoIncrement metadata i)})) rf)
              init
              (range 1 (inc (.getColumnCount metadata))))))))))
 
@@ -117,10 +120,10 @@
              column-name        (.getString rs "COLUMN_NAME")
              required?          (and no-default? not-nullable? no-auto-increment?)]
          (merge
-           {:name                      column-name
-            :database-type             (.getString rs "TYPE_NAME")
+           {:name                       column-name
+            :database-type              (.getString rs "TYPE_NAME")
             :database-is-auto-increment auto-increment?
-            :database-required         required?}
+            :database-required          required?}
            (when-let [remarks (.getString rs "REMARKS")]
              (when-not (str/blank? remarks)
                {:field-comment remarks})))))))
@@ -137,8 +140,13 @@
       ;;
       ;; 3. Filter out any duplicates between the two methods using `m/distinct-by`.
       (let [has-fields-without-type-info? (volatile! false)
+            ;; intented to fix syncing dynamic tables for snowflake.
+            ;; currently there is a bug in snowflake jdbc (snowflake#1574) in which it doesn't return columns for dynamic tables
+            jdbc-returns-no-field?        (volatile! true)
             jdbc-metadata                 (eduction
                                            (remove (fn [{:keys [database-type]}]
+                                                     (when @jdbc-returns-no-field?
+                                                       (vreset! jdbc-returns-no-field? false))
                                                      (when (str/blank? database-type)
                                                        (vreset! has-fields-without-type-info? true)
                                                        true)))
@@ -148,8 +156,8 @@
                                               (reduce
                                                rf
                                                init
-                                               (when @has-fields-without-type-info?
-                                                 (fallback-fields-metadata-from-select-query driver conn schema table-name)))))]
+                                               (when (or @jdbc-returns-no-field? @has-fields-without-type-info?)
+                                                 (fallback-fields-metadata-from-select-query driver conn db-name-or-nil schema table-name)))))]
         ;; VERY IMPORTANT! DO NOT REWRITE THIS TO BE LAZY! IT ONLY WORKS BECAUSE AS NORMAL-FIELDS GETS REDUCED,
         ;; HAS-FIELDS-WITHOUT-TYPE-INFO? WILL GET SET TO TRUE IF APPLICABLE AND THEN FALLBACK-FIELDS WILL RUN WHEN
         ;; IT'S TIME TO START EVALUATING THAT.

src/metabase/driver/sql_jdbc/sync/interface.clj

@@ -72,9 +72,9 @@
   overriden because SQLite is silly and only returns column information for views if the query returns a non-zero
   number of rows.
 
-    (fallback-metadata-query :postgres \"public\" \"my_table\")
-    ;; -> [\"SELECT * FROM public.my_table WHERE 1 <> 1 LIMIT 0\"]"
-  {:added "0.37.1" :arglists '([driver schema table])}
+    (fallback-metadata-query :postgres \"my_database\" \"public\" \"my_table\")
+    ;; -> [\"SELECT * FROM my_database.public.my_table WHERE 1 <> 1 LIMIT 0\"]"
+  {:added "0.37.1" :arglists '([driver db-name-or-nil schema-name table-name])}
   driver/dispatch-on-initialized-driver
   :hierarchy #'driver/hierarchy)