   (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

            link
            (:id raw-segment)
            (:completion-id raw-segment)
            (fuse-ack-vals (:onyx.core/task-map event) (:ack-val raw-segment) (get children raw-segment)))))))))

(defn join-output-paths [all to-add downstream]
  (cond (= to-add :all) (into #{} downstream)

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

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