port a table from one database to another

gistfile1.clj

;; out of memory 
(defn populate-table [tablename]
  (debug (str "Populating table: " tablename))
  (sql/with-connection xe2pg.config/*xe-connect*
    (sql/with-query-results rs [(str "select * from " tablename)]
      (sql/with-connection xe2pg.config/*pg-connect*
        (apply sql/insert-records tablename rs)))))

;; out of memory
(defn populate-table-seq [tablename]
  (debug (str "Populating table: " tablename))
  (sql/with-connection xe2pg.config/*xe-connect*
    (sql/with-query-results rs [(str "select * from " tablename)]
      (sql/with-connection xe2pg.config/*pg-connect*
        (doseq [record rs]
          (sql/insert-records tablename record))))))


(defn populate-table-seq-chunk [tablename]
  (debug (str "Populating table: " tablename))
  (sql/with-connection xe2pg.config/*xe-connect*
    (sql/with-query-results rs [(str "select * from " tablename)]
      (sql/with-connection xe2pg.config/*pg-connect*
        (doseq [chunk (partition-all 500 rs)]
          (debug (str tablename " - beginning next chunk"))
          (sql/transaction
           (apply sql/insert-records tablename chunk)))))))

;; this one works
(defn populate-table-internal-resultset [tablename]
  (debug (str "Populating table: " tablename))
  (sql/with-connection xe2pg.config/*xe-connect*
    (with-open [stmt (.prepareStatement (sql/connection) (str "select * from " tablename))]
      (with-open [rset (.executeQuery stmt)]
        (doseq [recordset (partition-all 500 (resultset-seq rset))]
          (debug (str tablename " - beginning next chunk"))
          (sql/with-connection xe2pg.config/*pg-connect*
            (sql/transaction
             (apply sql/insert-records tablename recordset))))))))


;; ------ parallel over a single table

(defn par-write-records [tablename records]
  (debug (str tablename " - beginning chunk"))
  (sql/with-connection xe2pg.config/*pg-connect*
    (sql/transaction
     (apply sql/insert-records tablename records))))

(defn populate-table-par-writes [tablename]
  (debug (str "Populatign table:" tablename))
  (sql/with-connection xe2pg.config/*xe-connect*
    (with-open [stmt (.prepareStatement (sql/connection) (str "select * from " tablename))]
      (with-open [rset (.executeQuery stmt)]
        (pmap (partial par-write-records tablename) (doall (partition-all 500 (resultset-seq rset))))))))