Testing Redshift hanging transactions

test.clj

(defn- x []
  (mt/with-driver :redshift
    (count (mt/rows (mt/run-mbql-query :venues)))
    (print \.)
    (flush)))

(defn- datasource ^javax.sql.DataSource []
  (:datasource (sql-jdbc.conn/db->pooled-connection-spec (mt/with-driver :redshift (mt/db)))))

(defn- print-results [^java.sql.ResultSet rs]
  (let [rsmeta              (.getMetaData rs)
        num-cols            (.getColumnCount rsmeta)
        cols-range          (vec (range 1 (inc num-cols)))
        col-names           (vec (for [i cols-range]
                                   (.getColumnLabel rsmeta i)))
        row-count           (atom 0)
        max-col-name-length (reduce max (map count col-names))
        col-spacers         (vec (for [col-name col-names]
                                   (str/join (repeat (- max-col-name-length (count col-name))
                                                     " "))))]
    (while (.next rs)
      (println (u/format-color 'yellow "\n<ROW %d>" (swap! row-count inc)))
      (doseq [i cols-range]
        (println (u/format-color 'blue (nth col-names (dec i)))
                 (nth col-spacers (dec i))
                 (pr-str (.getObject rs (int i))))))
    (println (u/format-color 'yellow "<DONE>"))
    @row-count))

(def ^:private open-transactions-sql
  (str "SELECT t.txn_owner AS owner, t.txn_db AS db, t.lock_mode AS mode, t.granted, t.lockable_object_type AS lock_type, "
       "       t.txn_start, q.starttime AS query_start, q.endtime AS query_end, q.aborted, tbl.table, substring(q.querytxt,1,40) AS query, "
       "       t.pid AS pid "
       "FROM svv_transactions t "
       "LEFT JOIN stl_query q"
       "       ON t.pid = q.pid "
       "LEFT JOIN svv_table_info tbl"
       "       ON t.relation = tbl.table_id "
       "WHERE TRUE "
       "  AND q.starttime > (current_timestamp - interval '2 minutes')"
       ;; "  AND t.pid = pg_backend_pid()"
       "  AND q.querytxt IS NOT NULL "
       "  AND q.querytxt LIKE '-- Metabase%' "
       "  AND t.lock_mode LIKE '%ExclusiveLock' "
       "  AND t.lockable_object_type = 'relation' "
       ;; "  AND t.granted = true "
       ;; "  AND t.txn_owner = 'cam' "
       ;; "  AND t.txn_db = 'testdb' "
       "ORDER BY t.txn_start DESC "
       "LIMIT 50;"
       ))

(defn- jdbc-spec []
  (sql-jdbc.conn/connection-details->spec :redshift (:details (mt/with-driver :redshift (mt/db)))))

(defn- run-query
  ([sql]
   (with-open [conn (.getConnection (datasource))]
     (run-query conn sql)))

  ([^java.sql.Connection conn, ^String sql]
   (doto conn
     (.setAutoCommit true)
     #_(.setReadOnly true)
     (.setTransactionIsolation java.sql.Connection/TRANSACTION_READ_UNCOMMITTED))
   (with-open [stmt (doto (.prepareStatement conn sql
                                             java.sql.ResultSet/TYPE_FORWARD_ONLY
                                             java.sql.ResultSet/CONCUR_READ_ONLY
                                             java.sql.ResultSet/CLOSE_CURSORS_AT_COMMIT)
                      (.setFetchDirection java.sql.ResultSet/FETCH_FORWARD))
               rs   (.executeQuery stmt)]
     (print-results rs))))

(defonce ^:private query-counter (atom 0))

(defn- add-counter [sql]
  (let [query-number (swap! query-counter inc)]
    (printf "Running query #%d\n" query-number)
    (format "-- <Query #%s>\n%s" query-number sql)))

(defn- y []
  (future
    (metabase.sync/sync-database! (mt/with-driver :redshift (mt/db)))
    (println "<FINISHED SYNC>"))
  (let [done (future (dorun (pmap (fn [_] (x)) (range 1000))))]
    (dotimes [_ 10]
      (run-query (add-counter open-transactions-sql))
      (Thread/sleep 2000))
    @done))