WIP POC0

deps.edn

@@ -113,6 +113,8 @@
                                              :exclusions  [org.slf4j/slf4j-api
                                                            junit/junit]}
   net.thisptr/jackson-jq                    {:mvn/version "1.2.0"}              ; Java implementation of the JQ json query language
+  org.apache.calcite/calcite-core           {:mvn/version "1.33.0"}             ; database federation for cross db joins
+  org.apache.commons/commons-dbcp2          {:mvn/version "2.9.0"}              ; tmp for calcite
   org.apache.commons/commons-compress       {:mvn/version "1.27.1"}             ; compression utils
   org.apache.commons/commons-lang3          {:mvn/version "3.17.0"}             ; helper methods for working with java.lang stuff
   org.apache.logging.log4j/log4j-1.2-api    {:mvn/version "2.24.3"}             ; apache logging framework

modules/drivers/deps.edn

@@ -15,6 +15,7 @@
   metabase/druid-jdbc         {:local/root "druid-jdbc"}
   metabase/hive-like          {:local/root "hive-like"}
   metabase/mongo              {:local/root "mongo"}
+  metabase/net                {:local/root "net"}
   metabase/oracle             {:local/root "oracle"}
   metabase/presto-jdbc        {:local/root "presto-jdbc"}
   metabase/redshift           {:local/root "redshift"}

modules/drivers/net/deps.edn

@@ -0,0 +1,7 @@
+{:paths
+ ["src" "resources"]
+
+ :deps
+ ;; calcite temporarily in core deps
+ ;; for sake of example using another pool
+ {}}

modules/drivers/net/resources/metabase-plugin.yaml

@@ -0,0 +1,20 @@
+info:
+  name: Metabase Database Network Driver
+  version: 0.0.1
+  description: Allows Metabase to interact with multiple databases in single query
+driver:
+  - name: net
+    display-name: Net
+    lazy-load: true
+    connection-properties:
+      - name: db-ids
+        display-name: Database ids
+        type: text
+        required: true
+        placeholder: <id>,<id>,...
+init:
+  # at this point supporting only mysql and postgres, core drivers, ie. no further deps
+  - step: load-namespace
+    namespace: metabase.driver.net
+  # - step: register-jdbc-driver
+  #   class: com.databricks.client.jdbc.Driver

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

@@ -0,0 +1,259 @@
+(ns metabase.driver.net
+  (:require
+   [clojure.java.jdbc :as jdbc]
+   [clojure.string :as str]
+   [metabase.driver :as driver]
+   [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
+   [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
+   [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
+   [metabase.driver.sql-jdbc.sync.interface :as sql-jdbc.sync.interface]
+   [metabase.driver.sql.query-processor :as sql.qp]
+   [metabase.util.net :as u.net]
+   [toucan2.core :as t2])
+  (:import
+   (java.sql Connection)
+   (org.apache.calcite.adapter.jdbc JdbcSchema)
+   (org.apache.calcite.jdbc CalciteConnection)
+   (org.apache.calcite.schema SchemaPlus)
+   (org.apache.commons.dbcp2 BasicDataSource)))
+
+(set! *warn-on-reflection* true)
+
+(driver/register! :net :parent :sql-jdbc)
+
+(doseq [feature driver/features]
+  (defmethod driver/database-supports? [:net feature] [_driver _feature _database] false))
+
+(doseq [feature [:left-join
+                 :right-join
+                 :inner-join]]
+  (defmethod driver/database-supports? [:net feature] [_driver _feature _database] true))
+
+(comment
+  (driver/database-supports? :net :left-join (t2/select-one :model/Database :id 5))
+  (t2/select-one :model/Database :id 5)
+  (t2/debug (t2/update! :model/Database :id 5 {:features [:left-join :right-join :inner-join]})))
+
+;;TODO -- is this needed when can-connect? and do-with-connection-with-options are overridden?
+;; probably not needed atm
+(defmethod sql-jdbc.conn/connection-details->spec :net
+  [_driver {:keys [db-ids] :as _details}]
+  nil)
+
+(defmethod driver/can-connect? :net
+  [_driver details]
+  ;; this assertion is necessary because of every
+  (let [{:keys [db-ids]} (u.net/preprocess-details details)
+        dbs (->> (t2/select :model/Database :id [:in db-ids])
+                 (map u.net/preprocess-slave-db-details))]
+    (and (pos-int? (count db-ids))
+         (every? #(driver/can-connect? (:engine %) (:details %)) dbs))))
+(comment
+  (driver/can-connect? :net (:details (toucan2.core/select-one :model/Database :id [:= 5]))))
+
+;; SYNC
+
+(defmethod sql-jdbc.sync.interface/database-type->base-type :net
+  [_driver database-type]
+  ({:VARCHAR :type/Text}
+   database-type))
+
+(defmethod driver/dbms-version :net
+  [& _args]
+  nil #_"WIP")
+
+(defmethod driver/db-default-timezone :net
+  [& _args]
+  "UTC")
+
+;; TODO
+(defmethod driver/describe-database :net
+  [_driver _database]
+  {:tables #{}})
+
+(defmethod driver/describe-table :net
+  [_driver]
+  {:fields #{}})
+
+(defmethod sql-jdbc.sync/describe-table-fields :net
+  [_driver _conn _table _db-name-or-nil]
+  #{})
+
+;; TODO
+(defmethod driver/db-default-timezone :net
+  [_driver _database]
+  "UTC")
+
+(defmethod driver/table-rows-seq :net
+  [_driver _database _table]
+  ;; hopefully
+  nil)
+
+(comment
+  ;; calcite connection -- OK
+  (let [props (doto (java.util.Properties.)
+                (.setProperty "lex" "JAVA"))]
+    (with-open [conn (java.sql.DriverManager/getConnection "jdbc:calcite:" props)]
+      (.isClosed conn)))
+  ;; decorating the connection -- OK
+  (let [props (doto (java.util.Properties.)
+                (.setProperty "lex" "JAVA"))]
+    (with-open [conn (java.sql.DriverManager/getConnection "jdbc:calcite:" props)]
+      (doto (.isClosed conn) println)
+      (let [calcite-conn (.unwrap conn CalciteConnection)
+            root-schema (.getRootSchema calcite-conn)
+            pg-ds (doto (BasicDataSource.)
+                    (.setUrl "jdbc:postgresql://localhost:5432/postgres_db")
+                    (.setUsername "metabase")
+                    (.setPassword "password")
+                    (.setDriverClassName "org.postgresql.Driver"))]
+        (.add root-schema "POSTGRES" (JdbcSchema/create root-schema
+                                                        "POSTGRES"
+                                                        pg-ds
+                                                        nil
+                                                        "public"))
+        (jdbc/query {:connection calcite-conn} ["SELECT c.customer_id FROM POSTGRES.customers c"]))))
+  ;; query mysql/maria customers mysql_db -- OK
+  (let [props (doto (java.util.Properties.)
+                (.setProperty "lex" "JAVA"))]
+    (with-open [conn (java.sql.DriverManager/getConnection "jdbc:calcite:" props)]
+      (doto (.isClosed conn) println)
+      (let [calcite-conn (.unwrap conn CalciteConnection)
+            root-schema (.getRootSchema calcite-conn)
+            mysql-ds (doto (BasicDataSource.)
+                       (.setUrl "jdbc:mysql://localhost:3306/mysql_db")
+                       (.setUsername "root")
+                       (.setPassword "")
+                       (.setDriverClassName "org.mariadb.jdbc.Driver"))]
+        (.add root-schema "MYSQL" (JdbcSchema/create root-schema
+                                                     "MYSQL"
+                                                     mysql-ds
+                                                     nil
+                                                     "mysql_db"))
+        #_(jdbc/query {:connection calcite-conn} ["SELECT v.name FROM POSTGRES.venues v limit 5"])
+        (jdbc/query {:connection calcite-conn} ["SELECT o.total_amount FROM MYSQL.orders o"]))))
+  ;; query mysql aka maria office-checkins
+  (let [props (doto (java.util.Properties.)
+                (.setProperty "lex" "JAVA"))]
+    (with-open [conn (java.sql.DriverManager/getConnection "jdbc:calcite:" props)]
+      (doto (.isClosed conn) println)
+      (let [calcite-conn (.unwrap conn CalciteConnection)
+            root-schema (.getRootSchema calcite-conn)
+            mysql-ds (doto (BasicDataSource.)
+                       (.setUrl "jdbc:mysql://localhost:3306/office-checkins")
+                       (.setUsername "root")
+                       (.setPassword "")
+                       (.setDriverClassName "org.mariadb.jdbc.Driver"))]
+        (.add root-schema "MYSQL" (JdbcSchema/create root-schema
+                                                     "MYSQL"
+                                                     mysql-ds
+                                                     nil
+                                                     "office-checkins"))
+        #_(jdbc/query {:connection calcite-conn} ["SELECT v.name FROM POSTGRES.venues v limit 5"])
+        #_(jdbc/query {:connection calcite-conn} ["SELECT o.person, CAST(o.`timestamp` as DATE) FROM MYSQL.checkins o"])
+        #_(jdbc/query {:connection calcite-conn} ["SELECT o.person, CAST(o.`timestamp` as DATE) FROM MYSQL.checkins o"])
+
+        (jdbc/query {:connection calcite-conn} ["SELECT * FROM MYSQL.checkins o"]))))
+
+  (let [props (doto (java.util.Properties.)
+                (.setProperty "lex" "JAVA"))]
+    (with-open [conn (java.sql.DriverManager/getConnection "jdbc:calcite:" props)]
+      (doto (.isClosed conn) println)
+      (let [calcite-conn (.unwrap conn CalciteConnection)
+            root-schema (.getRootSchema calcite-conn)
+            pg-ds (doto (BasicDataSource.)
+                    (.setUrl "jdbc:postgresql://localhost:5432/test-data")
+                    (.setUsername "metabase")
+                    (.setPassword "password")
+                    (.setDriverClassName "org.postgresql.Driver"))]
+        (.add root-schema "test-data (postgres)__2" (JdbcSchema/create root-schema
+                                                                       "test-data (postgres)__2"
+                                                                       pg-ds
+                                                                       nil
+                                                                       "public"))
+        (jdbc/query {:connection calcite-conn} ["SELECT * FROM `test-data (postgres)__2`.orders c limit 1"])))))
+
+(defn- link-databases-to-root-schema! [^SchemaPlus root-schema databases]
+  (def ddd databases)
+  (doseq [{:keys [engine details] :as slave-db} databases]
+    (let [schema-name (u.net/schema-for-slave-db slave-db)]
+      (case engine
+        :postgres (.add root-schema schema-name (JdbcSchema/create root-schema
+                                                                   schema-name
+                                                                   (doto (BasicDataSource.)
+                                                                     ;; https://jdbc.postgresql.org/documentation/use/#connecting-to-the-database
+                                                                     (.setUrl (str "jdbc:postgresql:" (:db details)))
+                                                                     (.setUsername "metabase")
+                                                                     (.addConnectionProperty "password" (:password details))
+                                                                     (.addConnectionProperty "host" (:host details))
+                                                                     (.addConnectionProperty "port" (:port details))
+
+                                                                     ;; we are using different connector but I think syntax will be the same
+                                                                     ;; https://dev.mysql.com/doc/connector-j/en/connector-j-reference-jdbc-url-format.html
+                                                                     #_(.addConnectionProperty "db" "test-data")
+                                                                     #_(.setPassword "password")
+                                                                     (.setDriverClassName "org.postgresql.Driver"))
+                                                                   nil
+                                                                   ;; TODO or schema public
+                                                                   "public"))
+        :mysql (.add root-schema schema-name (JdbcSchema/create root-schema
+                                                                schema-name
+                                                                (do (def de details)
+                                                                    (doto (BasicDataSource.)
+                                                                      (.setUrl (str "jdbc:mysql://" (:host details) ":" (:port details) "/" (:db details)))
+                                                                      (.setUsername (:user details))
+                                                                      (.setPassword (:password details ""))
+                                                                      (.setDriverClassName "org.mariadb.jdbc.Driver")))
+                                                                nil
+                                                                (:db details) #_"office-checkins"))))))
+
+;; TODO: I could ignore pooling for now
+(defmethod sql-jdbc.execute/do-with-connection-with-options :net
+  [_driver db-or-id-or-spec _options f]
+  ;; here
+  ;; 1 get connection calcite connection
+  ;; 2 use details to "decorate" it
+  ;; 3 profit
+  (let [details (cond (pos-int? db-or-id-or-spec)
+                      (:details (toucan2.core/select :model/Database :id db-or-id-or-spec))
+
+                      ;; database
+                      (and (map? db-or-id-or-spec) (contains? db-or-id-or-spec :id))
+                      (:details db-or-id-or-spec)
+
+                      :else
+                      (throw (Exception. "Unsupported")))
+        props (doto (java.util.Properties.)
+                (.setProperty "lex" "JAVA"))
+        _ (def dede details)
+        ;; TODO fix mysql
+        linked-dbs-ids #_[2] (mapv #(Long/parseLong %) (clojure.string/split (-> #_dede details :db-ids) #","))]
+    (assert (not-empty linked-dbs-ids))
+    (with-open [conn (java.sql.DriverManager/getConnection "jdbc:calcite:" props)]
+      (doto (.isClosed conn) println)
+      (let [calcite-conn ^CalciteConnection (.unwrap conn CalciteConnection)
+            root-schema (.getRootSchema calcite-conn)]
+        (link-databases-to-root-schema! root-schema (toucan2.core/select :model/Database :id [:in linked-dbs-ids]))
+        (f calcite-conn)))))
+
+;; TODO: read-column-thunk -- leaving this as sql-jdbc default for now
+
+;; TODO: set-parameter -- not interesting for this POC
+
+(defmethod sql-jdbc.execute/read-column-thunk [:net java.sql.Types/TIMESTAMP]
+  [_ rs _ i]
+  (fn []
+    (def a (.isBeforeFirst rs))
+    (def b (.isAfterLast rs))
+    #_(.getObject rs i String)
+    (.getTimestamp rs i)))
+
+(defmethod sql-jdbc.execute/read-column-thunk [:net java.sql.Types/DATE]
+  [_ rs _ i]
+  (fn []
+    (def a (.isBeforeFirst rs))
+    (def b (.isAfterLast rs))
+    #_(.getObject rs i String)
+    (.getDate rs i)))
+
+(defmethod sql.qp/quote-style :net [_] :mysql)

src/metabase/driver/sql_jdbc/execute.clj

@@ -590,7 +590,12 @@
 (defn- get-object-of-class-thunk [^ResultSet rs, ^long i, ^Class klass]
   ^{:name (format "(.getObject rs %d %s)" i (.getCanonicalName klass))}
   (fn []
-    (.getObject rs i klass)))
+    (try
+      (.getObject rs i klass)
+      (catch Exception e
+        (def eee e)
+        (def kkk klass)
+        (throw e)))))
 
 (defmethod read-column-thunk [:sql-jdbc Types/TIMESTAMP]
   [_ rs _ i]
@@ -676,6 +681,7 @@
         (fn [^Long i]
           (let [col-name     (.getColumnLabel rsmeta i)
                 db-type-name (.getColumnTypeName rsmeta i)
+                #_#__ (do (println "ahoj") (println db-type-name))
                 base-type    (sql-jdbc.sync.interface/database-type->base-type driver (keyword db-type-name))]
             (log/tracef "Column %d '%s' is a %s which is mapped to base type %s for driver %s\n"
                         i col-name db-type-name base-type driver)
@@ -742,6 +748,9 @@
     (fn [^Connection conn]
       (with-open [stmt          (statement-or-prepared-statement driver conn sql params qp.pipeline/*canceled-chan*)
                   ^ResultSet rs (try
+                                  #_(log/info "EXEX")
+                                  #_(log/info sql)
+                                  #_(log/info (type conn))
                                   (execute-statement-or-prepared-statement! driver stmt max-rows params sql)
                                   (catch Throwable e
                                     (throw (ex-info (tru "Error executing query: {0}" (ex-message e))

src/metabase/driver/util.clj

@@ -267,7 +267,7 @@
    critical places in the app, and we don't want a single driver to crash the app if it throws an exception, or delay the user
    if it takes a long time to execute."
   [driver feature database]
-  (let [f (if *memoize-supports?* memoized-supports?* supports?*)]
+  (let [f (if false #_*memoize-supports?* memoized-supports?* supports?*)]
     (f driver feature database)))
 
 (defn features

src/metabase/query_processor/pipeline.clj

@@ -76,6 +76,7 @@
                                                      rf-or-e
                                                      reducible-rows)]
                                (catch Throwable e
+                                 (def eee e)
                                  [::error (ex-info (i18n/tru "Error reducing result rows: {0}" (ex-message e))
                                                    {:type qp.error-type/qp}
                                                    e)]))

src/metabase/sync/sync_metadata/fields.clj

@@ -39,6 +39,8 @@
   * In general the methods in these namespaces return the number of rows updated; these numbers are summed and used
     for logging purposes by higher-level sync logic."
   (:require
+   [clojure.data :as data]
+   [medley.core :as m]
    [metabase.driver.util :as driver.u]
    [metabase.models.table :as table]
    [metabase.settings.core :refer [defsetting]]
@@ -51,6 +53,7 @@
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.schema :as ms]
+   [metabase.util.net :as u.net]
    [toucan2.core :as t2]
    [toucan2.util :as t2.util]))
 
@@ -80,46 +83,143 @@
      ;; `sync-instances`
      (sync-metadata/update-metadata! database table db-metadata (fields.our-metadata/our-metadata table))))
 
+(defn- maybe-delete-net-fields!
+  [master-fields slave-ids-delete]
+  (when (seq slave-ids-delete)
+    (let [master-ids-delete (keep (fn [{id       :id
+                                        slave-id :description
+                                        :as _master-field}]
+                                    (when ((set slave-ids-delete) slave-id)
+                                      id))
+                                  master-fields)]
+      (println (str "deleting " slave-ids-delete)
+               (t2/delete! :model/Field :id [:in master-ids-delete])))))
+
+(defn- maybe-update-net-fields!
+  [master-fields slave-ids-update slave-id->slave-field]
+  (when (seq slave-ids-update)
+    (let [master-id->update-data (into {}
+                                       (keep (fn [{id :id slave-id :description :as master-field}]
+                                               (when-some [slave-field (slave-id->slave-field slave-id)]
+                                                 ;; schema no, description has to be slave-id!!!, it is actually
+                                                 ;; description has mapping
+                                                 ;; id is separate...
+                                                 ;; ignoring updated at etc...
+                                                 [id (-> slave-field
+                                                         (dissoc :id :description :table_id))])))
+                                       master-fields)]
+      (str "updating " slave-ids-update)
+      ;; should be reduce actually
+      (doseq [[id data] master-id->update-data]
+        (t2/update! :model/Field :id id data)))))
+
+(defn- maybe-add-slave-fields!
+  [slave-fields slave-ids-add slave-table-id->master-table-id]
+  (when (seq slave-ids-add)
+    (let [add-data (keep (fn [{slave-id :id slave-table-id :table_id :as slave-field}]
+                           (when ((set slave-ids-add) slave-id)
+                             (let [add-data* (-> slave-field
+                                                 (assoc :description (str slave-id))
+                                                 (assoc :table_id (doto (slave-table-id->master-table-id slave-table-id)
+                                                                    (as-> $ (assert (pos-int? $)))))
+                                                 (dissoc :id :entity_id))]
+                               add-data*)))
+                         slave-fields)]
+      (println (str "adding " slave-ids-add))
+      (doseq [data add-data]
+        (t2/insert! :model/Field data)))))
+
+;; when back finish deleteion and stuff comming from diff -- seems easy...
+(defn- sync-fields-for-net
+  "Expects master tables successfully synced. Takes master `database`, maps its tables to slave databases fields. Slave
+  database fields are then synced with master database fields."
+  [database]
+  (let [master-db @(def mmdb (u.net/master-database database))
+        master-fields @(def mmff (u.net/master-fields master-db)) ; -> nil
+        slave-fields @(def sfsf (u.net/slave-fields master-db))
+        _ (comment
+            (map #(select-keys % [:id :name]) sfsf))
+        slave-id->slave-field @(def sid->sf (m/index-by :id slave-fields))
+        slave-id->master-field @(def sid->mf (m/index-by :description master-fields))
+        master-tables @(def mtmt (u.net/master-tables master-db))
+        slave-table-id->master-table-id @(def stid->mtid
+                                           (into {}
+                                                 (map (fn [{master-table-id :id slave-table-id :description :as _master-table}]
+                                                        [slave-table-id master-table-id]))
+                                                 master-tables))
+        [slave-ids-delete
+         slave-ids-add
+         slave-ids-update]
+        @(def ddiiff (data/diff (set (keys slave-id->master-field))
+                                (set (keys slave-id->slave-field))))]
+    (maybe-delete-net-fields! master-fields slave-ids-delete)
+    (maybe-update-net-fields! master-fields slave-ids-update slave-id->slave-field)
+    (maybe-add-slave-fields! slave-fields slave-ids-add slave-table-id->master-table-id)
+    ;; what those numbers should be?
+    {:total-fields 42
+     :fields-updated 21}))
+
+(comment
+  (t2/delete! :model/Table :db_id 9)
+
+  (t2/select :model/Field :table_id [:in (t2/select-fn-vec :id :model/Table :db_id 9)])
+  (t2/count :model/Field :table_id [:in (t2/select-fn-vec :id :model/Table :db_id 9)])
+
+  (sync-fields-for-net (t2/select-one :model/Database :id 9))
+
+  (sync-fields-for-net ddd))
+
+(comment
+  (def ddd (t2/select-one :model/Database :id 5))
+  (u.net/master-database ddd)
+  (u.net/master-tables ddd)
+  (u.net/master-fields ddd)
+  (u.net/slave-tables ddd)
+  (u.net/slave-fields ddd))
+
 (mu/defn sync-fields! :- [:map
                           [: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]
-  (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)
-          fields-metadata (if schemas?
-                            (fetch-metadata/fields-metadata database :schema-names (sync-util/sync-schemas database))
-                            (fetch-metadata/fields-metadata database))]
-      (transduce (comp
-                  (partition-by (juxt :table-name :table-schema))
-                  (map (fn [table-metadata]
-                         (let [fst     (first table-metadata)
-                               table   (t2/select-one :model/Table
-                                                      :db_id (:id database)
-                                                      :%lower.name (t2.util/lower-case-en (:table-name fst))
-                                                      :%lower.schema (some-> fst :table-schema t2.util/lower-case-en)
-                                                      {:where sync-util/sync-tables-clause})
-                               updated (if table
-                                         (try
+  (let [driver (driver.u/database->driver database)]
+    (if (= :net driver)
+      (sync-fields-for-net database)
+      (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)
+              fields-metadata (if schemas?
+                                (fetch-metadata/fields-metadata database :schema-names (sync-util/sync-schemas database))
+                                (fetch-metadata/fields-metadata database))]
+          (transduce (comp
+                      (partition-by (juxt :table-name :table-schema))
+                      (map (fn [table-metadata]
+                             (let [fst     (first table-metadata)
+                                   table   (t2/select-one :model/Table
+                                                          :db_id (:id database)
+                                                          :%lower.name (t2.util/lower-case-en (:table-name fst))
+                                                          :%lower.schema (some-> fst :table-schema t2.util/lower-case-en)
+                                                          {:where sync-util/sync-tables-clause})
+                                   updated (if table
+                                             (try
                                            ;; TODO: decouple nested field columns sync from field sync. This will allow
                                            ;; describe-fields to be used for field sync for databases with nested field columns
                                            ;; Also this should be a driver method, not a sql-jdbc.sync method
-                                           (let [all-metadata (fetch-metadata/include-nested-fields-for-table
-                                                               (set table-metadata)
-                                                               database
-                                                               table)]
-                                             (sync-and-update! database table all-metadata))
-                                           (catch Exception e
-                                             (log/error e)
-                                             0))
-                                         0)]
-                           {:total-fields   (count table-metadata)
-                            :updated-fields updated}))))
-                 (partial merge-with +)
-                 {:total-fields   0
-                  :updated-fields 0}
-                 fields-metadata))))
+                                               (let [all-metadata (fetch-metadata/include-nested-fields-for-table
+                                                                   (set table-metadata)
+                                                                   database
+                                                                   table)]
+                                                 (sync-and-update! database table all-metadata))
+                                               (catch Exception e
+                                                 (log/error e)
+                                                 0))
+                                             0)]
+                               {:total-fields   (count table-metadata)
+                                :updated-fields updated}))))
+                     (partial merge-with +)
+                     {:total-fields   0
+                      :updated-fields 0}
+                     fields-metadata))))))
 
 (mu/defn sync-fields-for-table!
   "Sync the Fields in the Metabase application database for a specific `table`."

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

@@ -207,6 +207,7 @@
                               metabase-field (get name->metabase-field field-name)]]
       (sync-nested-fields-of-one-field! table field-metadata metabase-field))))
 
+;; this should be modifd
 (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,
   creating Field objects or marking them active/inactive as needed."

src/metabase/sync/sync_metadata/tables.clj

@@ -19,8 +19,11 @@
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.schema :as ms]
+   [metabase.util.net :as u.net]
    [toucan2.core :as t2]))
 
+(set! *warn-on-reflection* true)
+
 ;;; ------------------------------------------------ "Crufty" Tables -------------------------------------------------
 
 ;; Crufty tables are ones we know are from frameworks like Rails or Django and thus automatically mark as `:cruft`
@@ -278,6 +281,98 @@
                 :schema schema
                 {:schema new-schema})))
 
+(def ^:private slave-sync-keys
+  [:name #_:description :entity_type :active :display_name :visibility_type #_:schema
+   :points_of_interest :caveats :show_in_getting_started :field_order :initial_sync_status :is_upload
+   :database_require_filter])
+
+(defn- maybe-delete-net-tables!
+  [slave-id->slave-table slave-id->master-table]
+  (let [slave-ids-to-delete (apply disj
+                                   (set (keys slave-id->master-table))
+                                   (keys slave-id->slave-table))
+        master-ids-to-delete (-> slave-id->master-table
+                                 (select-keys slave-ids-to-delete)
+                                 vals
+                                 (->> (map :id)))]
+    (if (seq master-ids-to-delete)
+      (t2/delete! :model/Table :id [:in master-ids-to-delete])
+      0)))
+
+;;TODO: schema
+(defn- maybe-add-new-net-tables!
+  [master-db-id slave-id->slave-table slave-id->master-table]
+  (let [slave-ids-to-add @(def sita (apply disj
+                                           (set (keys slave-id->slave-table))
+                                           (keys slave-id->master-table)))
+        new-master-tables-data @(def nmtd (-> slave-id->slave-table
+                                              (select-keys slave-ids-to-add)
+                                              vals
+                                              (->> (map (fn [{:keys [name id] :as slave-table}]
+                                                          (-> slave-table
+                                                              (select-keys slave-sync-keys)
+                                                              (assoc :db_id master-db-id)
+                                                              (assoc :description (str id))
+                                                              (assoc :schema (-> slave-table :slave_schema))))))))]
+    (if (seq new-master-tables-data)
+      (t2/insert! :model/Table new-master-tables-data)
+      0)))
+
+(comment
+  (apply disj
+         (set (keys sti->st))
+         (keys sti->mt))
+  (maybe-add-new-net-tables! 5 sti->st sti->mt)
+  (toucan2.core/delete! :model/Table :db_id 5))
+
+(defn- maybe-update-net-tables!
+  [master-db-id slave-id->slave-table slave-id->master-table]
+  (let [master-id->slave-id-for-update (into {}
+                                             (map #(select-keys % [:id :description]))
+                                             (select-keys slave-id->master-table
+                                                          (keys slave-id->slave-table)))
+        master-id->update-data (update-vals master-id->slave-id-for-update
+                                            (fn [slave-table]
+                                              (-> (select-keys slave-table slave-sync-keys)
+                                                  (assoc :db_id master-db-id)
+                                                  (assoc :description (-> slave-table :id str))
+                                                  (assoc :schema (-> slave-table :slave_schema)))))]
+    (reduce (fn [acc [master-id update-data]]
+              (+ acc (t2/update! :model/Table :id master-id update-data)))
+            0
+            master-id->update-data)))
+
+(defn- sync-tables-and-database-for-net
+  [database]
+  (let [master-db-id (:id database)
+        master-tables @(def msms (->> (t2/select :model/Table :db_id (:id database))
+                                      (mapv u.net/preprocess-our-table)))
+        slave-dbs @(def sds (u.net/preprocessed-slave-dbs database))
+        slave-db-id->slave-db (m/index-by :id slave-dbs)
+        slave-table-id->slave-table @(def sti->st (t2/select-pk->fn
+                                                   (fn [{:keys [db_id] :as table}]
+                                                     (assoc table :slave_schema
+                                                            (format "%s__%d"
+                                                                    (get-in slave-db-id->slave-db [db_id :name])
+                                                                    db_id)))
+                                                   :model/Table :db_id [:in (map :id slave-dbs)]))
+        ;; at this point description is transformed to slave-table-id
+        slave-table-id->master-table @(def sti->mt (m/index-by :description master-tables))
+        total (+ (maybe-delete-net-tables! slave-table-id->slave-table slave-table-id->master-table)
+                 (maybe-add-new-net-tables! master-db-id slave-table-id->slave-table slave-table-id->master-table)
+                 (maybe-update-net-tables! master-db-id slave-table-id->slave-table slave-table-id->master-table))]
+    {:updated-tables total
+     :total-tables   total}))
+
+(comment
+  ()
+  (driver/with-driver :net
+    (sync-tables-and-database-for-net (toucan2.core/select-one :model/Database :id 5)))
+
+  (sync-tables-and-database-for-net (toucan2.core/select-one :model/Database :id 9))
+
+  ())
+
 (mu/defn sync-tables-and-database!
   "Sync the Tables recorded in the Metabase application database with the ones obtained by calling `database`'s driver's
   implementation of `describe-database`.
@@ -287,47 +382,49 @@
 
   ([database :- i/DatabaseInstance db-metadata]
    ;; 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)
-         multi-level-support?  (driver.u/supports? driver :multi-level-schema database)
-         schemas-to-update     (when multi-level-support?
-                                 (adjusted-schemas driver database our-metadata))
-         our-metadata          (cond->> our-metadata
-                                 multi-level-support?
-                                 (into #{} (map (fn [table]
-                                                  (update table :schema #(get schemas-to-update %1 %1))))))
-         keep-name+schema-set  (fn [metadata]
-                                 (set (map name+schema metadata)))
-         [new-table-metadatas
-          old-table-metadatas] (data/diff
-                                (keep-name+schema-set (set (map name+schema db-table-metadatas)))
-                                (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))
+   (let [driver (driver.u/database->driver database)]
+     (if (= :net driver)
+       (sync-tables-and-database-for-net database)
+       (let [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)
+             multi-level-support?  (driver.u/supports? driver :multi-level-schema database)
+             schemas-to-update     (when multi-level-support?
+                                     (adjusted-schemas driver database our-metadata))
+             our-metadata          (cond->> our-metadata
+                                     multi-level-support?
+                                     (into #{} (map (fn [table]
+                                                      (update table :schema #(get schemas-to-update %1 %1))))))
+             keep-name+schema-set  (fn [metadata]
+                                     (set (map name+schema metadata)))
+             [new-table-metadatas
+              old-table-metadatas] (data/diff
+                                    (keep-name+schema-set (set (map name+schema db-table-metadatas)))
+                                    (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))
 
      ;; 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)))
+         (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)))
      ;; 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))))
+         (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))))
      ;; 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)))
+         (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)))
 
-     (sync-util/with-error-handling (format "Error updating table metadata for %s" (sync-util/name-for-logging database))
+         (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) database))
 
-     {:updated-tables (+ (count new-table-metadatas) (count old-table-metadatas))
-      :total-tables   (count our-metadata)})))
+         {:updated-tables (+ (count new-table-metadatas) (count old-table-metadatas))
+          :total-tables   (count our-metadata)})))))

src/metabase/util/net.clj

@@ -0,0 +1,100 @@
+(ns metabase.util.net
+  (:require
+   [clojure.string :as str]
+   [medley.core :as m]
+   [toucan2.core :as t2]))
+
+(defn preprocess-details
+  [details]
+  (-> details
+      (update :db-ids (fn [db-ids]
+                        (if (string? db-ids)
+                          (->> (str/split db-ids #",")
+                               (mapv #(Long/parseLong ^String %)))
+                          db-ids)))))
+(comment
+  (preprocess-details (:details (toucan2.core/select-one :model/Database :id [:= 5]))))
+
+(defn preprocess-db-details
+  [db]
+  (-> db
+      (update :details preprocess-details)
+      (assoc-in [:details :engine] (:engine db))))
+(comment
+  (preprocess-db-details (toucan2.core/select-one :model/Database :id [:= 5])))
+
+(defn preprocess-slave-db-details
+  [db]
+  (assert (nil? (get-in db [:details :engine])))
+  (assoc-in db [:details :engine] (:engine db)))
+
+(defn preprocessed-slave-dbs
+  [db]
+  (let [slave-ids (-> db preprocess-db-details (get-in [:details :db-ids]))]
+    (mapv preprocess-slave-db-details
+          (t2/select :model/Database :id [:in slave-ids]))))
+(comment
+  (slaves (t2/select-one :model/Database :id 5)))
+
+(defn fetch-tables-for-a-slave
+  [db])
+
+(defn preprocess-our-table
+  [table]
+  (update table :description #(Long/parseLong %)))
+
+(defn master-database
+  [db]
+  (preprocess-db-details db))
+
+(defn preprocess-master-table
+  [table]
+  (m/update-existing table :description #(Long/parseLong %)))
+
+(defn master-tables
+  [db]
+  (let [{:keys [id]} (master-database db)]
+    (t2/select-fn-vec preprocess-master-table :model/Table :db_id id)))
+
+(defn preprocess-master-field
+  [field]
+  (m/update-existing field :description #(Long/parseLong %)))
+
+(defn master-fields
+  [db]
+  (when-some [master-tables* (seq @(def kokocit (master-tables db)))]
+    (t2/select-fn-vec preprocess-master-field :model/Field :table_id [:in (map :id master-tables*)])))
+
+(defn slave-fields
+  [master-db]
+  (let [master-db* @(def mdmd (master-database master-db))
+        slave-db-ids (-> master-db* :details :db-ids)
+        slave-tables @(def stst (t2/select :model/Table :db_id [:in slave-db-ids]))
+        slave-table-ids (map :id slave-tables)
+        slave-fields (t2/select :model/Field :table_id [:in slave-table-ids])]
+    slave-fields))
+
+(defn- slave-databases
+  [master-database]
+  (let [db-ids (-> master-database :details :db-ids)]
+    (assert (and (< 1 (count db-ids)) (every? pos-int? db-ids)))
+    (t2/select :model/Database :id [:in db-ids])))
+
+;; continue here!!! based on this stuff make the fields sync work... requires lot of typing, however not that complicated
+(defn slave-tables
+  [master-database*]
+  (let [master-database* (master-database master-database*)
+        slave-databases* (slave-databases master-database*)
+        slave-db-id->slave-db (m/index-by :id slave-databases*)
+        slave-db-ids (keys slave-db-id->slave-db)
+        tables (t2/select :model/Table :db_id [:in slave-db-ids])]
+    tables))
+
+(comment
+  (slave-tables (t2/select-one :model/Database :id 5))
+  (->> (slave-tables (t2/select-one :model/Database :id 5))
+       (map #(select-keys % [:id :db_id :name :description]))))
+
+(defn schema-for-slave-db
+  [{:keys [id name] :as _slave-db}]
+  (str name "__" id))