     (reify
       Partition$PartitionFn
       (partitionFor [this elem num]
-       (f elem num)))))
+        (f elem num)))))
 
 (defn dpartition-by
   ([f num options ^PCollection pcoll]


(def
  ^{:arglists [['f 'pcoll] ['f 'options 'pcoll]]
    :added "0.1.0"
    :doc
    (with-opts-docstr

                              (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)
                                     values (mapv
                                             (fn [tag]
                                               (loop [^java.util.Iterator it (.iterator (.getAll raw-values tag))
                                                      acc (list)]
                                                 (if (.hasNext it)
                                                   (recur it (conj acc (.next it)))
                                                   acc)))
                                             ordered-tags)]
                                 (.output c (make-kv k values))))
                             (assoc opts
                                    :without-coercion-to-clj true
                                    :coder (make-kv-coder (.getKeyCoder (.getCoder rel))
                                                          (make-nippy-coder)))
                             rel)]
        final-rel)))))

(defn cogroup
