     (ds/filter even? foo)
     (ds/filter (fn [x] (even? (* x x))) foo)"
            pardo-schema)}
-  dfilter (map-op filter-fn {:label :filter}))
+  dfilter (map-op filter-fn {:label :filter :isomorph? true}))
 
 (defn generate-input
   {:doc (with-opts-docstr

          base-schema text-reader-schema)
   :added "0.1.0"}
  ([from options ^Pipeline p]
   (let [opts (assoc options :label (str "read-edn-file-from-"
                                         (clean-filename from)))]
     (apply-transform p (read-edn-file-transform from opts)
                      (merge base-schema text-reader-schema) opts)))
  ([from p] (read-edn-file from {} p)))


(def write-edn-file-by (partial write-file-by write-edn-file))
(def write-text-file-by (partial write-file-by write-text-file))

;;;;;;;;;;;
;; Joins ;;
;;;;;;;;;;;

(defn make-keyed-pcollection-tuple
  [pcolls]
  (let [empty-kpct (KeyedPCollectionTuple/empty (.getPipeline (first pcolls)))]
    (reduce
     (fn [coll-tuple [idx pcoll]]
       (let [tag (TupleTag. (str idx))
             new-coll-tuple (.and coll-tuple tag pcoll)]
         new-coll-tuple))
     empty-kpct (map-indexed (fn [idx v] [idx v]) pcolls))))

(defn cogroup-transform
  ([options]
