Looks like this is re-implementing "plus" on a flow. I prefer to keep this behind the Semigroup, as it makes it easier to support different flow representations down the road.

I think you want to implement Semigroup for pipe arrays. Then you can remove `with-merged-pipes` and add:

```clojure
;; We should probably call this "sum" and keep it in algebra.clj
(defn merge* [& flows] (reduce plus flows))

(defn lift-pipes [flows]
  (map #(add-op % (fn [p] (into-array Pipe [p])))
       flows))

(defn co-group*
  [flows group-fields decl-fields aggs & {:keys [reducers join] :or {join :inner}}]
  (-> flows
      lift-pipes
      merge*
      (add-op (fn [pipes]
                (-> (co-group pipes group-fields decl-fields join)
                    (set-reducers reducers)
                    (add-co-group-aggs aggs))))))
```