     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]

                 :from [table]
                 :where [:and
                         [:>= id low]
                         [:<= id high]]}

        results (jdbc/query pool (sql/format sql-map))]
    (println "emitting results: " (pr-str results))
    results
    ))

(defrecord SqlWriter [pool table]
  p/Plugin

     {:sql/lower-bound n-min
      :sql/upper-bound n-max}]))

(defn inject-partition-keys
  [table-partitioner event lifecycle]
  {:sql/table-partitioner table-partitioner})
