Merge pull request #1583 from metabase/fix-joins-on-other-schemas

Fix joins on non-default-schemas 

src/metabase/driver/generic_sql.clj

@@ -38,7 +38,7 @@
     "Return a korma form for truncating a date or timestamp field or value to a given resolution, or extracting a date component.")
 
   (excluded-schemas ^java.util.Set [this]
-                    "*OPTIONAL*. Set of string names of schemas to skip syncing tables from.")
+    "*OPTIONAL*. Set of string names of schemas to skip syncing tables from.")
 
   (set-timezone-sql ^String [this]
     "*OPTIONAL*. This should be a prepared JDBC SQL statement string to be used to set the timezone for the current transaction.

src/metabase/driver/generic_sql/query_processor.clj

@@ -159,11 +159,17 @@
 (defn apply-filter [_ korma-query {clause :filter}]
   (k/where korma-query (filter-clause->predicate clause)))
 
-(defn apply-join-tables [_ korma-query {join-tables :join-tables, {source-table-name :name} :source-table}]
-  (loop [korma-query korma-query, [{:keys [table-name pk-field source-field]} & more] join-tables]
-    (let [korma-query (k/join korma-query table-name
-                              (= (keyword (format "%s.%s" source-table-name (:field-name source-field)))
-                                 (keyword (format "%s.%s" table-name        (:field-name pk-field)))))]
+(defn apply-join-tables [_ korma-query {join-tables :join-tables, {source-table-name :name, source-schema :schema} :source-table}]
+  (loop [korma-query korma-query, [{:keys [table-name pk-field source-field schema]} & more] join-tables]
+    (let [table-name        (if (seq schema)
+                              (str schema \. table-name)
+                              table-name)
+          source-table-name (if (seq source-schema)
+                              (str source-schema \. source-table-name)
+                              source-table-name)
+          korma-query       (k/join korma-query table-name
+                                    (= (keyword (str source-table-name \. (:field-name source-field)))
+                                       (keyword (str table-name        \. (:field-name pk-field)))))]
       (if (seq more)
         (recur korma-query more)
         korma-query))))

src/metabase/driver/query_processor/resolve.clj

@@ -4,7 +4,6 @@
   (:require (clojure [set :as set]
                      [walk :as walk])
             [medley.core :as m]
-            [swiss.arrows :refer [-<>]]
             [metabase.db :refer [sel]]
             (metabase.driver.query-processor [interface :refer :all]
                                              [parse :as parse])
@@ -182,10 +181,11 @@
                                          (zipmap (map :table_id pk-fields) pk-fields))]
 
       ;; Now build the :join-tables clause
-      (vec (for [{table-id :id, table-name :name} join-tables]
+      (vec (for [{table-id :id, table-name :name, schema :schema} join-tables]
              (let [{pk-field-id :id, pk-field-name :name} (join-table-id->pk-field table-id)]
                (map->JoinTable {:table-id     table-id
                                 :table-name   table-name
+                                :schema       schema
                                 :pk-field     (map->JoinTableField {:field-id   pk-field-id
                                                                     :field-name pk-field-name})
                                 :source-field (let [fk-id (pk-field-id->fk-field-id pk-field-id)]
@@ -197,14 +197,13 @@
   [{{source-table-id :source-table} :query, database-id :database, :keys [table-ids fk-field-ids], :as expanded-query-dict}]
   {:pre [(integer? source-table-id)]}
   (let [table-ids       (conj table-ids source-table-id)
-        table-id->table (sel :many :id->fields [Table :schema :name :id] :id [in table-ids])
+        table-id->table (sel :many :id->fields [Table :schema :name :id], :id [in table-ids])
         join-tables     (vals (dissoc table-id->table source-table-id))]
-
-    (-<> expanded-query-dict
-         (assoc-in [:query :source-table] (or (table-id->table source-table-id)
+    (as-> expanded-query-dict <>
+      (assoc-in <> [:query :source-table] (or (table-id->table source-table-id)
                                               (throw (Exception. (format "Query expansion failed: could not find source table %d." source-table-id)))))
-         (assoc-in [:query :join-tables]  (join-tables-fetch-field-info source-table-id join-tables fk-field-ids))
-         (walk/postwalk #(resolve-table % table-id->table) <>))))
+      (assoc-in <> [:query :join-tables]  (join-tables-fetch-field-info source-table-id join-tables fk-field-ids))
+      (walk/postwalk #(resolve-table % table-id->table) <>))))
 
 
 ;;; # ------------------------------------------------------------ PUBLIC INTERFACE ------------------------------------------------------------

test/metabase/test/data/generic_sql.clj

@@ -104,8 +104,13 @@
             pk-field-name (pk-sql-type loader)
             pk-field-name)))
 
-(defn- default-drop-table-if-exists-sql [loader {:keys [databse-name]} {:keys [table-name]}]
-  (format "DROP TABLE IF EXISTS %s;" (qualify+quote-name loader databse-name table-name)))
+(defn- default-drop-table-if-exists-sql [loader {:keys [database-name]} {:keys [table-name]}]
+  (format "DROP TABLE IF EXISTS %s;" (qualify+quote-name loader database-name table-name)))
+
+(defn drop-table-if-exists-cascade-sql
+  "Alternate implementation of `drop-table-if-exists-sql` that adds `CASCADE` to the statement for DBs that support it."
+  [loader {:keys [database-name]} {:keys [table-name]}]
+  (format "DROP TABLE IF EXISTS %s CASCADE;" (qualify+quote-name loader database-name table-name)))
 
 (defn default-add-fk-sql [loader {:keys [database-name]} {:keys [table-name]} {dest-table-name :fk, field-name :field-name}]
   (let [quot            (partial quote-name loader)

test/metabase/test/data/postgres.clj

@@ -28,14 +28,10 @@
          (when (= context :db)
            {:db database-name})))
 
-(defn- drop-table-if-exists-sql [_ _ {:keys [table-name]}]
-  (format "DROP TABLE IF EXISTS \"%s\" CASCADE;" table-name))
-
-
 (extend PostgresDriver
   generic/IGenericSQLDatasetLoader
   (merge generic/DefaultsMixin
-         {:drop-table-if-exists-sql  drop-table-if-exists-sql
+         {:drop-table-if-exists-sql  generic/drop-table-if-exists-cascade-sql
           :pk-sql-type               (constantly "SERIAL")
           :field-base-type->sql-type (fn [_ base-type]
                                        (field-base-type->sql-type base-type))})