                                              ordered-tags)]
                                  (.output c (make-kv k values)))))))
                        (assoc opts
+                              :name (str root-name "-apply-requirements")
                               :without-coercion-to-clj true
                               :coder (make-kv-coder (.getKeyCoder (.getCoder rel))
                                                     (make-nippy-coder)))
                        rel)]
-
         final-rel)))))
 
 (defn cogroup

        (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]

  (let [nam (some-> options (:name) (name))
        clean-opts (dissoc options :name :coder :coll-name)
        configured-transform (with-opts schema clean-opts transform)
        bound (tapply pcoll nam configured-transform)]
    (-> bound
        (cond-> coder (.setCoder coder))
        (cond-> coll-name (.setName coll-name)))))
