finish out extraction

this is just dbms and timezone so far

src/metabase/sync/api/notify.clj

@@ -8,6 +8,7 @@
    [metabase.driver.util :as driver.u]
    [metabase.sync.core :as sync]
    [metabase.sync.sync-metadata :as sync-metadata]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.tables :as sync-tables]
    [metabase.sync.util :as sync-util]
    [metabase.util.i18n :refer [trs]]
@@ -57,7 +58,7 @@
         table  {:name   table-name
                 :schema schema-name}]
     (if (driver/table-exists? driver database table)
-      (let [created (sync-tables/create-or-reactivate-table! database table)]
+      (let [created (sync-tables/create-or-reactivate-table! database table persist.appdb/reader persist.appdb/writer)]
         (doto created
           sync/sync-table!
           sync-util/set-initial-table-sync-complete!))

src/metabase/sync/persist.clj

@@ -0,0 +1,112 @@
+(ns metabase.sync.persist
+  "Protocols for abstracting app-db access during sync.
+   The sync process needs to read and write Database, Table, and Field models.
+   These protocols allow swapping the persistence backend.")
+
+;; =============================================================================
+;; Current Protocol State (updated as we work through sync steps)
+;; =============================================================================
+
+(defprotocol SyncDatabaseWriter
+  "Protocol for writing sync metadata to the app database."
+
+  (set-dbms-version! [this database-id version]
+    "Set the DBMS version for a database. `version` is a map.")
+
+  (set-database-timezone! [this database-id timezone-id]
+    "Set the timezone for a database. `timezone-id` is a string.")
+
+  (set-database-details-version! [this database-id details version]
+    "Merge version into database details.")
+
+  (create-table! [this table-map]
+    "Insert a new table row. Returns the created instance.")
+
+  (reactivate-table! [this table-id changes]
+    "Update an inactive table to mark it active again with the given changes map.")
+
+  (retire-table! [this database-id schema table-name]
+    "Mark a table as inactive by database-id, schema, and name.")
+
+  (update-table! [this table-id changes]
+    "Apply a partial update to a table by id.")
+
+  (update-table-schema! [this database-id old-schema new-schema]
+    "Rename a schema on all matching tables in a database.")
+
+  (archive-table! [this table-id changes]
+    "Archive a deactivated table (set archived_at, rename). Returns the number of rows updated.")
+
+  (insert-fields! [this field-maps]
+    "Bulk insert Field rows. Returns a sequence of inserted PKs.")
+
+  (reactivate-fields! [this field-ids]
+    "Mark fields as active by their IDs.")
+
+  (retire-field! [this field-id]
+    "Mark a single field as inactive. Returns 1 if updated, 0 otherwise.")
+
+  (update-field! [this field-id changes]
+    "Apply a partial update to a field by id. Returns number of rows updated.")
+
+  (mark-fk! [this database-id fk-metadata]
+    "Mark a field as a FK pointing to a target field, based on FK metadata.
+     Returns 1 if the field was updated, 0 otherwise.")
+
+  (set-table-indexes! [this table-id indexed-field-ids]
+    "Update all fields in a table: those in `indexed-field-ids` get database_indexed=true, others get false.")
+
+  (batch-set-indexed! [this field-ids indexed?]
+    "Set database_indexed for a batch of field IDs (by parent_id=nil).")
+
+  (update-database! [this database-id changes]
+    "Apply a partial update to a database by id.")
+
+  (update-field-by-name! [this table-id field-name changes]
+    "Update a field by its name within a table."))
+
+(defprotocol SyncDatabaseReader
+  "Protocol for reading sync metadata from the app database."
+
+  (active-tables [this database-id]
+    "Return a set of active tables for a database.")
+
+  (all-tables [this database-id]
+    "Return a set of all tables (active + inactive) for a database.")
+
+  (find-inactive-table-id [this database-id schema table-name]
+    "Find an inactive table by db_id/schema/name. Returns the table id or nil.")
+
+  (get-table [this table-id]
+    "Get a single table by id.")
+
+  (archivable-tables [this database-id threshold-expr]
+    "Return tables eligible for archiving: inactive, not yet archived, not transform targets,
+     deactivated before the threshold.")
+
+  (active-fields [this table-id]
+    "Return active fields for a table, with the columns needed for sync.")
+
+  (matching-inactive-fields [this table-id field-names parent-id]
+    "Find inactive fields matching the given canonical names. Returns a sequence of field instances.")
+
+  (select-fields-by-ids [this field-ids]
+    "Fetch fields by a collection of IDs.")
+
+  (find-syncable-table [this database-id table-name table-schema]
+    "Find a syncable table by case-insensitive name/schema match. Returns best-matching table or nil.")
+
+  (field-ids-for-index-names [this table-id index-names]
+    "Return field IDs matching the given index names for a table (top-level fields only).")
+
+  (indexed-field-ids-for-table [this table-id]
+    "Return the set of field IDs currently marked as indexed for a table.")
+
+  (indexed-field-ids-for-database [this database-id]
+    "Return the set of field IDs currently marked as indexed for all tables in a database (top-level only).")
+
+  (field-ids-for-indexes [this database-id indexes]
+    "Resolve a collection of index descriptors to field IDs across the whole database.")
+
+  (find-active-table-id [this database-id table-name]
+    "Find an active table by db_id and name. Returns the table id or nil."))

src/metabase/sync/persist/appdb.clj

@@ -0,0 +1,258 @@
+(ns metabase.sync.persist.appdb
+  "Default implementation of the sync persist protocols backed by toucan2 / app-db."
+  (:require
+   [honey.sql :as sql]
+   [metabase.app-db.core :as mdb]
+   [metabase.models.interface :as mi]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.util :as sync-util]
+   [metabase.util :as u]
+   [toucan2.core :as t2]
+   [toucan2.util :as t2.util]))
+
+(defn- mark-fk-sql
+  "Returns [sql & params] for marking a FK according to the application DB's dialect."
+  [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]]
+                          :from   [[:metabase_field :f]]
+                          :join   [[:metabase_table :t] [:= :f.table_id :t.id]]
+                          :left-join [[:metabase_field_user_settings :u] [:= :f.id :u.field_id]]
+                          :where  [:and
+                                   [:= :u.fk_target_field_id nil]
+                                   [:= :u.semantic_type nil]
+                                   [:= :t.db_id db-id]
+                                   [:= [:lower :f.name] (u/lower-case-en column-name)]
+                                   [:= [:lower :t.name] (u/lower-case-en table-name)]
+                                   [:= [:lower :t.schema] (some-> table-schema u/lower-case-en)]
+                                   [:= :f.active true]
+                                   [:not= :f.visibility_type "retired"]
+                                   [:= :t.active true]
+                                   [:= :t.visibility_type nil]]})
+        fk-field-id-query (field-id-query db-id fk-table-schema fk-table-name fk-column-name)
+        pk-field-id-query (field-id-query db-id pk-table-schema pk-table-name pk-column-name)
+        valid-condition (fn [k]
+                          [:or
+                           [:= :f.fk_target_field_id nil]
+                           [:not= :f.fk_target_field_id k]])
+        q (case (mdb/db-type)
+            :mysql
+            {:update [:metabase_field :f]
+             :join   [[fk-field-id-query :fk] [:= :fk.id :f.id]
+                      [pk-field-id-query :pk]
+                      (valid-condition :pk.id)]
+             :set    {:fk_target_field_id :pk.id
+                      :has_field_values   [:case [:= :has_field_values "auto-list"] nil :else :has_field_values]
+                      :semantic_type      "type/FK"}}
+            :postgres
+            {:update [:metabase_field :f]
+             :from   [[fk-field-id-query :fk]]
+             :join   [[pk-field-id-query :pk] true]
+             :set    {:fk_target_field_id :pk.id
+                      :has_field_values   [:case [:= :has_field_values "auto-list"] nil :else :has_field_values]
+                      :semantic_type      "type/FK"}
+             :where  [:and
+                      [:= :fk.id :f.id]
+                      (valid-condition :pk.id)]}
+            :h2
+            {:update [:metabase_field :f]
+             :set    {:fk_target_field_id pk-field-id-query
+                      :has_field_values   [:case [:= :has_field_values "auto-list"] nil :else :has_field_values]
+                      :semantic_type      "type/FK"}
+             :where  [:and
+                      [:= :f.id fk-field-id-query]
+                      [:not= pk-field-id-query nil]
+                      (valid-condition pk-field-id-query)]})]
+    (sql/format q :dialect (mdb/quoting-style (mdb/db-type)))))
+
+(defrecord AppDbSyncWriter []
+  persist/SyncDatabaseWriter
+  (set-dbms-version! [_this database-id version]
+    (t2/update! :model/Database database-id {:dbms_version version}))
+
+  (set-database-timezone! [_this database-id timezone-id]
+    (t2/update! :model/Database database-id {:timezone timezone-id}))
+
+  (set-database-details-version! [_this database-id details version]
+    (t2/update! :model/Database database-id
+                {:details (assoc details :version version)}))
+
+  (create-table! [_this table-map]
+    (t2/insert-returning-instance! :model/Table table-map))
+
+  (reactivate-table! [_this table-id changes]
+    (t2/update! :model/Table table-id changes))
+
+  (retire-table! [_this database-id schema table-name]
+    (t2/update! :model/Table {:db_id  database-id
+                              :schema schema
+                              :name   table-name
+                              :active true}
+                {:active false}))
+
+  (update-table! [_this table-id changes]
+    (t2/update! :model/Table table-id changes))
+
+  (update-table-schema! [_this database-id old-schema new-schema]
+    (t2/update! :model/Table
+                :db_id database-id
+                :schema old-schema
+                {:schema new-schema}))
+
+  (archive-table! [_this table-id changes]
+    (t2/update! :model/Table
+                {:id table-id
+                 :active false}
+                changes))
+
+  (insert-fields! [_this field-maps]
+    (t2/insert-returning-pks! :model/Field field-maps))
+
+  (reactivate-fields! [_this field-ids]
+    (t2/update! :model/Field {:id [:in field-ids]}
+                {:active true}))
+
+  (retire-field! [_this field-id]
+    (t2/update! :model/Field field-id {:active false}))
+
+  (update-field! [_this field-id changes]
+    (t2/update! :model/Field field-id changes))
+
+  (mark-fk! [_this database-id fk-metadata]
+    (t2/query-one (mark-fk-sql database-id fk-metadata)))
+
+  (set-table-indexes! [_this table-id indexed-field-ids]
+    (t2/update! :model/Field {:table_id table-id}
+                {:database_indexed (if (seq indexed-field-ids)
+                                     [:case [:in :id indexed-field-ids] true :else false]
+                                     false)}))
+
+  (batch-set-indexed! [_this field-ids indexed?]
+    (when (seq field-ids)
+      (t2/update! :model/Field :parent_id nil :id [:in field-ids] {:database_indexed indexed?})))
+
+  (update-database! [_this database-id changes]
+    (t2/update! :model/Database database-id changes))
+
+  (update-field-by-name! [_this table-id field-name changes]
+    (t2/update! :model/Field {:name field-name :table_id table-id} changes)))
+
+(defn sync-writer
+  "Create an app-db backed sync writer."
+  []
+  (->AppDbSyncWriter))
+
+(def writer
+  "Singleton app-db backed sync writer."
+  (->AppDbSyncWriter))
+
+(defrecord AppDbSyncReader []
+  persist/SyncDatabaseReader
+  (active-tables [_this database-id]
+    (set (t2/select [:model/Table :id :name :schema :data_authority
+                     :description :database_require_filter :estimated_row_count
+                     :visibility_type :initial_sync_status :is_writable]
+                    :db_id database-id
+                    :active true)))
+
+  (all-tables [_this database-id]
+    (set (t2/select [:model/Table :id :name :schema :data_authority
+                     :description :database_require_filter :estimated_row_count
+                     :visibility_type :initial_sync_status :is_writable]
+                    :db_id database-id)))
+
+  (find-inactive-table-id [_this database-id schema table-name]
+    (t2/select-one-pk :model/Table
+                      :db_id database-id
+                      :schema schema
+                      :name table-name
+                      :active false))
+
+  (get-table [_this table-id]
+    (t2/select-one :model/Table table-id))
+
+  (archivable-tables [_this database-id threshold-expr]
+    (t2/select :model/Table
+               :db_id database-id
+               :active false
+               :archived_at nil
+               :transform_target false
+               :deactivated_at [:< threshold-expr]))
+
+  (active-fields [_this table-id]
+    (t2/select [:model/Field :name :database_type :base_type :effective_type :coercion_strategy :semantic_type
+                :parent_id :id :description :database_position :nfc_path
+                :database_is_auto_increment :database_required
+                :database_default :database_is_generated :database_is_nullable :database_is_pk
+                :database_partitioned :json_unfolding :position :preview_display]
+               :table_id  table-id
+               :active    true
+               {:order-by [[:position :asc] [:%lower.name :asc]]}))
+
+  (matching-inactive-fields [_this table-id field-names parent-id]
+    (when (seq field-names)
+      (t2/select :model/Field
+                 :table_id    table-id
+                 :%lower.name [:in field-names]
+                 :parent_id   parent-id
+                 :active      false)))
+
+  (select-fields-by-ids [_this field-ids]
+    (when (seq field-ids)
+      (t2/select :model/Field :id [:in field-ids])))
+
+  (find-syncable-table [_this database-id table-name table-schema]
+    (->> (t2/select :model/Table
+                    :db_id database-id
+                    :%lower.name (t2.util/lower-case-en table-name)
+                    :%lower.schema (some-> table-schema t2.util/lower-case-en)
+                    {:where sync-util/sync-tables-clause})
+         (sort-by (fn [item]
+                    [(not= (:schema item) table-schema)
+                     (not= (:name item) table-name)]))
+         first))
+
+  (field-ids-for-index-names [_this table-id index-names]
+    (when (seq index-names)
+      (t2/select-pks-vec :model/Field :name [:in index-names] :table_id table-id :parent_id nil)))
+
+  (indexed-field-ids-for-table [_this table-id]
+    (t2/select-pks-set :model/Field :table_id table-id :database_indexed true))
+
+  (indexed-field-ids-for-database [_this database-id]
+    (t2/select-pks-set :model/Field
+                       :table_id [:in {:select [[:t.id]]
+                                       :from [[(t2/table-name :model/Table) :t]]
+                                       :where [:= :t.db_id database-id]}]
+                       :parent_id nil
+                       :database_indexed true))
+
+  (find-active-table-id [_this database-id table-name]
+    (t2/select-one-pk :model/Table
+                      :db_id database-id
+                      :name table-name
+                      :active true))
+
+  (field-ids-for-indexes [_this database-id indexes]
+    (reduce
+     (fn [accum index-batch]
+       (let [normal-indexes (map (juxt #(:table-schema % "__null__") :table-name :field-name) index-batch)
+             query (t2/reducible-query {:select [[:f.id]]
+                                        :from [[(t2/table-name :model/Field) :f]]
+                                        :inner-join [[(t2/table-name :model/Table) :t] [:= :f.table_id :t.id]]
+                                        :where [:and [:in [:composite [:coalesce :t.schema "__null__"] :t.name :f.name] normal-indexes]
+                                                [:= :t.db_id database-id]
+                                                [:= :parent_id nil]]})]
+         (into accum (keep :id) query)))
+     #{}
+     (partition-all 5000 indexes))))
+
+(defn sync-reader
+  "Create an app-db backed sync reader."
+  []
+  (->AppDbSyncReader))
+
+(def reader
+  "Singleton app-db backed sync reader."
+  (->AppDbSyncReader))

src/metabase/sync/sync_metadata.clj

@@ -14,6 +14,7 @@
    [metabase.sync.sync-metadata.fks :as sync-fks]
    [metabase.sync.sync-metadata.indexes :as sync-indexes]
    [metabase.sync.sync-metadata.metabase-metadata :as metabase-metadata]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.sync-timezone :as sync-tz]
    [metabase.sync.sync-metadata.tables :as sync-tables]
    [metabase.sync.util :as sync-util]
@@ -46,19 +47,19 @@
   (format "Total number of indexes sync''d %d, %d added and %d removed"
           total-indexes added-indexes removed-indexes))
 
-(defn- make-sync-steps [db-metadata]
-  [(sync-util/create-sync-step "sync-dbms-version" sync-dbms-ver/sync-dbms-version! sync-dbms-version-summary)
-   (sync-util/create-sync-step "sync-timezone" sync-tz/sync-timezone! sync-timezone-summary)
+(defn- make-sync-steps [db-metadata reader writer]
+  [(sync-util/create-sync-step "sync-dbms-version" #(sync-dbms-ver/sync-dbms-version! % writer) sync-dbms-version-summary)
+   (sync-util/create-sync-step "sync-timezone" #(sync-tz/sync-timezone! % writer) sync-timezone-summary)
    ;; Make sure the relevant table models are up-to-date
-   (sync-util/create-sync-step "sync-tables" #(sync-tables/sync-tables-and-database! % db-metadata) sync-tables-summary)
+   (sync-util/create-sync-step "sync-tables" #(sync-tables/sync-tables-and-database! % db-metadata reader writer) sync-tables-summary)
    ;; Now for each table, sync the fields
-   (sync-util/create-sync-step "sync-fields" sync-fields/sync-fields! sync-fields-summary)
+   (sync-util/create-sync-step "sync-fields" #(sync-fields/sync-fields! % reader writer) sync-fields-summary)
    ;; Now for each table, sync the FKS. This has to be done after syncing all the fields to make sure target fields exist
-   (sync-util/create-sync-step "sync-fks" sync-fks/sync-fks! sync-fks-summary)
+   (sync-util/create-sync-step "sync-fks" #(sync-fks/sync-fks! % writer) sync-fks-summary)
    ;; Sync index info if the database supports it
-   (sync-util/create-sync-step "sync-indexes" sync-indexes/maybe-sync-indexes! sync-indexes-summary)
+   (sync-util/create-sync-step "sync-indexes" #(sync-indexes/maybe-sync-indexes! % reader writer) sync-indexes-summary)
    ;; Sync the metabase metadata table if it exists.
-   (sync-util/create-sync-step "sync-metabase-metadata" #(metabase-metadata/sync-metabase-metadata! % db-metadata))])
+   (sync-util/create-sync-step "sync-metabase-metadata" #(metabase-metadata/sync-metabase-metadata! % db-metadata reader writer))])
 
 (mu/defn sync-db-metadata!
   "Sync the metadata for a Metabase `database`. This makes sure child Table & Field objects are synchronized."
@@ -66,7 +67,7 @@
   (sync-util/sync-operation :sync-metadata database (format "Sync metadata for %s" (sync-util/name-for-logging database))
     (let [db-metadata (tracing/with-span :sync "sync.metadata.fetch-metadata" {:db/id (:id database)}
                         (fetch-metadata/db-metadata database))]
-      (u/prog1 (sync-util/run-sync-operation "sync" database (make-sync-steps db-metadata))
+      (u/prog1 (sync-util/run-sync-operation "sync" database (make-sync-steps db-metadata persist.appdb/reader persist.appdb/writer))
         (if (some sync-util/abandon-sync? (map second (:steps <>)))
           (sync-util/set-initial-database-sync-aborted! database)
           (sync-util/set-initial-database-sync-complete! database))))))

src/metabase/sync/sync_metadata/dbms_version.clj

@@ -3,9 +3,10 @@
    [metabase.driver :as driver]
    [metabase.driver.util :as driver.u]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.util.malli :as mu]
-   [metabase.util.malli.schema :as ms]
-   [toucan2.core :as t2]))
+   [metabase.util.malli.schema :as ms]))
 
 (def DBMSVersion
   "Schema for the expected output of [[sync-dbms-version!]]."
@@ -14,9 +15,12 @@
 
 (mu/defn sync-dbms-version! :- [:maybe DBMSVersion]
   "Get the DBMS version as provided by the driver and save it in the Database."
-  [database :- i/DatabaseInstance]
-  (let [driver  (driver.u/database->driver database)
-        version (driver/dbms-version driver database)]
-    (when (not= version (:dbms_version database))
-      (t2/update! :model/Database (:id database) {:dbms_version version}))
-    version))
+  ([database :- i/DatabaseInstance]
+   (sync-dbms-version! database persist.appdb/writer))
+  ([database :- i/DatabaseInstance
+    writer   :- [:fn #(satisfies? persist/SyncDatabaseWriter %)]]
+   (let [driver  (driver.u/database->driver database)
+         version (driver/dbms-version driver database)]
+     (when (not= version (:dbms_version database))
+       (persist/set-dbms-version! writer (:id database) version))
+     version)))

src/metabase/sync/sync_metadata/fields.clj

@@ -42,6 +42,8 @@
    [metabase.driver.util :as driver.u]
    [metabase.sync.fetch-metadata :as fetch-metadata]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.fields.our-metadata :as fields.our-metadata]
    [metabase.sync.sync-metadata.fields.sync-instances :as sync-instances]
    [metabase.sync.sync-metadata.fields.sync-metadata :as sync-metadata]
@@ -49,9 +51,7 @@
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.schema :as ms]
-   [metabase.warehouse-schema.models.table :as table]
-   [toucan2.core :as t2]
-   [toucan2.util :as t2.util]))
+   [metabase.warehouse-schema.models.table :as table]))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                                            PUTTING IT ALL TOGETHER                                             |
@@ -62,12 +62,14 @@
   properties (e.g. base type and comment/remark) as needed. Returns number of Fields synced."
   [database    :- i/DatabaseInstance
    table       :- i/TableInstance
-   db-metadata :- [:set i/TableMetadataField]]
-  (+ (sync-instances/sync-instances! table db-metadata (fields.our-metadata/our-metadata table))
+   db-metadata :- [:set i/TableMetadataField]
+   reader writer]
+  (+ (sync-instances/sync-instances! table db-metadata (fields.our-metadata/our-metadata table reader)
+                                     nil reader writer)
      ;; Now that tables are synced and fields created as needed make sure field properties are in sync.
      ;; Re-fetch our metadata because there might be some things that have changed after calling
      ;; `sync-instances`
-     (sync-metadata/update-metadata! database table db-metadata (fields.our-metadata/our-metadata table))))
+     (sync-metadata/update-metadata! database table db-metadata (fields.our-metadata/our-metadata table reader) writer)))
 
 (defn- select-best-matching-name
   "Returns a key function for use with [[sort-by]] that ranks items based on how closely their `:schema` and `:name` match the given target values.
@@ -81,7 +83,9 @@
                           [:updated-fields ms/IntGreaterThanOrEqualToZero]
                           [:total-fields   ms/IntGreaterThanOrEqualToZero]]
   "Sync the Fields in the Metabase application database for all the Tables in a `database`."
-  [database :- i/DatabaseInstance]
+  ([database :- i/DatabaseInstance]
+   (sync-fields! database persist.appdb/reader persist.appdb/writer))
+  ([database :- i/DatabaseInstance reader writer]
   (sync-util/with-error-handling (format "Error syncing Fields for Database ''%s''" (sync-util/name-for-logging database))
     (let [driver          (driver.u/database->driver database)
           schemas?        (driver.u/supports? driver :schemas database)]
@@ -90,13 +94,7 @@
                             (partition-by (juxt :table-name :table-schema))
                             (map (fn [table-metadata]
                                    (let [{:keys [table-name table-schema]} (first table-metadata)
-                                         table   (->> (t2/select :model/Table
-                                                                 :db_id (:id database)
-                                                                 :%lower.name (t2.util/lower-case-en table-name)
-                                                                 :%lower.schema (some-> table-schema t2.util/lower-case-en)
-                                                                 {:where sync-util/sync-tables-clause})
-                                                      (sort-by (select-best-matching-name table-schema table-name))
-                                                      first)
+                                         table   (persist/find-syncable-table reader (:id database) table-name table-schema)
                                          updated (if table
                                                    (try
                                                      ;; TODO: decouple nested field columns sync from field sync. This will allow
@@ -106,7 +104,7 @@
                                                                          (set table-metadata)
                                                                          database
                                                                          table)]
-                                                       (sync-and-update! database table all-metadata))
+                                                       (sync-and-update! database table all-metadata reader writer))
                                                      (catch Exception e
                                                        (log/error e)
                                                        0))
@@ -128,7 +126,7 @@
                      {:total-fields   0
                       :updated-fields 0}
                      (sync-util/sync-schemas database))
-          (sync! (fetch-metadata/fields-metadata database)))))))
+          (sync! (fetch-metadata/fields-metadata database))))))))
 
 (mu/defn sync-fields-for-table!
   "Sync the Fields in the Metabase application database for a specific `table`."
@@ -137,6 +135,11 @@
 
   ([database :- i/DatabaseInstance
     table    :- i/TableInstance]
+   (sync-fields-for-table! database table persist.appdb/reader persist.appdb/writer))
+
+  ([database :- i/DatabaseInstance
+    table    :- i/TableInstance
+    reader writer]
    (sync-util/with-error-handling (format "Error syncing Fields for Table ''%s''" (sync-util/name-for-logging table))
      (let [db-metadata (fetch-metadata/table-fields-metadata database table)
            ;; TODO: decouple nested field columns sync from field sync. This will allow
@@ -144,4 +147,4 @@
            ;; Also this should be a driver method, not a sql-jdbc.sync method
            db-metadata (fetch-metadata/include-nested-fields-for-table db-metadata database table)]
        {:total-fields   (count db-metadata)
-        :updated-fields (sync-and-update! database table db-metadata)}))))
+        :updated-fields (sync-and-update! database table db-metadata reader writer)}))))

src/metabase/sync/sync_metadata/fields/our_metadata.clj

@@ -6,11 +6,11 @@
   (:require
    [medley.core :as m]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.fields.common :as common]
    [metabase.util :as u]
-   [metabase.util.malli :as mu]
-   [metabase.warehouse-schema.models.table :as table]
-   [toucan2.core :as t2]))
+   [metabase.util.malli :as mu]))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                                         FETCHING OUR CURRENT METADATA                                          |
@@ -73,18 +73,17 @@
 
 (mu/defn- table->fields :- [:maybe [:sequential i/FieldInstance]]
   "Fetch active Fields from the Metabase application database for a given `table`."
-  [table :- i/TableInstance]
-  (t2/select [:model/Field :name :database_type :base_type :effective_type :coercion_strategy :semantic_type
-              :parent_id :id :description :database_position :nfc_path
-              :database_is_auto_increment :database_required
-              :database_default :database_is_generated :database_is_nullable :database_is_pk
-              :database_partitioned :json_unfolding :position :preview_display]
-             :table_id  (u/the-id table)
-             :active    true
-             {:order-by table/field-order-rule}))
+  ([table :- i/TableInstance]
+   (table->fields table persist.appdb/reader))
+  ([table  :- i/TableInstance
+    reader :- [:fn #(satisfies? persist/SyncDatabaseReader %)]]
+   (persist/active-fields reader (u/the-id table))))
 
 (mu/defn our-metadata :- [:set common/TableMetadataFieldWithID]
   "Return information we have about Fields for a `table` in the application database in (almost) exactly the same
    `TableMetadataField` format returned by `describe-table`."
-  [table :- i/TableInstance]
-  (-> table table->fields fields->our-metadata))
+  ([table :- i/TableInstance]
+   (our-metadata table persist.appdb/reader))
+  ([table  :- i/TableInstance
+    reader :- [:fn #(satisfies? persist/SyncDatabaseReader %)]]
+   (-> (table->fields table reader) fields->our-metadata)))

src/metabase/sync/sync_metadata/fields/sync_instances.clj

@@ -11,14 +11,15 @@
    [metabase.lib.schema.id :as lib.schema.id]
    [metabase.models.humanization :as humanization]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.fields.common :as common]
    [metabase.sync.sync-metadata.fields.our-metadata :as fields.our-metadata]
    [metabase.sync.util :as sync-util]
    [metabase.util :as u]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
-   [metabase.util.malli.schema :as ms]
-   [toucan2.core :as t2]))
+   [metabase.util.malli.schema :as ms]))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                                         CREATING / REACTIVATING FIELDS                                         |
@@ -29,26 +30,27 @@
   exist."
   [table               :- i/TableInstance
    new-field-metadatas :- [:maybe [:sequential i/TableMetadataField]]
-   parent-id           :- common/ParentID]
+   parent-id           :- common/ParentID
+   reader]
   (when (seq new-field-metadatas)
-    (t2/select     :model/Field
-                   :table_id    (u/the-id table)
-                   :%lower.name [:in (map common/canonical-name new-field-metadatas)]
-                   :parent_id   parent-id
-                   :active      false)))
+    (persist/matching-inactive-fields reader
+                                     (u/the-id table)
+                                     (map common/canonical-name new-field-metadatas)
+                                     parent-id)))
 
 (mu/defn- insert-new-fields! :- [:maybe [:sequential ::lib.schema.id/field]]
   "Insert new Field rows for for all the Fields described by `new-field-metadatas`. Returns IDs of newly inserted
   Fields."
   [table               :- i/TableInstance
    new-field-metadatas :- [:maybe [:sequential i/TableMetadataField]]
-   parent-id           :- common/ParentID]
+   parent-id           :- common/ParentID
+   writer]
   (when (seq new-field-metadatas)
-    (t2/insert-returning-pks! :model/Field
-                              (for [{:keys [base-type coercion-strategy database-is-auto-increment database-partitioned database-position
-                                            database-is-generated database-is-nullable database-default pk?
-                                            database-required database-type effective-type field-comment json-unfolding nfc-path visibility-type]
-                                     field-name :name :as field} (sort-by :database-position new-field-metadatas)
+    (persist/insert-fields! writer
+                            (for [{:keys [base-type coercion-strategy database-is-auto-increment database-partitioned database-position
+                                          database-is-generated database-is-nullable database-default pk?
+                                          database-required database-type effective-type field-comment json-unfolding nfc-path visibility-type]
+                                   field-name :name :as field} (sort-by :database-position new-field-metadatas)
                                     :let [semantic-type (common/semantic-type field)
                                           has-field-values (when (sync-util/can-be-list? base-type semantic-type)
                                                              :auto-list)]]
@@ -94,19 +96,19 @@
   handle nested Fields."
   [table               :- i/TableInstance
    new-field-metadatas :- [:maybe [:sequential i/TableMetadataField]]
-   parent-id           :- common/ParentID]
-  (let [fields-to-reactivate (matching-inactive-fields table new-field-metadatas parent-id)]
+   parent-id           :- common/ParentID
+   reader writer]
+  (let [fields-to-reactivate (matching-inactive-fields table new-field-metadatas parent-id reader)]
     ;; if the fields already exist but were just marked inactive then reäctivate them
     (when (seq fields-to-reactivate)
-      (t2/update! :model/Field {:id [:in (map u/the-id fields-to-reactivate)]}
-                  {:active true}))
+      (persist/reactivate-fields! writer (map u/the-id fields-to-reactivate)))
     (let [reactivated?  (comp (set (map common/canonical-name fields-to-reactivate))
                               common/canonical-name)
           ;; If we reactivated the fields, no need to insert them; insert new rows for any that weren't reactivated
-          new-field-ids (insert-new-fields! table (remove reactivated? new-field-metadatas) parent-id)]
+          new-field-ids (insert-new-fields! table (remove reactivated? new-field-metadatas) parent-id writer)]
       ;; now return the newly created or reactivated Fields
       (when-let [new-and-updated-fields (seq (map u/the-id (concat fields-to-reactivate new-field-ids)))]
-        (t2/select :model/Field :id [:in new-and-updated-fields])))))
+        (persist/select-fields-by-ids reader new-and-updated-fields)))))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                          SYNCING INSTANCES OF 'ACTIVE' FIELDS (FIELDS IN DB METADATA)                          |
@@ -127,7 +129,8 @@
   [table        :- i/TableInstance
    db-metadata  :- [:set i/TableMetadataField]
    our-metadata :- [:set common/TableMetadataFieldWithID]
-   parent-id    :- common/ParentID]
+   parent-id    :- common/ParentID
+   reader writer]
   (let [known-fields (m/index-by common/canonical-name our-metadata)
         our-metadata (atom our-metadata)]
     {:num-updates
@@ -138,7 +141,7 @@
                                               (pr-str (map :name db-field-chunk)))
          (let [known-field?        (comp known-fields common/canonical-name)
                new-fields          (remove known-field? db-field-chunk)
-               new-field-instances (create-or-reactivate-fields! table new-fields parent-id)]
+               new-field-instances (create-or-reactivate-fields! table new-fields parent-id reader writer)]
            ;; save any updates to `our-metadata`
            (swap! our-metadata into (fields.our-metadata/fields->our-metadata new-field-instances parent-id))
            ;; now return count of rows updated
@@ -155,9 +158,10 @@
   "Mark an `old-field` belonging to `table` as inactive if corresponding Field object exists. Does *NOT* recurse over
   nested Fields. Returns `1` if a Field was marked inactive, `nil` otherwise."
   [table          :- i/TableInstance
-   metabase-field :- common/TableMetadataFieldWithID]
+   metabase-field :- common/TableMetadataFieldWithID
+   writer]
   (log/infof "Marking Field ''%s'' as inactive." (common/field-metadata-name-for-logging table metabase-field))
-  (when (pos? (t2/update! :model/Field (u/the-id metabase-field) {:active false}))
+  (when (pos? (persist/retire-field! writer (u/the-id metabase-field)))
     1))
 
 (mu/defn- retire-fields! :- ms/IntGreaterThanOrEqualToZero
@@ -166,13 +170,14 @@
   Returns `1` if a Field was marked inactive."
   [table        :- i/TableInstance
    db-metadata  :- [:set i/TableMetadataField]
-   our-metadata :- [:set common/TableMetadataFieldWithID]]
+   our-metadata :- [:set common/TableMetadataFieldWithID]
+   writer]
   ;; retire all the Fields not present in `db-metadata`, and count how many rows were actually affected
   (sync-util/sum-for [metabase-field our-metadata
                       :when          (not (common/matching-field-metadata metabase-field db-metadata))]
     (sync-util/with-error-handling (format "Error retiring %s"
                                            (common/field-metadata-name-for-logging table metabase-field))
-      (retire-field! table metabase-field))))
+      (retire-field! table metabase-field writer))))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                  HIGH-LEVEL INSTANCE SYNCING LOGIC (CREATING/REACTIVATING/RETIRING/UPDATING)                   |
@@ -185,7 +190,8 @@
   `field-metadata` (from synced DB) and `metabase-field` (from application DB)."
   [table          :- i/TableInstance
    field-metadata :- [:maybe i/TableMetadataField]
-   metabase-field :- [:maybe common/TableMetadataFieldWithID]]
+   metabase-field :- [:maybe common/TableMetadataFieldWithID]
+   reader writer]
   (let [nested-fields-metadata (:nested-fields field-metadata)
         metabase-nested-fields (:nested-fields metabase-field)]
     (when (or (seq nested-fields-metadata)
@@ -194,7 +200,8 @@
        table
        (set nested-fields-metadata)
        (set metabase-nested-fields)
-       (some-> metabase-field u/the-id)))))
+       (some-> metabase-field u/the-id)
+       reader writer))))
 
 (mu/defn- sync-nested-field-instances! :- [:maybe ms/IntGreaterThanOrEqualToZero]
   "Recursively sync Field instances (i.e., rows in application DB) for *all* the nested Fields of all Fields in
@@ -202,7 +209,8 @@
   Not for the flattened nested fields for JSON columns in normal RDBMSes (nested field columns)"
   [table        :- i/TableInstance
    db-metadata  :- [:set i/TableMetadataField]
-   our-metadata :- [:set common/TableMetadataFieldWithID]]
+   our-metadata :- [:set common/TableMetadataFieldWithID]
+   reader writer]
   (let [name->field-metadata (m/index-by common/canonical-name db-metadata)
         name->metabase-field (m/index-by common/canonical-name our-metadata)
         all-field-names      (set (concat (keys name->field-metadata)
@@ -210,7 +218,7 @@
     (sync-util/sum-for [field-name all-field-names
                         :let [field-metadata (get name->field-metadata field-name)
                               metabase-field (get name->metabase-field field-name)]]
-      (sync-nested-fields-of-one-field! table field-metadata metabase-field))))
+      (sync-nested-fields-of-one-field! table field-metadata metabase-field reader writer))))
 
 (mu/defn sync-instances! :- ms/IntGreaterThanOrEqualToZero
   "Sync rows in the Field table with `db-metadata` describing the current schema of the Table currently being synced,
@@ -218,12 +226,13 @@
   ([table        :- i/TableInstance
     db-metadata  :- [:set i/TableMetadataField]
     our-metadata :- [:set common/TableMetadataFieldWithID]]
-   (sync-instances! table db-metadata our-metadata nil))
+   (sync-instances! table db-metadata our-metadata nil persist.appdb/reader persist.appdb/writer))
 
   ([table        :- i/TableInstance
     db-metadata  :- [:set i/TableMetadataField]
     our-metadata :- [:set common/TableMetadataFieldWithID]
-    parent-id    :- common/ParentID]
+    parent-id    :- common/ParentID
+    reader writer]
    ;; syncing the active instances makes important changes to `our-metadata` that need to be passed to recursive
    ;; calls, such as adding new Fields or making inactive ones active again. Keep updated version returned by
    ;; `sync-active-instances!`
@@ -231,7 +240,7 @@
                (sync-util/name-for-logging table)
                (pr-str (sort (map common/canonical-name db-metadata)))
                (pr-str (sort (map common/canonical-name our-metadata))))
-   (let [{:keys [num-updates our-metadata]} (sync-active-instances! table db-metadata our-metadata parent-id)]
+   (let [{:keys [num-updates our-metadata]} (sync-active-instances! table db-metadata our-metadata parent-id reader writer)]
      (+ num-updates
-        (retire-fields! table db-metadata our-metadata)
-        (sync-nested-field-instances! table db-metadata our-metadata)))))
+        (retire-fields! table db-metadata our-metadata writer)
+        (sync-nested-field-instances! table db-metadata our-metadata reader writer)))))

src/metabase/sync/sync_metadata/fields/sync_metadata.clj

@@ -5,6 +5,8 @@
   (:require
    [clojure.string :as str]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.crufty :as crufty]
    [metabase.sync.sync-metadata.fields.common :as common]
    [metabase.sync.util :as sync-util]
@@ -12,8 +14,7 @@
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.schema :as ms]
-   [metabase.warehouse-schema.models.field-user-settings :as schema.field-user-settings]
-   [toucan2.core :as t2]))
+   [metabase.warehouse-schema.models.field-user-settings :as schema.field-user-settings]))
 
 (defn- crufty-field? [db field-metadata]
   (crufty/name? (:name field-metadata)
@@ -33,7 +34,8 @@
   [database       :- i/DatabaseInstance
    table          :- i/TableInstance
    field-metadata :- i/TableMetadataField
-   metabase-field :- common/TableMetadataFieldWithID]
+   metabase-field :- common/TableMetadataFieldWithID
+   writer]
   (let [{old-database-type              :database-type
          old-base-type                  :base-type
          old-field-comment              :field-comment
@@ -209,7 +211,7 @@
            {:preview_display false}))]
     ;; if any updates need to be done, do them and return 1 (because 1 Field was updated), otherwise return 0
     (if (and (seq updates)
-             (pos? (t2/update! :model/Field (u/the-id metabase-field) updates)))
+             (pos? (persist/update-field! writer (u/the-id metabase-field) updates)))
       1
       0)))
 
@@ -220,23 +222,30 @@
   [database       :- i/DatabaseInstance
    table          :- i/TableInstance
    field-metadata :- i/TableMetadataField
-   metabase-field :- common/TableMetadataFieldWithID]
+   metabase-field :- common/TableMetadataFieldWithID
+   writer]
   (let [nested-fields-metadata (:nested-fields field-metadata)
         metabase-nested-fields (:nested-fields metabase-field)]
     (if (seq metabase-nested-fields)
-      (update-metadata! database table (set nested-fields-metadata) (set metabase-nested-fields))
+      (update-metadata! database table (set nested-fields-metadata) (set metabase-nested-fields) writer)
       0)))
 
 (mu/defn update-metadata! :- ms/IntGreaterThanOrEqualToZero
   "Make sure things like PK status and base-type are in sync with what has come back from the DB. Recursively updates
   nested Fields. Returns total number of Fields updated."
-  [database     :- i/DatabaseInstance
-   table        :- i/TableInstance
-   db-metadata  :- [:set i/TableMetadataField]
-   our-metadata :- [:set common/TableMetadataFieldWithID]]
-  (sync-util/sum-for [metabase-field our-metadata]
-    ;; only update metadata for 'existing' Fields that are present in our Metadata (i.e., present in the application
-    ;; database) and that are still considered active (i.e., present in DB metadata)
-    (when-let [field-metadata (common/matching-field-metadata metabase-field db-metadata)]
-      (+ (update-field-metadata-if-needed! database table field-metadata metabase-field)
-         (update-nested-fields-metadata! database table field-metadata metabase-field)))))
+  ([database     :- i/DatabaseInstance
+    table        :- i/TableInstance
+    db-metadata  :- [:set i/TableMetadataField]
+    our-metadata :- [:set common/TableMetadataFieldWithID]]
+   (update-metadata! database table db-metadata our-metadata persist.appdb/writer))
+  ([database     :- i/DatabaseInstance
+    table        :- i/TableInstance
+    db-metadata  :- [:set i/TableMetadataField]
+    our-metadata :- [:set common/TableMetadataFieldWithID]
+    writer]
+   (sync-util/sum-for [metabase-field our-metadata]
+     ;; only update metadata for 'existing' Fields that are present in our Metadata (i.e., present in the application
+     ;; database) and that are still considered active (i.e., present in DB metadata)
+     (when-let [field-metadata (common/matching-field-metadata metabase-field db-metadata)]
+       (+ (update-field-metadata-if-needed! database table field-metadata metabase-field writer)
+          (update-nested-fields-metadata! database table field-metadata metabase-field writer))))))

src/metabase/sync/sync_metadata/fks.clj

@@ -1,99 +1,23 @@
 (ns metabase.sync.sync-metadata.fks
   "Logic for updating FK properties of Fields from metadata fetched from a physical DB."
   (:require
-   [honey.sql :as sql]
-   [metabase.app-db.core :as mdb]
    [metabase.driver.util :as driver.u]
    [metabase.sync.fetch-metadata :as fetch-metadata]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.util :as sync-util]
    [metabase.util :as u]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
-   [metabase.warehouse-schema.models.table :as table]
-   [toucan2.core :as t2]))
-
-(defn ^:private mark-fk-sql
-  "Returns [sql & params] for [[mark-fk!]] according to the application DB's dialect."
-  [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
-                          ;; in subqueries in MySQL) because it's possible for schema, table, or column names to be
-                          ;; non-unique when lower-cased for some DBs. We have been doing case-insensitive matching
-                          ;; since #5510 so this preserves behaviour to avoid possible regressions.
-                          ;; It's possible this is to avoid
-                          :from   [[:metabase_field :f]]
-                          :join   [[:metabase_table :t] [:= :f.table_id :t.id]]
-                          :left-join [[:metabase_field_user_settings :u] [:= :f.id :u.field_id]]
-                          :where  [:and
-                                   ;; ensure we are not overriding user-set fks
-                                   [:= :u.fk_target_field_id nil]
-                                   [:= :u.semantic_type nil]
-
-                                   [:= :t.db_id db-id]
-                                   [:= [:lower :f.name] (u/lower-case-en column-name)]
-                                   [:= [:lower :t.name] (u/lower-case-en table-name)]
-                                   [:= [:lower :t.schema] (some-> table-schema u/lower-case-en)]
-                                   [:= :f.active true]
-                                   [:not= :f.visibility_type "retired"]
-                                   [:= :t.active true]
-                                   [:= :t.visibility_type nil]]})
-        fk-field-id-query (field-id-query db-id fk-table-schema fk-table-name fk-column-name)
-        pk-field-id-query (field-id-query db-id pk-table-schema pk-table-name pk-column-name)
-
-        ;; Only update if either:
-        ;; - fk_target_field_id is NULL and the new target is not NULL
-        ;; - fk_target_field_id is not NULL but the new target is different and not NULL
-        valid-condition
-        (fn [k]
-          [:or
-           [:= :f.fk_target_field_id nil]
-           [:not= :f.fk_target_field_id k]])
-
-        q (case (mdb/db-type)
-            :mysql
-            {:update [:metabase_field :f]
-             :join   [[fk-field-id-query :fk] [:= :fk.id :f.id]
-                      [pk-field-id-query :pk]
-                      (valid-condition :pk.id)]
-             :set    {:fk_target_field_id :pk.id
-                      ;; We need to reset has_field_values when it is auto-list as FKs should not be marked as such
-                      :has_field_values   [:case [:= :has_field_values "auto-list"] nil :else :has_field_values]
-                      :semantic_type      "type/FK"}}
-            :postgres
-            {:update [:metabase_field :f]
-             :from   [[fk-field-id-query :fk]]
-             :join   [[pk-field-id-query :pk] true]
-             :set    {:fk_target_field_id :pk.id
-                      ;; We need to reset has_field_values when it is auto-list as FKs should not be marked as such
-                      :has_field_values   [:case [:= :has_field_values "auto-list"] nil :else :has_field_values]
-                      :semantic_type      "type/FK"}
-             :where  [:and
-                      [:= :fk.id :f.id]
-                      (valid-condition :pk.id)]}
-            :h2
-            {:update [:metabase_field :f]
-             :set    {:fk_target_field_id pk-field-id-query
-                      ;; We need to reset has_field_values when it is auto-list as FKs should not be marked as such
-                      :has_field_values   [:case [:= :has_field_values "auto-list"] nil :else :has_field_values]
-                      :semantic_type      "type/FK"}
-             :where  [:and
-                      [:= :f.id fk-field-id-query]
-                      [:not= pk-field-id-query nil]
-                      (valid-condition pk-field-id-query)]})]
-    (sql/format q :dialect (mdb/quoting-style (mdb/db-type)))))
+   [metabase.warehouse-schema.models.table :as table]))
 
 (mu/defn- mark-fk!
   "Updates the `fk_target_field_id` of a Field. Returns 1 if the Field was successfully updated, 0 otherwise."
   [database :- i/DatabaseInstance
-   metadata :- i/FKMetadataEntry]
-  (u/prog1 (t2/query-one (mark-fk-sql (:id database) metadata))
+   metadata :- i/FKMetadataEntry
+   writer]
+  (u/prog1 (persist/mark-fk! writer (: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)
@@ -110,12 +34,17 @@
 
   ([database :- i/DatabaseInstance
     table    :- i/TableInstance]
+   (sync-fks-for-table! database table persist.appdb/writer))
+
+  ([database :- i/DatabaseInstance
+    table    :- i/TableInstance
+    writer]
    (sync-util/with-error-handling (format "Error syncing FKs for %s" (sync-util/name-for-logging table))
      (let [schema-names (when (driver.u/supports? (driver.u/database->driver database) :schemas database)
                           [(:schema table)])
            fk-metadata  (into [] (fetch-metadata/fk-metadata database :schema-names schema-names :table-names [(:name table)]))]
        {:total-fks   (count fk-metadata)
-        :updated-fks (sync-util/sum-numbers #(mark-fk! database %) fk-metadata)}))))
+        :updated-fks (sync-util/sum-numbers #(mark-fk! database % writer) fk-metadata)}))))
 
 (mu/defn sync-fks!
   "Sync the foreign keys in a `database`. This sets appropriate values for relevant Fields in the Metabase application
@@ -124,26 +53,28 @@
   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]
-  (u/prog1 (sync-util/with-error-handling (format "Error syncing FKs for %s" (sync-util/name-for-logging database))
-             (let [driver       (driver.u/database->driver database)
-                   schema-names (when (driver.u/supports? driver :schemas database)
-                                  (sync-util/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)))
-    ;; 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)))
+  ([database :- i/DatabaseInstance]
+   (sync-fks! database persist.appdb/writer))
+  ([database :- i/DatabaseInstance writer]
+   (u/prog1 (sync-util/with-error-handling (format "Error syncing FKs for %s" (sync-util/name-for-logging database))
+              (let [driver       (driver.u/database->driver database)
+                    schema-names (when (driver.u/supports? driver :schemas database)
+                                   (sync-util/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 writer) 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)))
+     ;; 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/sync_metadata/indexes.clj

@@ -4,10 +4,11 @@
    [metabase.driver :as driver]
    [metabase.driver.util :as driver.u]
    [metabase.sync.fetch-metadata :as fetch-metadata]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.util :as sync-util]
    [metabase.util.log :as log]
-   [metabase.warehouse-schema.models.field :as field]
-   [toucan2.core :as t2]))
+   [metabase.warehouse-schema.models.field :as field]))
 
 (def ^:private empty-stats
   {:total-indexes 0
@@ -15,74 +16,51 @@
    :removed-indexes 0})
 
 (defn- indexes->field-ids
-  [table-id indexes]
+  [table-id indexes reader]
   (when (seq indexes)
     (let [normal-indexes           (->> indexes (filter #(= (:type %) :normal-column-index)) (map :value))
           nested-indexes           (->> indexes (filter #(= (:type %) :nested-column-index)) (map :value))
           normal-indexes-field-ids (when (seq normal-indexes)
-                                     (t2/select-pks-vec :model/Field :name [:in normal-indexes] :table_id table-id :parent_id nil))
+                                    (persist/field-ids-for-index-names reader table-id normal-indexes))
           nested-indexes-field-ids (remove nil? (map #(field/nested-field-names->field-id table-id %) nested-indexes))]
       (set (filter some? (concat normal-indexes-field-ids nested-indexes-field-ids))))))
 
 (defn maybe-sync-indexes-for-table!
   "Sync the indexes for `table` if the driver supports storing index info."
-  [database table]
-  (if (driver.u/supports? (driver.u/database->driver database) :index-info database)
-    (sync-util/with-error-handling (format "Error syncing Indexes for %s" (sync-util/name-for-logging table))
-      (let [indexes                    (fetch-metadata/index-metadata database table)
-            indexed-field-ids          (indexes->field-ids (:id table) indexes)
-            existing-indexed-field-ids (t2/select-pks-set :model/Field :table_id (:id table) :database_indexed true)
-            [removing adding]          (data/diff existing-indexed-field-ids indexed-field-ids)]
-        (doseq [field-id removing]
-          (log/infof "Unmarking Field %d as indexed" field-id))
-        (doseq [field-id adding]
-          (log/infof "Marking Field %d as indexed" field-id))
-        (if (or (seq adding) (seq removing))
-          (do (t2/update! :model/Field {:table_id (:id table)}
-                          {:database_indexed (if (seq indexed-field-ids)
-                                               [:case [:in :id indexed-field-ids] true :else false]
-                                               false)})
-              {:total-indexes   (count indexed-field-ids)
-               :added-indexes   (count adding)
-               :removed-indexes (count removing)})
-          empty-stats)))
-    empty-stats))
-
-(defn- all-indexes->field-ids
-  [database-id indexes]
-  (reduce
-   (fn [accum index-batch]
-     (let [normal-indexes (map (juxt #(:table-schema % "__null__") :table-name :field-name) index-batch)
-           query (t2/reducible-query {:select [[:f.id]]
-                                      :from [[(t2/table-name :model/Field) :f]]
-                                      :inner-join [[(t2/table-name :model/Table) :t] [:= :f.table_id :t.id]]
-                                      :where [:and [:in [:composite [:coalesce :t.schema "__null__"] :t.name :f.name] normal-indexes]
-                                              [:= :t.db_id database-id]
-                                              [:= :parent_id nil]]})]
-       (into accum (keep :id) query)))
-   #{}
-   ;; break the indexes up in groups of 5000 to avoid max
-   ;; parameter limit of 65,535. See #52746 for details.
-   (partition-all 5000 indexes)))
+  ([database table]
+   (maybe-sync-indexes-for-table! database table persist.appdb/reader persist.appdb/writer))
+  ([database table reader writer]
+   (if (driver.u/supports? (driver.u/database->driver database) :index-info database)
+     (sync-util/with-error-handling (format "Error syncing Indexes for %s" (sync-util/name-for-logging table))
+       (let [indexes                    (fetch-metadata/index-metadata database table)
+             indexed-field-ids          (indexes->field-ids (:id table) indexes reader)
+             existing-indexed-field-ids (persist/indexed-field-ids-for-table reader (:id table))
+             [removing adding]          (data/diff existing-indexed-field-ids indexed-field-ids)]
+         (doseq [field-id removing]
+           (log/infof "Unmarking Field %d as indexed" field-id))
+         (doseq [field-id adding]
+           (log/infof "Marking Field %d as indexed" field-id))
+         (if (or (seq adding) (seq removing))
+           (do (persist/set-table-indexes! writer (:id table) indexed-field-ids)
+               {:total-indexes   (count indexed-field-ids)
+                :added-indexes   (count adding)
+                :removed-indexes (count removing)})
+           empty-stats)))
+     empty-stats)))
 
 (def ^:dynamic *update-partition-size*
   "Size of the partition of indexes to update using one `t2/update!` call. Dynamic for testing purposes."
   5000)
 
 (defn- sync-all-indexes!
-  [database]
+  [database reader writer]
   (sync-util/with-error-handling "Error syncing Indexes"
     (let [indexes (fetch-metadata/log-if-error
                    "index-metadata"
                     (into [] (driver/describe-indexes (driver.u/database->driver database) database)))
           database-id (:id database)
-          indexed-field-ids (all-indexes->field-ids database-id indexes)
-          existing-indexed-field-ids (t2/select-pks-set :model/Field
-                                                        :table_id [:in {:select [[:t.id]]
-                                                                        :from [[(t2/table-name :model/Table) :t]]
-                                                                        :where [:= :t.db_id database-id]}]
-                                                        :parent_id nil
-                                                        :database_indexed true)
+          indexed-field-ids (persist/field-ids-for-indexes reader database-id indexes)
+          existing-indexed-field-ids (persist/indexed-field-ids-for-database reader database-id)
           [removing adding]           (data/diff existing-indexed-field-ids indexed-field-ids)
           removing-count              (count removing)
           adding-count                (count adding)]
@@ -92,14 +70,14 @@
         (log/tracef "Unmarking Fields as indexed: %s" (pr-str field-ids)))
       (doseq [field-ids (partition-all *update-partition-size* removing)]
         (log/infof "Executing batch update of at most %d fields" *update-partition-size*)
-        (t2/update! :model/Field :parent_id nil :id [:in field-ids] {:database_indexed false}))
+        (persist/batch-set-indexed! writer field-ids false))
       ;; Set database_indexed of fields having index.
       (log/infof "Marking %d fields as indexed" adding-count)
       (doseq [field-ids (partition-all 100 adding)]
         (log/tracef "Marking Fields as indexed: %s" (pr-str field-ids)))
       (doseq [field-ids (partition-all *update-partition-size* adding)]
         (log/infof "Executing batch update of at most %d fields" *update-partition-size*)
-        (t2/update! :model/Field :parent_id nil :id [:in field-ids] {:database_indexed true}))
+        (persist/batch-set-indexed! writer field-ids true))
       (if (or (seq adding) (seq removing))
         {:total-indexes   (count indexed-field-ids)
          :added-indexes   adding-count
@@ -108,12 +86,14 @@
 
 (defn maybe-sync-indexes!
   "Sync the indexes for all tables in `database` if the driver supports storing index info."
-  [database]
-  (if (driver.u/supports? (driver.u/database->driver database) :index-info database)
-    (if (driver.u/supports? (driver.u/database->driver database) :describe-indexes database)
-      (sync-all-indexes! database)
-      (transduce (map #(maybe-sync-indexes-for-table! database %))
-                 (partial merge-with +)
-                 empty-stats
-                 (sync-util/reducible-sync-tables database)))
-    empty-stats))
+  ([database]
+   (maybe-sync-indexes! database persist.appdb/reader persist.appdb/writer))
+  ([database reader writer]
+   (if (driver.u/supports? (driver.u/database->driver database) :index-info database)
+     (if (driver.u/supports? (driver.u/database->driver database) :describe-indexes database)
+       (sync-all-indexes! database reader writer)
+       (transduce (map #(maybe-sync-indexes-for-table! database % reader writer))
+                  (partial merge-with +)
+                  empty-stats
+                  (sync-util/reducible-sync-tables database)))
+     empty-stats)))

src/metabase/sync/sync_metadata/metabase_metadata.clj

@@ -11,12 +11,13 @@
    [metabase.driver.util :as driver.u]
    [metabase.sync.fetch-metadata :as fetch-metadata]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.util :as sync-util]
    [metabase.util :as u]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
-   [metabase.util.malli.schema :as ms]
-   [toucan2.core :as t2]))
+   [metabase.util.malli.schema :as ms]))
 
 (def ^:private KeypathComponents
   [:map
@@ -41,21 +42,18 @@
   "Set a property for a Field or Table in `database`. Returns `true` if a property was successfully set."
   [database                          :- i/DatabaseInstance
    {:keys [table-name field-name k]} :- KeypathComponents
-   value]
+   value
+   reader writer]
   (boolean
    ;; ignore legacy entries that try to set field_type since it's no longer part of Field
    (when-not (= k :field_type)
      ;; fetch the corresponding Table, then set the Table or Field property
      (if table-name
-       (when-let [table-id (t2/select-one-pk :model/Table
-                                             ;; TODO: this needs to support schemas
-                                             :db_id  (u/the-id database)
-                                             :name   table-name
-                                             :active true)]
+       (when-let [table-id (persist/find-active-table-id reader (u/the-id database) table-name)]
          (if field-name
-           (t2/update! :model/Field {:name field-name, :table_id table-id} {k value})
-           (t2/update! :model/Table table-id {k value})))
-       (t2/update! :model/Database (u/the-id database) {k value})))))
+           (persist/update-field-by-name! writer table-id field-name {k value})
+           (persist/update-table! writer table-id {k value})))
+       (persist/update-database! writer (u/the-id database) {k value})))))
 
 (mu/defn- sync-metabase-metadata-table!
   "Databases may include a table named `_metabase_metadata` (case-insensitive) which includes descriptions or other
@@ -77,10 +75,11 @@
   implement optional fn `:table-rows-seq`."
   [driver
    database                :- i/DatabaseInstance
-   metabase-metadata-table :- i/DatabaseMetadataTable]
+   metabase-metadata-table :- i/DatabaseMetadataTable
+   reader writer]
   (doseq [{:keys [keypath value]} (driver/table-rows-seq driver database metabase-metadata-table)]
     (sync-util/with-error-handling (format "Error handling metabase metadata entry: set %s -> %s" keypath value)
-      (or (set-property! database (parse-keypath keypath) value)
+      (or (set-property! database (parse-keypath keypath) value reader writer)
           (log/error (u/format-color 'red "Error syncing _metabase_metadata: no matching keypath: %s" keypath))))))
 
 (mu/defn is-metabase-metadata-table?
@@ -96,6 +95,9 @@
    (sync-metabase-metadata! database (fetch-metadata/db-metadata database)))
 
   ([database :- i/DatabaseInstance db-metadata]
+   (sync-metabase-metadata! database db-metadata persist.appdb/reader persist.appdb/writer))
+
+  ([database :- i/DatabaseInstance db-metadata reader writer]
    (sync-util/with-error-handling (format "Error syncing _metabase_metadata table for %s"
                                           (sync-util/name-for-logging database))
      (let [driver (driver.u/database->driver database)]
@@ -105,5 +107,5 @@
          ;; Hopefully this is never the case.
          (doseq [table (:tables db-metadata)]
            (when (is-metabase-metadata-table? table)
-             (sync-metabase-metadata-table! driver database table))))
+             (sync-metabase-metadata-table! driver database table reader writer))))
        {}))))

src/metabase/sync/sync_metadata/sync_timezone.clj

@@ -6,10 +6,11 @@
    [metabase.lib.schema.expression.temporal
     :as lib.schema.expression.temporal]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.util.i18n :refer [trs]]
    [metabase.util.log :as log]
-   [metabase.util.malli :as mu]
-   [toucan2.core :as t2]))
+   [metabase.util.malli :as mu]))
 
 (set! *warn-on-reflection* true)
 
@@ -36,13 +37,16 @@
 (mu/defn sync-timezone! :- [:map [:timezone-id [:maybe ::lib.schema.expression.temporal/timezone-id]]]
   "Query `database` for its current time to determine its timezone. The results of this function are used by the sync
   process to update the timezone if it's different."
-  [database :- i/DatabaseInstance]
-  (let [driver  (driver.u/database->driver database)
-        zone-id (driver/db-default-timezone driver database)]
-    (validate-zone-id driver zone-id)
-    (let [zone-id (some-> zone-id str)
-          zone-id (if (= zone-id "Z") "UTC" zone-id)]
-      (log/infof "%s database %s default timezone is %s" driver (pr-str (:id database)) (pr-str zone-id))
-      (when-not (= zone-id (:timezone database))
-        (t2/update! :model/Database (:id database) {:timezone zone-id}))
-      {:timezone-id zone-id})))
+  ([database :- i/DatabaseInstance]
+   (sync-timezone! database persist.appdb/writer))
+  ([database :- i/DatabaseInstance
+    writer   :- [:fn #(satisfies? persist/SyncDatabaseWriter %)]]
+   (let [driver  (driver.u/database->driver database)
+         zone-id (driver/db-default-timezone driver database)]
+     (validate-zone-id driver zone-id)
+     (let [zone-id (some-> zone-id str)
+           zone-id (if (= zone-id "Z") "UTC" zone-id)]
+       (log/infof "%s database %s default timezone is %s" driver (pr-str (:id database)) (pr-str zone-id))
+       (when-not (= zone-id (:timezone database))
+         (persist/set-database-timezone! writer (:id database) zone-id))
+       {:timezone-id zone-id}))))

src/metabase/sync/sync_metadata/tables.clj

@@ -13,14 +13,15 @@
    [metabase.models.interface :as mi]
    [metabase.sync.fetch-metadata :as fetch-metadata]
    [metabase.sync.interface :as i]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.crufty :as crufty]
    [metabase.sync.sync-metadata.metabase-metadata :as metabase-metadata]
    [metabase.sync.util :as sync-util]
    [metabase.util :as u]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
-   [metabase.util.malli.schema :as ms]
-   [toucan2.core :as t2]))
+   [metabase.util.malli.schema :as ms]))
 
 (set! *warn-on-reflection* true)
 
@@ -86,11 +87,10 @@
 (mu/defn- update-database-metadata!
   "If there is a version in the db-metadata update the DB to have that in the DB model"
   [database    :- i/DatabaseInstance
-   db-metadata :- i/DatabaseMetadata]
+   db-metadata :- i/DatabaseMetadata
+   writer      :- [:fn #(satisfies? persist/SyncDatabaseWriter %)]]
   (log/infof "Found new version for DB: %s" (:version db-metadata))
-  (t2/update! :model/Database (u/the-id database)
-              {:details
-               (assoc (:details database) :version (:version db-metadata))}))
+  (persist/set-database-details-version! writer (u/the-id database) (:details database) (:version db-metadata)))
 
 (mu/defn- cruft-dependent-cols [{table-name :name :as table}
                                 database
@@ -119,82 +119,80 @@
                             :else                   "incomplete")
      :visibility_type     (when is-crufty? :cruft)}))
 
+(defn- build-table-map
+  "Build the map to insert for a new table."
+  [database table]
+  (merge (cruft-dependent-cols table database ::create)
+         {:active                  true
+          :db_id                   (:id database)
+          :schema                  (:schema table)
+          :description             (:description table)
+          :database_require_filter (:database_require_filter table)
+          :display_name            (or (:display_name table)
+                                       (humanization/name->human-readable-name (:name table)))
+          :name                    (:name table)
+          :is_writable             (:is_writable table)}
+         (when (:data_source table)
+           {:data_source (:data_source table)})
+         (when (:data_authority table)
+           {:data_authority (:data_authority table)})
+         (when (:is_sample database)
+           {:data_authority :ingested
+            :data_source    :ingested})))
+
 (defn create-table!
   "Creates a new table in the database, ready to be synced.
    Throws an exception if there is already a table with the same name, schema and database ID."
-  [database table]
-  (t2/insert-returning-instance!
-   :model/Table
-   (merge (cruft-dependent-cols table database ::create)
-          {:active                  true
-           :db_id                   (:id database)
-           :schema                  (:schema table)
-           :description             (:description table)
-           :database_require_filter (:database_require_filter table)
-           :display_name            (or (:display_name table)
-                                        (humanization/name->human-readable-name (:name table)))
-           :name                    (:name table)
-           :is_writable             (:is_writable table)}
-          (when (:data_source table)
-            {:data_source (:data_source table)})
-          (when (:data_authority table)
-            {:data_authority (:data_authority table)})
-          (when (:is_sample database)
-            {:data_authority :ingested
-             :data_source    :ingested}))))
+  [database table writer]
+  (persist/create-table! writer (build-table-map database table)))
 
 (defn create-or-reactivate-table!
   "Create a single new table in the database, or mark it as active if it already exists."
-  [database {schema :schema table-name :name :as table}]
-  (if-let [existing-id (t2/select-one-pk :model/Table
-                                         :db_id (u/the-id database)
-                                         :schema schema
-                                         :name table-name
-                                         :active false)]
-    (let [table (t2/select-one :model/Table existing-id)]
+  [database {schema :schema table-name :name :as table} reader writer]
+  (if-let [existing-id (persist/find-inactive-table-id reader (u/the-id database) schema table-name)]
+    (let [table (persist/get-table reader existing-id)]
       ;; if the table already exists but is marked *inactive*, mark it as *active*
-      (t2/update! :model/Table existing-id (cond-> (cruft-dependent-cols table database ::reactivate)
+      (persist/reactivate-table! writer existing-id
+                                 (cond-> (cruft-dependent-cols table database ::reactivate)
 
-                                             ;; do not unhide tables w/ cruft settings
-                                             (some? (:visibility_type table))
-                                             (dissoc :visibility_type)
+                                   ;; do not unhide tables w/ cruft settings
+                                   (some? (:visibility_type table))
+                                   (dissoc :visibility_type)
 
-                                             true
-                                             (assoc :active true)
+                                   true
+                                   (assoc :active true)
 
-                                             (:is_sample database)
-                                             (assoc :data_authority :ingested
-                                                    :data_source    :ingested))))
+                                   (:is_sample database)
+                                   (assoc :data_authority :ingested
+                                          :data_source    :ingested))))
     ;; otherwise create a new Table
-    (create-table! database table)))
+    (create-table! database table writer)))
 
 ;; TODO - should we make this logic case-insensitive like it is for fields?
 
 (mu/defn- create-or-reactivate-tables!
   "Create `new-tables` for database, or if they already exist, mark them as active."
   [database :- i/DatabaseInstance
-   new-table-metadatas :- [:set i/DatabaseMetadataTable]]
+   new-table-metadatas :- [:set i/DatabaseMetadataTable]
+   reader writer]
   (doseq [table-metadata new-table-metadatas]
     (log/info "Found new table:"
               (sync-util/name-for-logging (mi/instance :model/Table table-metadata))))
   (doseq [table-metadata new-table-metadatas]
-    (create-or-reactivate-table! database table-metadata)))
+    (create-or-reactivate-table! database table-metadata reader writer)))
 
 (mu/defn- retire-tables!
   "Mark any `old-tables` belonging to `database` as inactive."
   [database   :- i/DatabaseInstance
    old-tables :- [:set [:map
                         [:name ::lib.schema.common/non-blank-string]
-                        [:schema [:maybe ::lib.schema.common/non-blank-string]]]]]
+                        [:schema [:maybe ::lib.schema.common/non-blank-string]]]]
+   writer]
   (log/info "Marking tables as inactive:"
             (for [table old-tables]
               (sync-util/name-for-logging (mi/instance :model/Table table))))
   (doseq [{schema :schema table-name :name :as _table} old-tables]
-    (t2/update! :model/Table {:db_id  (u/the-id database)
-                              :schema schema
-                              :name   table-name
-                              :active true}
-                {:active false})))
+    (persist/retire-table! writer (u/the-id database) schema table-name)))
 
 (def ^:private keys-to-update
   [:description :database_require_filter :estimated_row_count :visibility_type :initial_sync_status :is_writable])
@@ -203,7 +201,8 @@
   "Update the table metadata if it has changed."
   [table-metadata :- i/DatabaseMetadataTable
    metabase-table :- (ms/InstanceOf :model/Table)
-   metabase-database :- (ms/InstanceOf :model/Database)]
+   metabase-database :- (ms/InstanceOf :model/Database)
+   writer]
   (let [old-table               (select-keys metabase-table keys-to-update)
         new-table               (-> (zipmap keys-to-update (repeat nil))
                                     (merge table-metadata
@@ -235,18 +234,20 @@
                  (get metabase-table k)
                  v))
     (when (seq changes)
-      (t2/update! :model/Table (:id metabase-table) changes))))
+      (persist/update-table! writer (:id metabase-table) changes))))
 
 (mu/defn- update-tables-metadata-if-needed!
   [table-metadatas :- [:set i/DatabaseMetadataTable]
    metabase-tables :- [:set (ms/InstanceOf :model/Table)]
-   metabase-database :- (ms/InstanceOf :model/Database)]
+   metabase-database :- (ms/InstanceOf :model/Database)
+   writer]
   (let [name+schema->table-metadata (m/index-by (juxt :name :schema) table-metadatas)
         name+schema->metabase-table (m/index-by (juxt :name :schema) metabase-tables)]
     (doseq [name+schema (set/intersection (set (keys name+schema->table-metadata)) (set (keys name+schema->metabase-table)))]
       (update-table-metadata-if-needed! (name+schema->table-metadata name+schema)
                                         (name+schema->metabase-table name+schema)
-                                        metabase-database))))
+                                        metabase-database
+                                        writer))))
 
 (mu/defn- table-set :- [:set i/DatabaseMetadataTable]
   "So there exist tables for the user and metabase metadata tables for internal usage by metabase.
@@ -258,25 +259,17 @@
                       (sync-util/is-temp-transform-table? table))))
         (:tables db-metadata)))
 
-(mu/defn- select-tables :- [:set (ms/InstanceOf :model/Table)]
-  "Selects the columns we need for `:model/Table`, with some optional filters"
-  [database :- i/DatabaseInstance
-   & filters]
-  (set (apply
-        t2/select
-        (into [:model/Table :id :name :schema :data_authority] keys-to-update)
-        :db_id (u/the-id database)
-        filters)))
-
 (mu/defn- db->our-metadata :- [:set (ms/InstanceOf :model/Table)]
   "Return information about what Tables we have for this DB in the Metabase application DB. Only includes active tables."
-  [database :- i/DatabaseInstance]
-  (select-tables database :active true))
+  [database :- i/DatabaseInstance
+   reader]
+  (persist/active-tables reader (u/the-id database)))
 
 (mu/defn- db->our-tables :- [:set (ms/InstanceOf :model/Table)]
   "Return *all* tables we have for this DB in the Metabase appDB, including inactive ones."
-  [database :- i/DatabaseInstance]
-  (select-tables database))
+  [database :- i/DatabaseInstance
+   reader]
+  (persist/all-tables reader (u/the-id database)))
 
 (mu/defn- adjusted-schemas :- [:maybe [:map-of :string :string]]
   "Returns a map of schemas that should be adjusted to their new names."
@@ -292,14 +285,11 @@
    (into #{} (map :schema our-tables))))
 
 (defn- adjust-table-schemas!
-  [database schemas-to-update]
+  [database schemas-to-update writer]
   (when schemas-to-update
     (log/infof "Renaming schemas: %s" (pr-str schemas-to-update)))
   (doseq [[schema new-schema] schemas-to-update]
-    (t2/update! :model/Table
-                :db_id (:id database)
-                :schema schema
-                {:schema new-schema})))
+    (persist/update-table-schema! writer (:id database) schema new-schema)))
 
 (def ^:private
   ^{:doc "threshold after which deactivated tables will be archived"}
@@ -309,7 +299,7 @@
   "Mark tables that have been deactivated for longer than the configured threshold as archived
   and suffixes their names. Skips tables with `transform_target = true` (provisional transform
   output entries) since transforms still reference them by their original name."
-  [database]
+  [database reader writer]
   (let [;; we use UTC offset time for suffix, may not match db time but
         ;; it doesn't matter much, the source of time truth is `archived_at`,
         ;; we're just using this as a cheap namespace
@@ -317,12 +307,7 @@
         threshold-expr (apply
                         (requiring-resolve 'metabase.util.honey-sql-2/add-interval-honeysql-form)
                         (mdb/db-type) :%now archive-tables-threshold)
-        tables-to-archive (t2/select :model/Table
-                                     :db_id (u/the-id database)
-                                     :active false
-                                     :archived_at nil
-                                     :transform_target false
-                                     :deactivated_at [:< threshold-expr])
+        tables-to-archive (persist/archivable-tables reader (u/the-id database) threshold-expr)
         archived (atom 0)]
     (doseq [table tables-to-archive
             :let [new-name (str (:name table) suffix)]]
@@ -339,11 +324,9 @@
                   ;; in the extremely unlikely case that there already exists a table with our
                   ;; archived name, we let it fail from hitting the unique constraints violation
                   ;; and just report the failure
-                  [(t2/update! :model/Table
-                               {:id (:id table)
-                                :active false}
-                               {:archived_at (mi/now)
-                                :name new-name})]
+                  [(persist/archive-table! writer (:id table)
+                                           {:archived_at (mi/now)
+                                            :name new-name})]
                   (catch Throwable t
                     [0 t]))]
             (when (zero? did-update)
@@ -361,16 +344,19 @@
    (sync-tables-and-database! database (fetch-metadata/db-metadata database)))
 
   ([database :- i/DatabaseInstance db-metadata]
+   (sync-tables-and-database! database db-metadata persist.appdb/reader persist.appdb/writer))
+
+  ([database :- i/DatabaseInstance db-metadata reader writer]
    ;; determine what's changed between what info we have and what's in the DB
    (let [driver                (driver.u/database->driver database)
          db-table-metadatas    (table-set db-metadata)
          name+schema           #(select-keys % [:name :schema])
          name+schema->db-table (m/index-by name+schema db-table-metadatas)
-         our-metadata          (db->our-metadata database)
+         our-metadata          (db->our-metadata database reader)
          multi-level-support?  (driver.u/supports? driver :multi-level-schema database)
          schemas-to-update     (when multi-level-support?
                                  ;; we want to adjust the schemas for all tables (not just active ones from `our-metadata`)
-                                 (adjusted-schemas driver database (db->our-tables database)))
+                                 (adjusted-schemas driver database (db->our-tables database reader)))
          our-metadata          (cond->> our-metadata
                                  multi-level-support?
                                  (into #{} (map (fn [table]
@@ -383,31 +369,31 @@
                                 (keep-name+schema-set (set (map name+schema our-metadata))))]
      (sync-util/with-error-handling (format "Error updating table schemas for %s"
                                             (sync-util/name-for-logging database))
-       (adjust-table-schemas! database schemas-to-update))
+       (adjust-table-schemas! database schemas-to-update writer))
 
      ;; update database metadata from database
      (when (some? (:version db-metadata))
        (sync-util/with-error-handling (format "Error creating/reactivating tables for %s"
                                               (sync-util/name-for-logging database))
-         (update-database-metadata! database db-metadata)))
+         (update-database-metadata! database db-metadata writer)))
      ;; create new tables as needed or mark them as active again
      (when (seq new-table-metadatas)
        (let [new-tables-info (set (map #(get name+schema->db-table (name+schema %)) new-table-metadatas))]
          (sync-util/with-error-handling (format "Error creating/reactivating tables for %s"
                                                 (sync-util/name-for-logging database))
-           (create-or-reactivate-tables! database new-tables-info))))
+           (create-or-reactivate-tables! database new-tables-info reader writer))))
      ;; mark old tables as inactive
      (when (seq old-table-metadatas)
        (sync-util/with-error-handling (format "Error retiring tables for %s" (sync-util/name-for-logging database))
-         (retire-tables! database old-table-metadatas)))
+         (retire-tables! database old-table-metadatas writer)))
 
      (sync-util/with-error-handling (format "Error updating table metadata for %s" (sync-util/name-for-logging database))
        ;; we need to fetch the tables again because we might have retired tables in the previous steps
-       (update-tables-metadata-if-needed! db-table-metadatas (db->our-metadata database) database))
+       (update-tables-metadata-if-needed! db-table-metadatas (db->our-metadata database reader) database writer))
 
      (let [archived-tables (sync-util/with-error-handling (format "Error archiving tables for %s"
                                                                   (sync-util/name-for-logging database))
-                             (archive-tables! database))]
+                             (archive-tables! database reader writer))]
 
        {:updated-tables (+ (count new-table-metadatas) (count old-table-metadatas) (or archived-tables 0))
         :total-tables   (count our-metadata)}))))

test/metabase/sync/sync_metadata/fields/sync_metadata_test.clj

@@ -3,6 +3,7 @@
    [clojure.test :refer :all]
    [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
    [metabase.sync.core :as sync]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.fields.sync-metadata :as sync-metadata]
    [metabase.test :as mt]
    [metabase.util :as u]
@@ -27,7 +28,8 @@
           db
           table
           new-metadata-from-sync
-          metadata-in-application-db)
+          metadata-in-application-db
+          persist.appdb/writer)
          @update-operations)))))
 
 (deftest database-type-changed-test

test/metabase/sync/sync_metadata/indexes_test.clj

@@ -6,6 +6,8 @@
    [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
    [metabase.driver.sql.util :as sql.u]
    [metabase.sync.core :as sync]
+   [metabase.sync.persist :as persist]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.indexes :as sync.indexes]
    [metabase.test :as mt]
    [metabase.test.data.interface :as tx]
@@ -60,7 +62,7 @@
             many-indexes (into indexes (repeat 100000 {:table-schema "public",
                                                        :table-name "fake_table",
                                                        :field-name "id"}))
-            field-ids (#'sync.indexes/all-indexes->field-ids (mt/id) many-indexes)]
+            field-ids (persist/field-ids-for-indexes persist.appdb/reader (mt/id) many-indexes)]
         (is (seq field-ids))))))
 
 (deftest sync-all-indexes!-test
@@ -93,7 +95,7 @@
                   :let [sql (sql.tx/create-index-sql driver/*driver* "first_table" [field])]]
             (jdbc/execute! (sql-jdbc.conn/db->pooled-connection-spec (mt/db)) sql))
           (binding [sync.indexes/*update-partition-size* 2]
-            (#'sync.indexes/sync-all-indexes! (mt/db)))
+            (#'sync.indexes/sync-all-indexes! (mt/db) persist.appdb/reader persist.appdb/writer))
           (is (every? :database_indexed
                       (t2/select :model/Field
                                  {:where [:in :table_id (t2/select-fn-vec :id :model/Table :db_id (mt/id))]}))))
@@ -103,7 +105,7 @@
                                                       driver/*driver* :index (str "idx_first_table_" field)))]]
             (jdbc/execute! (sql-jdbc.conn/db->pooled-connection-spec (mt/db)) sql))
           (binding [sync.indexes/*update-partition-size* 2]
-            (#'sync.indexes/sync-all-indexes! (mt/db)))
+            (#'sync.indexes/sync-all-indexes! (mt/db) persist.appdb/reader persist.appdb/writer))
           (is (every? (complement :database_indexed)
                       (t2/select :model/Field
                                  {:where [:and

test/metabase/sync/sync_metadata/tables_test.clj

@@ -10,6 +10,7 @@
    [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
    [metabase.sync.core :as sync]
    [metabase.sync.sync-metadata :as sync-metadata]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.tables :as sync-tables]
    [metabase.test :as mt]
    [metabase.test.data.interface :as tx]
@@ -66,7 +67,7 @@
     (mt/with-temp [:model/Database db {}
                    :model/Table    table-1 {:name "Table 1" :db_id (u/the-id db)}
                    :model/Table    _       {:name "Table 2" :db_id (u/the-id db)}]
-      (#'sync-tables/retire-tables! db #{{:name "Table 1" :schema (:schema table-1)}})
+      (#'sync-tables/retire-tables! db #{{:name "Table 1" :schema (:schema table-1)}} persist.appdb/writer)
       (is (= {"Table 1" false "Table 2" true}
              (t2/select-fn->fn :name :active :model/Table :db_id (u/the-id db)))))))
 
@@ -195,7 +196,7 @@
                    :model/Table table-3 {:name "active_table"
                                          :db_id (u/the-id db)
                                          :active true}]
-      (#'sync-tables/archive-tables! db)
+      (#'sync-tables/archive-tables! db persist.appdb/reader persist.appdb/writer)
 
       (testing "Old deactivated table is archived with suffix"
         (let [archived-table (t2/select-one :model/Table (:id table-1))]
@@ -226,7 +227,7 @@
                                             :active           false
                                             :transform_target false
                                             :deactivated_at   (t/minus (t/offset-date-time) (t/days 30))}]
-      (#'sync-tables/archive-tables! db)
+      (#'sync-tables/archive-tables! db persist.appdb/reader persist.appdb/writer)
 
       (testing "Transform target table is not archived or renamed"
         (let [table (t2/select-one :model/Table (:id provisional))]
@@ -248,7 +249,7 @@
                                        :archived_at (t/minus (t/offset-date-time) (t/days 5))}]
       (let [original-name (:name table)
             original-archived-at (:archived_at table)]
-        (#'sync-tables/archive-tables! db)
+        (#'sync-tables/archive-tables! db persist.appdb/reader persist.appdb/writer)
 
         (let [updated-table (t2/select-one :model/Table (:id table))]
           (is (= original-name (:name updated-table)))
@@ -285,7 +286,7 @@
                                                 :db_id (u/the-id db)
                                                 :active false
                                                 :deactivated_at (t/minus (t/offset-date-time) (t/days 20))}]
-      (#'sync-tables/archive-tables! db)
+      (#'sync-tables/archive-tables! db persist.appdb/reader persist.appdb/writer)
 
       (testing "the original table was archived and renamed"
         (let [archived-table (t2/select-one :model/Table (:id original-table))]
@@ -313,15 +314,18 @@
                                                 :data_authority :unconfigured
                                                 :is_writable    false}]
       ;; Simulate what happens during sync: the driver reports both tables as writable
-      (let [select-cols (into [:model/Table :id :name :schema :data_authority] @#'sync-tables/keys-to-update)]
+      (let [select-cols (into [:model/Table :id :name :schema :data_authority] @#'sync-tables/keys-to-update)
+            writer persist.appdb/writer]
         (#'sync-tables/update-table-metadata-if-needed!
          {:name "computed_table" :schema nil :is_writable true}
          (t2/select-one select-cols (:id computed-table))
-         db)
+         db
+         writer)
         (#'sync-tables/update-table-metadata-if-needed!
          {:name "normal_table" :schema nil :is_writable true}
          (t2/select-one select-cols (:id normal-table))
-         db))
+         db
+         writer))
       (testing "computed table should remain non-writable"
         (is (false? (t2/select-one-fn :is_writable :model/Table (:id computed-table)))))
       (testing "normal table should be updated to writable"
@@ -335,11 +339,11 @@
             normal-table-metadata {:name "normal_table"}]
 
         (testing "creating a table in a sample database"
-          (let [created-table (sync-tables/create-table! sample-db sample-table-metadata)]
+          (let [created-table (sync-tables/create-table! sample-db sample-table-metadata persist.appdb/writer)]
             (is (= :ingested (:data_authority created-table)))))
 
         (testing "creating a table in a normal database"
-          (let [created-table (sync-tables/create-table! normal-db normal-table-metadata)]
+          (let [created-table (sync-tables/create-table! normal-db normal-table-metadata persist.appdb/writer)]
             (is (= :unconfigured (:data_authority created-table)))))
 
         (testing "reactivating a table in a sample database"
@@ -347,7 +351,7 @@
                                                       :name           "existing_sample_table"
                                                       :active         false
                                                       :data_authority :computed}]
-            (sync-tables/create-or-reactivate-table! sample-db {:name "existing_sample_table"})
+            (sync-tables/create-or-reactivate-table! sample-db {:name "existing_sample_table"} persist.appdb/reader persist.appdb/writer)
             (let [updated-table (t2/select-one :model/Table (:id existing-table))]
               (is (= :ingested (:data_authority updated-table)))
               (is (:active updated-table)))))))))

test/metabase/sync/util_test.clj

@@ -319,7 +319,7 @@
     (testing "If a non-recoverable error occurs during sync, `initial-sync-status` on the database is set to `aborted`"
       (let [_  (t2/update! :model/Database (mt/id) {:initial_sync_status "incomplete"})
             db (t2/select-one :model/Database :id (mt/id))]
-        (with-redefs [sync-metadata/make-sync-steps (fn [_]
+        (with-redefs [sync-metadata/make-sync-steps (fn [& _]
                                                       [(sync-util/create-sync-step
                                                         "fake-step"
                                                         (fn [_] (throw (java.net.ConnectException.))))])]
@@ -346,9 +346,9 @@
         (let [syncing-chan   (a/chan)
               completed-chan (a/chan)]
           (let [sync-fields! sync-fields/sync-fields!]
-            (with-redefs [sync-fields/sync-fields! (fn [database]
+            (with-redefs [sync-fields/sync-fields! (fn [database & args]
                                                      (a/>!! syncing-chan ::syncing)
-                                                     (sync-fields! database))]
+                                                     (apply sync-fields! database args))]
               (future
                 (sync/sync-database! (mt/db))
                 (a/>!! completed-chan ::sync-completed))

test/metabase/upload/impl_test.clj

@@ -25,6 +25,7 @@
    [metabase.permissions.models.data-permissions :as data-perms]
    [metabase.permissions.models.permissions-group :as perms-group]
    [metabase.query-processor :as qp]
+   [metabase.sync.persist.appdb :as persist.appdb]
    [metabase.sync.sync-metadata.tables :as sync-tables]
    [metabase.test :as mt]
    [metabase.test.data.impl :as data.impl]
@@ -115,7 +116,8 @@
   (let [table-name  (ddl.i/format-name driver/*driver* table-name)
         schema-name (or (some->> schema-name (ddl.i/format-name driver/*driver*))
                         (sql.tx/session-schema driver/*driver*))
-        table       (sync-tables/create-or-reactivate-table! database {:name table-name :schema schema-name})]
+        table       (sync-tables/create-or-reactivate-table! database {:name table-name :schema schema-name}
+                                                             persist.appdb/reader persist.appdb/writer)]
     (t2/update! :model/Table (:id table) {:is_upload true})
     (binding [upload/*auxiliary-sync-steps* :synchronous]
       (#'upload/scan-and-sync-table! database table))