   (let [segments (p-ext/apply-fn event input)]
     (if (sequential? segments) segments (vector segments))))
 
+
 (defn apply-fn-single [{:keys [onyx.core/batch onyx.core/decompressed] :as event}]
+  ; PERF: Tight inner loop where a lot of time is spent 
   (merge
    event
    (reduce

            results (map (partial build-next-segment thawed) segments)
            tagged (acker/prefuse-vals (map :ack-val results))]
        (-> rets
            (update-in [:onyx.core/results] concat results)
            (assoc-in [:onyx.core/children thawed] tagged))))
    {:onyx.core/results []
     :onyx.core/children {}}
    decompressed)))

(defn apply-fn-batch [{:keys [onyx.core/batch onyx.core/decompressed] :as event}]

  (merge
   event
   (when (and children (not (:onyx/side-effects-only? (:onyx.core/task-map event))))
     (doseq [[raw-segment tagged] children]
       (when (:ack-val raw-segment)
         (let [link (operation/peer-link event (:acker-id raw-segment) :acker-peer-site)]
           (extensions/internal-ack-message
