          (>! ch :done)
        (catch Exception e
          (fatal e))))
-    {:sql/pool pool
-     :sql/read-ch ch
-     :sql/pending-messages pending-messages}))
+    (merge event-map {:sql/pool pool
+                       :sql/read-ch ch
+                       :sql/pending-messages pending-messages})))
 
 (defn close-partition-keys
   [{:keys [sql/pool] :as event} lifecycle]

        step (bigint (/ (- n-max n-min) steps-num))
        ranges (partition-all 2 1 (range n-min n-max step))
        columns (or (:sql/columns task-map) [:*])]
    (map (fn [[l h]]
           {:low (util/bigint-to-bytes l)
            :high (util/bigint-to-bytes (dec (or h (inc n-max))))
            :table (:sql/table task-map)
            :id (:sql/id task-map)
            :columns columns})
         ranges)))

(defn partition-table [{:keys [onyx.core/task-map sql/pool] :as event}]
  (let [table (name (:sql/table task-map))

(defn start-commit-loop! [log checkpoint-key content checkpoint-ch checkpoint-loop-ms]
  (go-loop [updated-content content]
           (let [timeout-ch (timeout checkpoint-loop-ms)
                 [acked ch] (alts!! [timeout-ch checkpoint-ch] :priority true)]
             (cond (= ch timeout-ch)
                   (do
                     (extensions/force-write-chunk log :chunk updated-content checkpoint-key)
                     (recur updated-content))
                   (and (= ch checkpoint-ch)
