Merge pull request #9504 from metabase/fix-oracle

Fix oracle

src/metabase/driver.clj

@@ -80,7 +80,7 @@
   [driver]
   (not (isa? hierarchy driver ::concrete)))
 
-(defn- driver->expected-namespace [driver]
+(s/defn ^:private driver->expected-namespace [driver :- s/Keyword]
   (symbol
    (or (namespace driver)
        (str "metabase.driver." (name driver)))))

src/metabase/driver/sql_jdbc/connection.clj

@@ -28,10 +28,6 @@
 ;;; |                                           Creating Connection Pools                                            |
 ;;; +----------------------------------------------------------------------------------------------------------------+
 
-(defonce ^:private ^{:doc "A map of our currently open connection pools, keyed by Database `:id`."}
-  database-id->connection-pool
-  (atom {}))
-
 (def ^:private data-warehouse-connection-pool-properties
   "c3p0 connection pool properties for connected data warehouse DBs. See
   https://www.mchange.com/projects/c3p0/#configuration_properties for descriptions of properties."
@@ -44,7 +40,7 @@
    ;; prevent overly large pools by condensing them when connections are idle for 15m+
    "maxIdleTimeExcessConnections" (* 15 60)})
 
-(defn- create-connection-pool
+(defn- create-pool!
   "Create a new C3P0 `ComboPooledDataSource` for connecting to the given DATABASE."
   [{:keys [id engine details], :as database}]
   {:pre [(map? database)]}
@@ -54,32 +50,58 @@
     (assoc (connection-pool/connection-pool-spec spec data-warehouse-connection-pool-properties)
       :ssh-tunnel (:tunnel-connection details-with-tunnel))))
 
+(defn- destroy-pool! [database-id pool-spec]
+  (log/debug (u/format-color 'red (tru "Closing old connection pool for database {0} ..." database-id)))
+  (connection-pool/destroy-connection-pool! (:datasource pool-spec))
+  (when-let [ssh-tunnel (:ssh-tunnel pool-spec)]
+    (.disconnect ^com.jcraft.jsch.Session ssh-tunnel)))
+
+(defonce ^:private ^{:doc "A map of our currently open connection pools, keyed by Database `:id`."}
+  database-id->connection-pool
+  (atom {}))
+
+(defn- set-pool!
+  "Atomically update the current connection pool for Database with `database-id`. Use this function instead of modifying
+  `database-id->connection-pool` directly because it properly closes down old pools in a thread-safe way, ensuring no
+  more than one pool is ever open for a single database."
+  [database-id pool-spec-or-nil]
+  (let [[old-id->pool] (swap-vals! database-id->connection-pool assoc database-id pool-spec-or-nil)]
+    ;; if we replaced a different pool with the new pool that is different from the old one, destroy the old pool
+    (when-let [old-pool-spec (get old-id->pool database-id)]
+      (when-not (identical? old-pool-spec pool-spec-or-nil)
+        (destroy-pool! database-id old-pool-spec))))
+  nil)
+
 (defn notify-database-updated
   "Default implementation of `driver/notify-database-updated` for JDBC SQL drivers. We are being informed that a
   DATABASE has been updated, so lets shut down the connection pool (if it exists) under the assumption that the
   connection details have changed."
   [_ database]
-  (when-let [pool (get @database-id->connection-pool (u/get-id database))]
-    (log/debug (u/format-color 'red (tru "Closing connection pool for database {0} ..." (u/get-id database))))
-    ;; remove the cached reference to the pool so we don't try to use it anymore
-    (swap! database-id->connection-pool dissoc (u/get-id database))
-    ;; now actively shut down the pool so that any open connections are closed
-    (connection-pool/destroy-connection-pool! (:datasource pool))
-    (when-let [ssh-tunnel (:ssh-tunnel pool)]
-      (.disconnect ^com.jcraft.jsch.Session ssh-tunnel))))
+  (set-pool! (u/get-id database) nil))
+
+(def ^:private db->pooled-spec-lock (Object.))
 
 (defn db->pooled-connection-spec
   "Return a JDBC connection spec that includes a cp30 `ComboPooledDataSource`.
    Theses connection pools are cached so we don't create multiple ones to the same DB."
   [database-or-id]
-  (if (contains? @database-id->connection-pool (u/get-id database-or-id))
-    ;; we have an existing pool for this database, so use it
-    (get @database-id->connection-pool (u/get-id database-or-id))
-    ;; create a new pool and add it to our cache, then return it
-    (let [db (if (map? database-or-id) database-or-id (db/select-one [Database :id :engine :details]
-                                                        :id database-or-id))]
-      (u/prog1 (create-connection-pool db)
-        (swap! database-id->connection-pool assoc (u/get-id database-or-id) <>)))))
+  (or
+   ;; we have an existing pool for this database, so use it
+   (get @database-id->connection-pool (u/get-id database-or-id))
+   ;; Even tho `set-pool!` will properly shut down old pools if two threads call this method at the same time, we
+   ;; don't want to end up with a bunch of simultaneous threads creating pools only to have them destroyed the very
+   ;; next instant. This will cause their queries to fail. Thus we should do the usual locking here and make sure only
+   ;; one thread will be creating a pool at a given instant.
+   (locking db->pooled-spec-lock
+     (or
+      ;; check if another thread created the pool while we were waiting to acquire the lock
+      (get @database-id->connection-pool (u/get-id database-or-id))
+      ;; create a new pool and add it to our cache, then return it
+      (let [db (if (map? database-or-id)
+                 database-or-id
+                 (db/select-one [Database :id :engine :details] :id database-or-id))]
+        (u/prog1 (create-pool! db)
+          (set-pool! (u/get-id database-or-id) <>)))))))
 
 
 ;;; +----------------------------------------------------------------------------------------------------------------+

src/metabase/driver/sql_jdbc/execute.clj

@@ -168,6 +168,8 @@
   (with-ensured-connection [conn db]
     ;; This is normally done for us by java.jdbc as a result of our `jdbc/query` call
     (with-open [^PreparedStatement stmt (jdbc/prepare-statement conn sql opts)]
+      ;; specifiy that we'd like this statement to close once its dependent result sets are closed
+      (.closeOnCompletion stmt)
       ;; Need to run the query in another thread so that this thread can cancel it if need be
       (try
         (let [query-future (future (jdbc/query conn (into [stmt] params) opts))]

src/metabase/driver/sql_jdbc/sync.clj

@@ -86,10 +86,10 @@
 (defn- get-tables
   "Fetch a JDBC Metadata ResultSet of tables in the DB, optionally limited to ones belonging to a given schema."
   [^DatabaseMetaData metadata, ^String schema-or-nil, ^String db-name-or-nil]
-  (vec
-   (jdbc/metadata-result
-    (.getTables metadata db-name-or-nil schema-or-nil "%" ; tablePattern "%" = match all tables
-                (into-array String ["TABLE", "VIEW", "FOREIGN TABLE", "MATERIALIZED VIEW"])))))
+  ;; tablePattern "%" = match all tables
+  (with-open [rs (.getTables metadata db-name-or-nil schema-or-nil "%"
+                             (into-array String ["TABLE", "VIEW", "FOREIGN TABLE", "MATERIALIZED VIEW"]))]
+    (vec (jdbc/metadata-result rs))))
 
 (defn fast-active-tables
   "Default, fast implementation of `ISQLDriver/active-tables` best suited for DBs with lots of system tables (like
@@ -99,15 +99,16 @@
    This is as much as 15x faster for Databases with lots of system tables than `post-filtered-active-tables` (4
    seconds vs 60)."
   [driver, ^DatabaseMetaData metadata, & [db-name-or-nil]]
-  (let [all-schemas (set (map :table_schem (jdbc/metadata-result (.getSchemas metadata))))
-        schemas     (set/difference all-schemas (excluded-schemas driver))]
-    (set (for [schema schemas
-               table  (get-tables metadata schema db-name-or-nil)]
-           (let [remarks (:remarks table)]
-             {:name        (:table_name table)
-              :schema      schema
-              :description (when-not (str/blank? remarks)
-                             remarks)})))))
+  (with-open [rs (.getSchemas metadata)]
+    (let [all-schemas (set (map :table_schem (jdbc/metadata-result rs)))
+          schemas     (set/difference all-schemas (excluded-schemas driver))]
+      (set (for [schema schemas
+                 table  (get-tables metadata schema db-name-or-nil)]
+             (let [remarks (:remarks table)]
+               {:name        (:table_name table)
+                :schema      schema
+                :description (when-not (str/blank? remarks)
+                               remarks)}))))))
 
 (defn post-filtered-active-tables
   "Alternative implementation of `ISQLDriver/active-tables` best suited for DBs with little or no support for schemas.
@@ -124,7 +125,8 @@
 (defn get-catalogs
   "Returns a set of all of the catalogs found via `metadata`"
   [^DatabaseMetaData metadata]
-  (set (map :table_cat (jdbc/result-set-seq (.getCatalogs metadata)))))
+  (with-open [rs (.getCatalogs metadata)]
+    (set (map :table_cat (jdbc/metadata-result rs)))))
 
 (defn- database-type->base-type-or-warn
   "Given a `database-type` (e.g. `VARCHAR`) return the mapped Metabase type (e.g. `:type/Text`)."
@@ -145,30 +147,30 @@
 (defn describe-table-fields
   "Returns a set of column metadata for `schema` and `table-name` using `metadata`. "
   [^DatabaseMetaData metadata, driver, {^String schema :schema, ^String table-name :name}, & [^String db-name-or-nil]]
-  (set
-   (for [{database-type :type_name
-          column-name   :column_name
-          remarks       :remarks} (jdbc/metadata-result
-                                   (.getColumns metadata db-name-or-nil schema table-name nil))]
-     (merge
-      {:name          column-name
-       :database-type database-type
-       :base-type     (database-type->base-type-or-warn driver database-type)}
-      (when (not (str/blank? remarks))
-        {:field-comment remarks})
-      (when-let [special-type (calculated-special-type driver column-name database-type)]
-        {:special-type special-type})))))
+  (with-open [rs (.getColumns metadata db-name-or-nil schema table-name nil)]
+    (set
+     (for [{database-type :type_name
+            column-name   :column_name
+            remarks       :remarks} (jdbc/metadata-result rs)]
+       (merge
+        {:name          column-name
+         :database-type database-type
+         :base-type     (database-type->base-type-or-warn driver database-type)}
+        (when (not (str/blank? remarks))
+          {:field-comment remarks})
+        (when-let [special-type (calculated-special-type driver column-name database-type)]
+          {:special-type special-type}))))))
 
 (defn add-table-pks
   "Using `metadata` find any primary keys for `table` and assoc `:pk?` to true for those columns."
   [^DatabaseMetaData metadata, table]
-  (let [pks (set (map :column_name (jdbc/metadata-result
-                                    (.getPrimaryKeys metadata nil nil (:name table)))))]
-    (update table :fields (fn [fields]
-                            (set (for [field fields]
-                                   (if-not (contains? pks (:name field))
-                                     field
-                                     (assoc field :pk? true))))))))
+  (with-open [rs (.getPrimaryKeys metadata nil nil (:name table))]
+    (let [pks (set (map :column_name (jdbc/metadata-result rs)))]
+      (update table :fields (fn [fields]
+                              (set (for [field fields]
+                                     (if-not (contains? pks (:name field))
+                                       field
+                                       (assoc field :pk? true)))))))))
 
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
@@ -199,10 +201,10 @@
   "Default implementation of `driver/describe-table-fks` for SQL JDBC drivers. Uses JDBC DatabaseMetaData."
   [driver db-or-id-or-spec table & [^String db-name-or-nil]]
   (jdbc/with-db-metadata [metadata (->spec db-or-id-or-spec)]
-    (set
-     (for [result (jdbc/metadata-result
-                   (.getImportedKeys metadata db-name-or-nil, ^String (:schema table), ^String (:name table)))]
-       {:fk-column-name   (:fkcolumn_name result)
-        :dest-table       {:name   (:pktable_name result)
-                           :schema (:pktable_schem result)}
-        :dest-column-name (:pkcolumn_name result)}))))
+    (with-open [rs (.getImportedKeys metadata db-name-or-nil, ^String (:schema table), ^String (:name table))]
+      (set
+       (for [result (jdbc/metadata-result rs)]
+         {:fk-column-name   (:fkcolumn_name result)
+          :dest-table       {:name   (:pktable_name result)
+                             :schema (:pktable_schem result)}
+          :dest-column-name (:pkcolumn_name result)})))))

test/metabase/driver/sql_jdbc/sync_test.clj

+(ns metabase.driver.sql-jdbc.sync-test
+  (:require [clojure.java.jdbc :as jdbc]
+            [metabase.driver :as driver]
+            [metabase.driver.sql-jdbc
+             [connection :as sql-jdbc.conn]
+             [sync :as sql-jdbc.sync]]
+            [metabase.test.data :as data]
+            [metabase.test.data.datasets :as datasets])
+  (:import java.sql.ResultSet))
+
+(defn- sql-jdbc-drivers-with-default-describe-database-impl
+  "All SQL JDBC drivers that use the default SQL JDBC implementation of `describe-database`. (As far as I know, this is
+  all of them.)"
+  []
+  (set
+   (filter
+    #(identical? (get-method driver/describe-database :sql-jdbc) (get-method driver/describe-database %))
+    (descendants driver/hierarchy :sql-jdbc))))
+
+(defn- describe-database-with-open-resultset-count
+  "Just like `describe-database`, but instead of returning the database description returns the number of ResultSet
+  objects the sync process left open. Make sure you wrap ResultSets with `with-open`! Otherwise some JDBC drivers like
+  Oracle and Redshift will keep open cursors indefinitely."
+  [driver db]
+  (let [orig-result-set-seq jdbc/result-set-seq
+        resultsets          (atom [])]
+    ;; swap out `jdbc/result-set-seq` which is what ultimately gets called on result sets with a function that will
+    ;; stash the ResultSet object in an atom so we can check whether its closed later
+    (with-redefs [jdbc/result-set-seq (fn [^ResultSet rs & more]
+                                        (swap! resultsets conj rs)
+                                        (apply orig-result-set-seq rs more))]
+      ;; taking advantage of the fact that `sql-jdbc.sync/describe-database` can accept JBDC connections instead of
+      ;; databases; by doing this we can keep the connection open and check whether resultsets are still open before
+      ;; they would normally get closed
+      (jdbc/with-db-connection [conn (sql-jdbc.conn/db->pooled-connection-spec db)]
+        (sql-jdbc.sync/describe-database driver conn)
+        (reduce + (for [rs @resultsets]
+                    (if (.isClosed rs) 0 1)))))))
+
+;; make sure that running the sync process doesn't leak cursors because it's not closing the ResultSets
+;; See issues #4389, #6028, and #6467 (Oracle) and #7609 (Redshift)
+(datasets/expect-with-drivers (sql-jdbc-drivers-with-default-describe-database-impl)
+  0
+  (describe-database-with-open-resultset-count driver/*driver* (data/db)))

test/metabase/query_processor_test/query_cancellation_test.clj

 (ns metabase.query-processor-test.query-cancellation-test
+  "TODO - This is sql-jdbc specific, so it should go in a sql-jdbc test namespace."
   (:require [clojure.java.jdbc :as jdbc]
-            [expectations :refer :all]
+            [expectations :refer [expect]]
             [metabase.test.util :as tu]
             [metabase.test.util.log :as tu.log]))
 
 (deftype FakePreparedStatement [called-cancel?]
   java.sql.PreparedStatement
+  (closeOnCompletion [_]) ; no-op
   (cancel [_] (deliver called-cancel? true))
   (close [_] true))
 

test/metabase/test/data.clj

@@ -7,6 +7,7 @@
             [clojure.tools.logging :as log]
             [medley.core :as m]
             [metabase
+             [driver :as driver]
              [query-processor :as qp]
              [sync :as sync]
              [util :as u]]
@@ -283,6 +284,21 @@
     ;; make sure we're returing an up-to-date copy of the DB
     (Database (u/get-id db))))
 
+(defonce ^:private ^{:arglists '([driver]), :doc "We'll have a very bad time if any sort of test runs that calls
+  `data/db` for the first time calls it multiple times in parallel -- for example my Oracle test that runs 30 sync
+  calls at the same time to make sure nothing explodes and cursors aren't leaked. To make sure this doesn't happen
+  we'll keep a map of driver->lock and only allow a given driver to create one Database at a time. Because each DB has
+  its own lock we can still create different DBs for different drivers at the same time."}
+  driver->create-database-lock
+  (let [locks (atom {})]
+    (fn [driver]
+      (let [driver (driver/the-driver driver)]
+        (or
+         (@locks driver)
+         (do
+           (swap! locks update driver #(or % (Object.)))
+           (@locks driver)))))))
+
 (defn get-or-create-database!
   "Create DBMS database associated with DATABASE-DEFINITION, create corresponding Metabase
   `Databases`/`Tables`/`Fields`, and sync the `Database`. DRIVER should be an object that implements
@@ -292,8 +308,12 @@
    (get-or-create-database! (tx/driver) database-definition))
   ([driver database-definition]
    (let [driver (tx/the-driver-with-test-extensions driver)]
-     (or (tx/metabase-instance database-definition driver)
-         (create-database! database-definition driver)))))
+     (or
+      (tx/metabase-instance database-definition driver)
+      (locking (driver->create-database-lock driver)
+        (or
+         (tx/metabase-instance database-definition driver)
+         (create-database! database-definition driver)))))))
 
 
 (defn do-with-temp-db