     true)
 
   (write-batch
-    [_ {:keys [onyx.core/results onyx.core/task-map sql/pool]} replica _]
+    [_ {:keys [onyx.core/results]} replica _]
     (doseq [msg (mapcat :leaves (:tree results))]
       (jdbc/with-db-transaction
         [conn pool]
-        (doseq [row (:rows (:message msg))]
-          (jdbc/update! conn (:sql/table task-map) row (sql-dsl/where (:where (:message msg)))))))
+        (doseq [row (:rows msg)]
+          (jdbc/update! conn table row (sql-dsl/where (:where msg))))))
     true))
 
 (defn upsert-rows [pipeline-data]

    (doseq [msg (mapcat :leaves (:tree results))]
      (jdbc/with-db-transaction
        [conn pool]
        (doseq [row (:rows (:message msg))]
          (jdbc/insert! conn table row))))
    true))


    (if-let [seg (first @rst)]
      (do (vswap! rst rest)
          (vswap! offset inc)
          seg)
      (do (vreset! completed? true)
          nil))))
