 ;;   [:c [nil '(22)]])
 
 ```
-
 See https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/join/CoGroupByKey and for a more idiomatic approach to joins [[cogroup-by]] and [[join-by]]"
           base-schema)
    :added "0.1.0"}
-  [specs options pcolls]
-  (let [opts (assoc options :label :cogroup)
-        pcolltuple (make-keyed-pcollection-tuple pcolls)]
-    (apply-transform pcolltuple (cogroup-transform specs opts) base-schema opts)))
+  ([specs options]
+   (let [opts (assoc options :label :cogroup)]
+     (apply-transform (make-group-specs specs) (cogroup-transform opts) named-schema opts)))
+  ([specs] (cogroup specs nil)))
+
+(defn cogroup-by-transform
+  [options reduce-fn]
+  (ptransform
+   :cogroup-by
+   [^GroupSpecs group-specs]
+   (let [grouped-colls (cogroup group-specs options)
+         root-name (or (:name options) "cogroup-by")]
+     (if reduce-fn
+       (dmap reduce-fn (assoc options :name (str root-name "-collector")) grouped-colls)
+       grouped-colls))))
 
 (defn cogroup-by
   {:doc (with-opts-docstr
-          "Takes a specification of the join between pcolls, and returns a PCollection of KVs with values being list of values corresponding to the key-fn. The specification is a list of triple
+          "Takes a specification of the join between pcolls, a function producing the key on which to join and returns a PCollection of KVs with values being list of values corresponding to the key-fn. The specification is a list of triple.
 
  Only one option is supported:
 

        (recur it)))))

(defn cogroup-transform
  ([specs {:keys [join-nil?] :as options}]
   (let [opts (assoc options :label :raw-cogroup)]
     (ptransform
      :cogroup
      [^KeyedPCollectionTuple pcolltuple]
      (let [ordered-tags (->> pcolltuple
                              (.getKeyedCollections)
                              (map #(.getTupleTag %))
                              (sort-by #(.getId %)))
            rel (apply-transform pcolltuple (CoGroupByKey/create) base-schema opts)
            final-rel (pardo
                       (fn [^DoFn$ProcessContext c]
                         (let [^KV kv (.element c)
                               k (.getKey kv)
                               ^CoGbkResult raw-values (.getValue kv)
                               required-list (->> specs
                                                  (map-indexed
                                                   (fn [idx {:keys [type]}]
                                                     (when (= type :required) idx)))
                                                  (remove nil?))
                               required-count (count required-list)]
                           ;; skip if a required part of the group is empty
                           (when-not (some identity
                                           (map (fn [tag {:keys [type]}]
                                                  (if (= type :required)
                                                    (not (.hasNext (.iterator (.getAll raw-values tag))))
                                                    false))
                                                ordered-tags specs))
                             (if (and (not join-nil?) (nil? k))
                               (cond
                                 (= required-count 0) (doseq [[idx tag] (map-indexed (fn [idx tag] [idx tag]) ordered-tags)]
                                                        (greedy-emit-cogbkresult raw-values (count ordered-tags) idx tag c))
                                 (= required-count 1) (greedy-emit-cogbkresult raw-values
                                                                               (count ordered-tags)
                                                                               (first required-list)
                                                                               (nth ordered-tags
                                                                                    (first required-list))
                                                                               c)
                                 :else nil)
                               (let [values (mapv
                                             (fn [tag]

(ds/cogroup [{:type :required} {:type :optional}] {:name :my-cogroup} [pcoll1 pcoll2])

```
See https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/join/CoGroupByKey and for a more idiomatic approach to joins [[cogroup-by]] and [[join-by]]"
          named-schema)
   :added "0.1.0"}
  ([options specs reduce-fn]
   (let [group-specs (make-group-specs specs)
         cogroup-by-tr (cogroup-by-transform options reduce-fn) ]
     (apply-transform group-specs cogroup-by-tr named-schema options)))
  ([options specs] (cogroup-by options specs nil)))

(defn join-by
  ([options specs join-fn]
   (let [nam (or (:name options) "join-by")]
     (pt->>
      (or (:name options) "join-by")
      (cogroup-by (assoc options :name (str nam "-cogroup-by")) specs)
      (dmapcat (fn [^KV kv]
                 (let [results (.getValue kv)
                       results-ok (map #(if (empty? %) [nil] %) results)
                       raw-res (apply combo/cartesian-product results-ok)
                       res (map (fn [prod] (apply join-fn prod)) raw-res)]
                   res))
               {:name (let [root-name (or (:name options) "join-by")]
                        (str root-name "-cartesian-product"))
                :without-coercion-to-clj true }))))
  ([specs join-fn] (join-by {} specs join-fn)))

(defn ddistinct
