Faster sync-fks (#38970)

docs/developers-guide/driver-changelog.md

@@ -4,6 +4,18 @@ title: Driver interface changelog
 
 # Driver Interface Changelog
 
+## Metabase 0.50.0
+
+- The multimethod `metabase.driver/describe-table-fks` has been deprecated in favor of `metabase.driver/describe-fks`.
+  `metabase.driver/describe-table-fks` will be removed in 0.53.0.
+
+- The multimethod `metabase.driver/describe-fks` has been added. The method needs to be implemented if the database
+  supports the `:foreign-keys` and `:describe-fks` features. It replaces the `metabase.driver/describe-table-fks`
+  method, which is now deprecated.
+
+- The multimethod `metabase.driver.sql-jdbc.sync.describe-table/describe-fks-sql` has been added. The method needs
+  to be implemented if you want to use the default JDBC implementation of `metabase.driver/describe-fks`.
+
 ## Metabase 0.49.0
 
 - The multimethod `metabase.driver/alter-columns!` has been added. This method is used to alter a table's columns in the

modules/drivers/presto-jdbc/src/metabase/driver/presto_jdbc.clj

@@ -617,6 +617,7 @@
 
 ;;; The Presto JDBC driver DOES NOT support the `.getImportedKeys` method so just return `nil` here so the `:sql-jdbc`
 ;;; implementation doesn't try to use it.
+#_{:clj-kondo/ignore [:deprecated-var]}
 (defmethod driver/describe-table-fks :presto-jdbc
   [_driver _database _table]
   nil)

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

@@ -4,6 +4,7 @@
    [cheshire.core :as json]
    [clojure.java.jdbc :as jdbc]
    [clojure.string :as str]
+   [honey.sql :as sql]
    [java-time.api :as t]
    [metabase.driver :as driver]
    [metabase.driver.sql :as driver.sql]
@@ -36,6 +37,7 @@
 
 (doseq [[feature supported?] {:test/jvm-timezone-setting false
                               :nested-field-columns      false
+                              :describe-fks              true
                               :connection-impersonation  true}]
   (defmethod driver/database-supports? [:redshift feature] [_driver _feat _db] supported?))
 
@@ -49,6 +51,31 @@
   [& args]
   (apply (get-method driver/describe-table :sql-jdbc) args))
 
+(defmethod sql-jdbc.sync/describe-fks-sql :redshift
+  [driver & {:keys [schema-names table-names]}]
+  (sql/format {:select (vec
+                        {:fk_ns.nspname       "fk-table-schema"
+                         :fk_table.relname    "fk-table-name"
+                         :fk_column.attname   "fk-column-name"
+                         :pk_ns.nspname       "pk-table-schema"
+                         :pk_table.relname    "pk-table-name"
+                         :pk_column.attname   "pk-column-name"})
+               :from   [[:pg_constraint :c]]
+               :join   [[:pg_class     :fk_table]  [:= :c.conrelid :fk_table.oid]
+                        [:pg_namespace :fk_ns]     [:= :c.connamespace :fk_ns.oid]
+                        [:pg_attribute :fk_column] [:= :c.conrelid :fk_column.attrelid]
+                        [:pg_class     :pk_table]  [:= :c.confrelid :pk_table.oid]
+                        [:pg_namespace :pk_ns]     [:= :pk_table.relnamespace :pk_ns.oid]
+                        [:pg_attribute :pk_column] [:= :c.confrelid :pk_column.attrelid]]
+               :where  [:and
+                        [:= :c.contype [:raw "'f'::char"]]
+                        [:= :fk_column.attnum [:raw "ANY(c.conkey)"]]
+                        [:= :pk_column.attnum [:raw "ANY(c.confkey)"]]
+                        (when table-names [:in :fk_table.relname table-names])
+                        (when schema-names [:in :fk_ns.nspname schema-names])]
+               :order-by [:fk-table-schema :fk-table-name]}
+              :dialect (sql.qp/quote-style driver)))
+
 (defmethod driver/db-start-of-week :redshift
   [_]
   :sunday)

modules/drivers/redshift/test/metabase/driver/redshift_test.clj

@@ -226,32 +226,26 @@
             qual-tbl-nm  (format "\"%s\".\"%s\"" (redshift.test/unique-session-schema) tbl-nm)
             view-nm      "late_binding_view"
             qual-view-nm (format "\"%s\".\"%s\"" (redshift.test/unique-session-schema) view-nm)]
-        (t2.with-temp/with-temp [Database database {:engine :redshift, :details db-details}]
-          (try
-            ;; create a table with a CHARACTER VARYING and a NUMERIC column, and a late bound view that selects from it
-            (execute!
-             (str "DROP TABLE IF EXISTS %1$s;%n"
-                  "CREATE TABLE %1$s(weird_varchar CHARACTER VARYING(50), numeric_col NUMERIC(10,2));%n"
-                  "CREATE OR REPLACE VIEW %2$s AS SELECT * FROM %1$s WITH NO SCHEMA BINDING;")
-             qual-tbl-nm
-             qual-view-nm)
+        (mt/with-temp [:model/Database database {:engine :redshift, :details db-details}]
+          ;; create a table with a CHARACTER VARYING and a NUMERIC column, and a late bound view that selects from it
+          (execute!
+           (str "DROP TABLE IF EXISTS %1$s;%n"
+                "CREATE TABLE %1$s(weird_varchar CHARACTER VARYING(50), numeric_col NUMERIC(10,2));%n"
+                "CREATE OR REPLACE VIEW %2$s AS SELECT * FROM %1$s WITH NO SCHEMA BINDING;")
+           qual-tbl-nm
+           qual-view-nm)
             ;; sync the schema again to pick up the new view (and table, though we aren't checking that)
-            (sync/sync-database! database {:scan :schema})
-            (is (contains?
-                 (t2/select-fn-set :name Table :db_id (u/the-id database)) ; the new view should have been synced
-                 view-nm))
-            (let [table-id (t2/select-one-pk Table :db_id (u/the-id database), :name view-nm)]
+          (sync/sync-database! database {:scan :schema})
+          (is (contains?
+               (t2/select-fn-set :name Table :db_id (u/the-id database)) ; the new view should have been synced
+               view-nm))
+          (let [table-id (t2/select-one-pk Table :db_id (u/the-id database), :name view-nm)]
               ;; and its columns' :base_type should have been identified correctly
-              (is (= [{:name "numeric_col",   :database_type "numeric(10,2)",         :base_type :type/Decimal}
-                      {:name "weird_varchar", :database_type "character varying(50)", :base_type :type/Text}]
-                     (map
-                      mt/derecordize
-                      (t2/select [Field :name :database_type :base_type] :table_id table-id {:order-by [:name]})))))
-            (finally
-              (execute! (str "DROP TABLE IF EXISTS %s;%n"
-                             "DROP VIEW IF EXISTS %s;")
-                        qual-tbl-nm
-                        qual-view-nm))))))))
+            (is (= [{:name "numeric_col",   :database_type "numeric(10,2)",         :base_type :type/Decimal}
+                    {:name "weird_varchar", :database_type "character varying(50)", :base_type :type/Text}]
+                   (map
+                    mt/derecordize
+                    (t2/select [Field :name :database_type :base_type] :table_id table-id {:order-by [:name]}))))))))))
 
 (deftest redshift-lbv-sync-error-test
   (mt/test-driver

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

@@ -524,16 +524,15 @@
 (defn- describe-table-fks
   "Stolen from [[sql-jdbc.describe-table]].
   The only change is that it calls the stolen function [[describe-table-fks*]]."
-  [driver db-or-id-or-spec-or-conn table & [db-name-or-nil]]
-  (if (instance? Connection db-or-id-or-spec-or-conn)
-    (describe-table-fks* driver db-or-id-or-spec-or-conn table db-name-or-nil)
-    (sql-jdbc.execute/do-with-connection-with-options
-     driver
-     db-or-id-or-spec-or-conn
-     nil
-     (fn [conn]
-       (describe-table-fks* driver conn table db-name-or-nil)))))
+  [driver db-or-id-or-spec table & [db-name-or-nil]]
+  (sql-jdbc.execute/do-with-connection-with-options
+   driver
+   db-or-id-or-spec
+   nil
+   (fn [conn]
+     (describe-table-fks* driver conn table db-name-or-nil))))
 
+#_{:clj-kondo/ignore [:deprecated-var]}
 (defmethod driver/describe-table-fks :snowflake
   [driver database table]
   (describe-table-fks driver database table (db-name database)))

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

@@ -300,6 +300,7 @@
       (is (= #{{:fk-column-name   "category_id"
                 :dest-table       {:name "categories", :schema "PUBLIC"}
                 :dest-column-name "id"}}
+             #_{:clj-kondo/ignore [:deprecated-var]}
              (driver/describe-table-fks :snowflake (assoc (mt/db) :name "ABC") (t2/select-one Table :id (mt/id :venues))))))))
 
 (defn- format-env-key ^String [env-key]

src/metabase/driver.clj

@@ -339,15 +339,32 @@
 (defmethod escape-entity-name-for-metadata :default [_driver table-name] table-name)
 
 (defmulti describe-table-fks
-  "Return information about the foreign keys in a `table`. Required for drivers that support `:foreign-keys`. Results
-  should match the [[metabase.sync.interface/FKMetadata]] schema."
-  {:added "0.32.0" :arglists '([driver database table])}
+  "Return information about the foreign keys in a `table`. Required for drivers that support `:foreign-keys` but not
+  `:describe-fks`. Results should match the [[metabase.sync.interface/FKMetadata]] schema."
+  {:added "0.32.0" :deprecated "0.50.0" :arglists '([driver database table])}
   dispatch-on-initialized-driver
   :hierarchy #'hierarchy)
 
+#_{:clj-kondo/ignore [:deprecated-var]}
 (defmethod describe-table-fks ::driver [_ _ _]
   nil)
 
+(defmulti describe-fks
+  "Returns a reducible collection of maps, each containing information about foreign keys.
+  Takes keyword arguments to narrow down the results to a set of `schema-names` or `table-names`.
+
+  Results match [[metabase.sync.interface/FKMetadataEntry]].
+  Results are optionally filtered by `schema-names` and `table-names` provided.
+  Results are ordered by `fk-table-schema` and `fk-table-name` in ascending order.
+
+  Required for drivers that support `:describe-fks`."
+  {:added "0.50.0" :arglists '([driver database & {:keys [schema-names table-names]}])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmethod describe-fks ::driver [_ _]
+  nil)
+
 ;;; this is no longer used but we can leave it around for not for documentation purposes. Maybe we can actually do
 ;;; something useful with it like write a test that validates that drivers return correct connection details?
 
@@ -547,7 +564,10 @@
     :native-requires-specified-collection
 
     ;; Does the driver support column(s) support storing index info
-    :index-info})
+    :index-info
+
+    ;; Does the driver support a faster `sync-fks` step by fetching all FK metadata in a single collection?
+    :describe-fks})
 
 (defmulti database-supports?
   "Does this driver and specific instance of a database support a certain `feature`?

src/metabase/driver/sql_jdbc.clj

@@ -93,10 +93,15 @@
   [driver database table]
   (sql-jdbc.sync/describe-table driver database table))
 
+#_{:clj-kondo/ignore [:deprecated-var]}
 (defmethod driver/describe-table-fks :sql-jdbc
   [driver database table]
   (sql-jdbc.sync/describe-table-fks driver database table))
 
+(defmethod driver/describe-fks :sql-jdbc
+  [driver database & {:as args}]
+  (sql-jdbc.sync/describe-fks driver database args))
+
 (defmethod driver/describe-table-indexes :sql-jdbc
   [driver database table]
   (sql-jdbc.sync/describe-table-indexes driver database table))

src/metabase/driver/sql_jdbc/execute.clj

@@ -714,6 +714,32 @@
               results-metadata {:cols (column-metadata driver rsmeta)}]
           (respond results-metadata (reducible-rows driver rs rsmeta qp.pipeline/*canceled-chan*))))))))
 
+(defn reducible-query
+  "Returns a reducible collection of rows as maps from `db` and a given SQL query. This is similar to [[jdbc/reducible-query]] but reuses the
+  driver-specific configuration for the Connection and Statement/PreparedStatement. This is slightly different from [[execute-reducible-query]]
+  in that it is not intended to be used as part of middleware. Keywordizes column names. "
+  [db [sql & params]]
+  (let [driver (:engine db)]
+    (reify clojure.lang.IReduceInit
+      (reduce [_ rf init]
+        (do-with-connection-with-options
+         driver
+         db
+         nil
+         (fn [^Connection conn]
+           (with-open [stmt          (statement-or-prepared-statement driver conn sql params nil)
+                       ^ResultSet rs (try
+                                       (let [max-rows 0] ; 0 means no limit
+                                         (execute-statement-or-prepared-statement! driver stmt max-rows params sql))
+                                       (catch Throwable e
+                                         (throw (ex-info (tru "Error executing query: {0}" (ex-message e))
+                                                         {:driver driver
+                                                          :sql    (str/split-lines (driver/prettify-native-form driver sql))
+                                                          :params params}
+                                                         e))))]
+             ;; TODO - we should probably be using [[reducible-rows]] instead to convert to the correct types
+             (reduce rf init (jdbc/reducible-result-set rs {})))))))))
+
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                                                 Actions Stuff                                                  |

src/metabase/driver/sql_jdbc/sync.clj

@@ -28,6 +28,8 @@
 
  [sql-jdbc.describe-table
   add-table-pks
+  describe-fks
+  describe-fks-sql
   describe-table
   describe-table-fields
   describe-table-fks

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

 (ns metabase.driver.sql-jdbc.sync.describe-table
-  "SQL JDBC impl for `describe-table`, `describe-table-fks`, and `describe-nested-field-columns`."
+  "SQL JDBC impl for `describe-table`, `describe-fks`, `describe-table-fks`, and `describe-nested-field-columns`.
+  `describe-table-fks` is deprecated and will be replaced by `describe-fks` in the future."
   (:require
    [cheshire.core :as json]
    [clojure.java.jdbc :as jdbc]
@@ -263,15 +264,26 @@
 
 (defn describe-table-fks
   "Default implementation of [[metabase.driver/describe-table-fks]] for SQL JDBC drivers. Uses JDBC DatabaseMetaData."
-  [driver db-or-id-or-spec-or-conn table & [db-name-or-nil]]
-  (if (instance? Connection db-or-id-or-spec-or-conn)
-    (describe-table-fks* driver db-or-id-or-spec-or-conn table db-name-or-nil)
-    (sql-jdbc.execute/do-with-connection-with-options
-     driver
-     db-or-id-or-spec-or-conn
-     nil
-     (fn [^Connection conn]
-       (describe-table-fks* driver conn table db-name-or-nil)))))
+  [driver db-or-id-or-spec table & [db-name-or-nil]]
+  (sql-jdbc.execute/do-with-connection-with-options
+   driver
+   db-or-id-or-spec
+   nil
+   (fn [^Connection conn]
+     (describe-table-fks* driver conn table db-name-or-nil))))
+
+(defmulti describe-fks-sql
+ "Returns a SQL query ([sql & params]) for use in the default JDBC implementation of [[metabase.driver/describe-fks]],
+ i.e. [[describe-fks]]."
+ {:added    "0.50.0"
+  :arglists '([driver & {:keys [schema-names table-names]}])}
+ driver/dispatch-on-initialized-driver
+ :hierarchy #'driver/hierarchy)
+
+(defn describe-fks
+  "Default implementation of [[metabase.driver/describe-fks]] for JDBC drivers. Uses JDBC DatabaseMetaData."
+  [driver db & args]
+  (sql-jdbc.execute/reducible-query db (describe-fks-sql driver args)))
 
 (defn describe-table-indexes
   "Default implementation of [[metabase.driver/describe-table-indexes]] for SQL JDBC drivers. Uses JDBC DatabaseMetaData."

src/metabase/sync/fetch_metadata.clj

@@ -3,11 +3,13 @@
   tables, schemas, and fields, and their types. For example, with SQL databases, these functions use the JDBC
   DatabaseMetaData to get this information."
   (:require
+   [metabase.config :as config]
    [metabase.driver :as driver]
    [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
    [metabase.driver.util :as driver.u]
    [metabase.sync.interface :as i]
-   [metabase.util.malli :as mu]))
+   [metabase.util.malli :as mu]
+   [metabase.util.malli.fn :as mu.fn]))
 
 (mu/defn db-metadata :- i/DatabaseMetadata
   "Get basic Metadata about a `database` and its Tables. Doesn't include information about the Fields."
@@ -20,13 +22,31 @@
    table    :- i/TableInstance]
   (driver/describe-table (driver.u/database->driver database) database table))
 
-(mu/defn fk-metadata :- i/FKMetadata
+(mu/defn fk-metadata
+  "Effectively a wrapper for [[metabase.driver/describe-fks]] that also validates the output against the schema."
+  [database :- i/DatabaseInstance & {:as args}]
+  (cond->> (driver/describe-fks (driver.u/database->driver database) database args)
+    ;; Validate the output against the schema, except in prod.
+    ;; This is a workaround for the fact that [[mu/defn]] can't check reducible collections yet
+    (not config/is-prod?)
+    (eduction (map #(mu.fn/validate-output {} i/FKMetadataEntry %)))))
+
+(mu/defn table-fk-metadata :- [:maybe [:sequential i/FKMetadataEntry]]
   "Get information about the foreign keys belonging to `table`."
   [database :- i/DatabaseInstance
    table    :- i/TableInstance]
   (let [driver (driver.u/database->driver database)]
     (when (driver/database-supports? driver :foreign-keys database)
-      (driver/describe-table-fks driver database table))))
+      (if (driver/database-supports? driver :describe-fks database)
+        (vec (driver/describe-fks driver database :table-names [(:name table)] :schema-names [(:schema table)]))
+        #_{:clj-kondo/ignore [:deprecated-var]}
+        (vec (for [x (driver/describe-table-fks driver database table)]
+               {:fk-table-name   (:name table)
+                :fk-table-schema (:schema table)
+                :fk-column-name  (:fk-column-name x)
+                :pk-table-name   (:name (:dest-table x))
+                :pk-table-schema (:schema (:dest-table x))
+                :pk-column-name  (:dest-column-name x)}))))))
 
 (mu/defn nfc-metadata :- [:maybe [:set i/TableMetadataField]]
   "Get information about the nested field column fields within `table`."

src/metabase/sync/interface.clj

@@ -82,7 +82,7 @@
     "Schema for the expected output of [[metabase.driver.sql-jdbc.sync/describe-nested-field-columns]]."
     [:maybe [:set TableMetadataField]]))
 
-(mr/def ::FKMetadataEntry
+(mr/def ::TableFKMetadataEntry
   [:map
    [:fk-column-name   ::lib.schema.common/non-blank-string]
    [:dest-table       [:map
@@ -90,16 +90,29 @@
                        [:schema [:maybe ::lib.schema.common/non-blank-string]]]]
    [:dest-column-name ::lib.schema.common/non-blank-string]])
 
-(def FKMetadataEntry
+(def TableFKMetadataEntry
   "Schema for an individual entry in `FKMetadata`."
-  [:ref ::FKMetadataEntry])
+  [:ref ::TableFKMetadataEntry])
 
-(mr/def ::FKMetadata
-  [:maybe [:set FKMetadataEntry]])
+(mr/def ::TableFKMetadata
+  [:maybe [:set TableFKMetadataEntry]])
 
-(def FKMetadata
+(def TableFKMetadata
   "Schema for the expected output of `describe-table-fks`."
-  [:ref ::FKMetadata])
+  [:ref ::TableFKMetadata])
+
+(mr/def ::FKMetadataEntry
+  [:map
+   [:fk-table-name    ::lib.schema.common/non-blank-string]
+   [:fk-table-schema  [:maybe ::lib.schema.common/non-blank-string]]
+   [:fk-column-name   ::lib.schema.common/non-blank-string]
+   [:pk-table-name    ::lib.schema.common/non-blank-string]
+   [:pk-table-schema  [:maybe ::lib.schema.common/non-blank-string]]
+   [:pk-column-name   ::lib.schema.common/non-blank-string]])
+
+(def FKMetadataEntry
+  "Schema for an entry in the expected output of [[metabase.driver/describe-fks]]."
+  [:ref ::FKMetadataEntry])
 
 ;; These schemas are provided purely as conveniences since adding `:import` statements to get the corresponding
 ;; classes from the model namespaces also requires a `:require`, which `clj-refactor` seems more than happy to strip

src/metabase/sync/sync_metadata/fks.clj

@@ -3,6 +3,8 @@
   (:require
    [honey.sql :as sql]
    [metabase.db.connection :as mdb.connection]
+   [metabase.driver :as driver]
+   [metabase.driver.util :as driver.u]
    [metabase.models.table :as table]
    [metabase.sync.fetch-metadata :as fetch-metadata]
    [metabase.sync.interface :as i]
@@ -15,13 +17,12 @@
 
 (defn ^:private mark-fk-sql
   "Returns [sql & params] for [[mark-fk!]] according to the application DB's dialect."
-  [{:keys [db-id
-           fk-table-name
-           fk-table-schema
-           fk-column-name
-           pk-table-name
-           pk-table-schema
-           pk-column-name]}]
+  [db-id {:keys [fk-table-name
+                 fk-table-schema
+                 fk-column-name
+                 pk-table-name
+                 pk-table-schema
+                 pk-column-name]}]
   (let [field-id-query (fn [db-id table-schema table-name column-name]
                          {:select [[[:min :f.id] :id]]
                           ;; Cal 2024-03-04: We use `min` to limit this subquery to one result (limit 1 isn't allowed
@@ -75,21 +76,36 @@
 (mu/defn ^:private mark-fk!
   "Updates the `fk_target_field_id` of a Field. Returns 1 if the Field was successfully updated, 0 otherwise."
   [database :- i/DatabaseInstance
-   table    :- i/TableInstance
-   fk       :- i/FKMetadataEntry]
-  (u/prog1 (t2/query-one (mark-fk-sql {:db-id           (:id database)
-                                       :fk-table-name   (:name table)
-                                       :fk-table-schema (:schema table)
-                                       :fk-column-name  (:fk-column-name fk)
-                                       :pk-table-name   (:name (:dest-table fk))
-                                       :pk-table-schema (:schema (:dest-table fk))
-                                       :pk-column-name  (:dest-column-name fk)}))
-    (when (= <> 1)
-      (log/info (u/format-color 'cyan "Marking foreign key from %s %s -> %s %s"
-                                (sync-util/table-name-for-logging table)
-                                (sync-util/field-name-for-logging :name (:fk-column-name fk))
-                                (sync-util/table-name-for-logging (:dest-table fk))
-                                (sync-util/field-name-for-logging :name (:dest-column-name fk)))))))
+   metadata :- i/FKMetadataEntry]
+  (u/prog1 (t2/query-one (mark-fk-sql (:id database) metadata))
+  (when (= <> 1)
+    (log/info (u/format-color 'cyan "Marking foreign key from %s %s -> %s %s"
+                              (sync-util/table-name-for-logging :name (:fk-table-name metadata)
+                                                                :schema (:fk-table-schema metadata))
+                              (sync-util/field-name-for-logging :name (:fk-column-name metadata))
+                              (sync-util/table-name-for-logging :name (:fk-table-name metadata)
+                                                                :schema (:fk-table-schema metadata))
+                              (sync-util/field-name-for-logging :name (:pk-column-name metadata)))))))
+
+(mu/defn sync-fks-for-db!
+  "Sync the foreign keys for a `database`."
+  [database :- i/DatabaseInstance]
+  (sync-util/with-error-handling (format "Error syncing FKs for %s" (sync-util/name-for-logging database))
+    (let [schema-names (sync-util/db->sync-schemas database)
+          fk-metadata  (fetch-metadata/fk-metadata database :schema-names schema-names)]
+      (transduce (map (fn [x]
+                        (let [[updated failed] (try [(mark-fk! database x) 0]
+                                                    (catch Exception e
+                                                      (log/error e)
+                                                      [0 1]))]
+                          {:total-fks    1
+                           :updated-fks  updated
+                           :total-failed failed})))
+                 (partial merge-with +)
+                 {:total-fks    0
+                  :updated-fks  0
+                  :total-failed 0}
+                 fk-metadata))))
 
 (mu/defn sync-fks-for-table!
   "Sync the foreign keys for a specific `table`."
@@ -99,27 +115,31 @@
   ([database :- i/DatabaseInstance
     table    :- i/TableInstance]
    (sync-util/with-error-handling (format "Error syncing FKs for %s" (sync-util/name-for-logging table))
-     (let [fks-to-update (fetch-metadata/fk-metadata database table)]
-       {:total-fks   (count fks-to-update)
-        :updated-fks (sync-util/sum-numbers (fn [fk]
-                                              (mark-fk! database table fk))
-                                            fks-to-update)}))))
+     (let [fk-metadata (fetch-metadata/table-fk-metadata database table)]
+       {:total-fks   (count fk-metadata)
+        :updated-fks (sync-util/sum-numbers #(mark-fk! database %) fk-metadata)}))))
 
 (mu/defn sync-fks!
   "Sync the foreign keys in a `database`. This sets appropriate values for relevant Fields in the Metabase application
-  DB based on values from the `FKMetadata` returned by [[metabase.driver/describe-table-fks]]."
+  DB based on values from the `FKMetadata` returned by [[metabase.driver/describe-table-fks]].
+
+  If the driver supports the `:describe-fks` feature, [[metabase.driver/describe-fks]] is used to fetch the FK metadata.
+
+  This function also sets all the tables that should be synced to have `initial-sync-status=complete` once the sync is done."
   [database :- i/DatabaseInstance]
-  (reduce (fn [update-info table]
-            (let [table         (t2.realize/realize table)
-                  table-fk-info (sync-fks-for-table! database table)]
-              ;; Mark the table as done with its initial sync once this step is done even if it failed, because only
-              ;; sync-aborting errors should be surfaced to the UI (see
-              ;; `:metabase.sync.util/exception-classes-not-to-retry`).
-              (sync-util/set-initial-table-sync-complete! table)
-              (if (instance? Exception table-fk-info)
-                (update update-info :total-failed inc)
-                (merge-with + update-info table-fk-info))))
-          {:total-fks    0
-           :updated-fks  0
-           :total-failed 0}
-          (sync-util/db->reducible-sync-tables database)))
+  (u/prog1 (if (driver/database-supports? (driver.u/database->driver database) :describe-fks database)
+             (sync-fks-for-db! database)
+             (reduce (fn [update-info table]
+                       (let [table         (t2.realize/realize table)
+                             table-fk-info (sync-fks-for-table! database table)]
+                         (if (instance? Exception table-fk-info)
+                           (update update-info :total-failed inc)
+                           (merge-with + update-info table-fk-info))))
+                     {:total-fks    0
+                      :updated-fks  0
+                      :total-failed 0}
+                     (sync-util/db->reducible-sync-tables database)))
+    ;; Mark the table as done with its initial sync once this step is done even if it failed, because only
+    ;; sync-aborting errors should be surfaced to the UI (see
+    ;; `:metabase.sync.util/exception-classes-not-to-retry`).
+    (sync-util/set-initial-table-sync-complete-for-db! database)))

src/metabase/sync/util.clj

@@ -303,6 +303,17 @@
   (when (not= (:initial_sync_status table) "complete")
     (t2/update! :model/Table (u/the-id table) {:initial_sync_status "complete"})))
 
+(def ^:private sync-tables-kv-args
+  {:active          true
+   :visibility_type nil})
+
+(defn set-initial-table-sync-complete-for-db!
+  "Marks initial sync for all tables in `db` as complete so that it becomes usable in the UI, if not already
+  set."
+  [database-or-id]
+  (t2/update! :model/Table (merge sync-tables-kv-args {:db_id (u/the-id database-or-id)})
+              {:initial_sync_status "complete"}))
+
 (defn set-initial-database-sync-complete!
   "Marks initial sync as complete for this database so that this is reflected in the UI, if not already set"
   [database]
@@ -320,11 +331,11 @@
 ;;; +----------------------------------------------------------------------------------------------------------------+
 
 (def ^:private sync-tables-clause
-  [:and [:= :active true]
-        [:= :visibility_type nil]])
+  (into [:and] (for [[k v] sync-tables-kv-args]
+                 [:= k v])))
 
 (defn db->sync-tables
-  "Return all the Tables that should go through the sync processes for `database-or-id`."
+  "Returns all the Tables that have their metadata sync'd for `database-or-id`."
   [database-or-id]
   (t2/select :model/Table, :db_id (u/the-id database-or-id), {:where sync-tables-clause}))
 
@@ -333,6 +344,13 @@
   [database-or-id]
   (t2/reducible-select :model/Table, :db_id (u/the-id database-or-id), {:where sync-tables-clause}))
 
+(defn db->sync-schemas
+  "Returns all the Schemas that have their metadata sync'd for `database-or-id`."
+  [database-or-id]
+  (vec (map :schema (t2/query {:select-distinct [:schema]
+                               :from            [:metabase_table]
+                               :where           [:and sync-tables-clause [:= :db_id (u/the-id database-or-id)]]}))))
+
 (defmulti name-for-logging
   "Return an appropriate string for logging an object in sync logging messages. Should be something like
 

test/metabase/driver/sql_jdbc_test.clj

@@ -76,6 +76,7 @@
             :dest-table       {:name   "CATEGORIES"
                                :schema "PUBLIC"}
             :dest-column-name "ID"}}
+         #_{:clj-kondo/ignore [:deprecated-var]}
          (driver/describe-table-fks :h2 (mt/db) (t2/select-one Table :id (mt/id :venues))))))
 
 (deftest ^:parallel table-rows-sample-test

test/metabase/sync/sync_metadata/fields_test.clj

@@ -9,6 +9,8 @@
    [metabase.models :refer [Field Table]]
    [metabase.query-processor :as qp]
    [metabase.sync :as sync]
+   [metabase.sync.sync-metadata.fields :as sync-fields]
+   [metabase.sync.sync-metadata.fks :as sync-fks]
    [metabase.sync.util-test :as sync.util-test]
    [metabase.test :as mt]
    [metabase.test.data.one-off-dbs :as one-off-dbs]
@@ -298,3 +300,33 @@
                                  :steps
                                  (m/find-first (comp #{"sync-fields"} first)))]
         (is (=? ["sync-fields" {:total-fields 2 :updated-fields 2}] field-sync-info))))))
+
+(mt/defdataset country
+  [["continent"
+    [{:field-name "name", :base-type :type/Text}]
+    [["Africa"]]]
+   ["country"
+    [{:field-name "name", :base-type :type/Text}
+     {:field-name "continent_id", :base-type :type/Integer :fk :continent}]
+    [["Ghana" 1]]]])
+
+(deftest sync-fks-and-fields-for-table-test
+  (testing "[[sync-fields/sync-fields-for-table!]] and [[sync-fks/sync-fks-for-table!]] should sync fields and fks"
+    (mt/test-drivers (mt/normal-drivers-with-feature :foreign-keys)
+      (mt/dataset country
+        (let [tables (t2/select :model/Table :db_id (mt/id))]
+          ;; 1. delete the fields that were just synced
+          (t2/delete! :model/Field :table_id [:in (map :id tables)])
+          ;; 2. sync the metadata for each table
+          (run! sync-fields/sync-fields-for-table! tables)
+          (run! sync-fks/sync-fks-for-table! tables)
+          (let [continent-id-field (t2/select-one :model/Field :%lower.name "id" :table_id (mt/id :continent))]
+            (is (= #{{:name "name",         :semantic_type nil,      :fk_target_field_id nil}
+                     {:name "id",           :semantic_type :type/PK, :fk_target_field_id nil}
+                     {:name "continent_id", :semantic_type :type/FK, :fk_target_field_id (:id continent-id-field)}}
+                   (set (map #(into {} %)
+                             (t2/select [Field
+                                         [:%lower.name :name]
+                                         :semantic_type
+                                         :fk_target_field_id]
+                                        :table_id [:in (map :id tables)])))))))))))

test/metabase/sync_test.clj

@@ -78,6 +78,7 @@
   [_ _ table]
   (get sync-test-tables (:name table)))
 
+#_{:clj-kondo/ignore [:deprecated-var]}
 (defmethod driver/describe-table-fks ::sync-test
   [_ _ table]
   (set (when (= "movie" (:name table))

test/metabase/test/mock/moviedb.clj

@@ -85,6 +85,7 @@
       (update :fields (partial map-indexed (fn [idx field]
                                              (assoc field :database-position idx))))))
 
+#_{:clj-kondo/ignore [:deprecated-var]}
 (defmethod driver/describe-table-fks ::moviedb [_ _ table]
   (-> (get moviedb-tables (:name table))
       :fks