         fields (cfields field-names)
         cogroup-opts (get-in flowdef [:cogroup-opts pipe])]
     (if-not (nil? cogroup-opts)
-      (let [buffer (case (:group-type cogroup-opts)
+      (let [key-separate-from-value (> (count (first (map :args code-defs))) (:num-streams cogroup-opts))
+            buffer (case (:group-type cogroup-opts)
                      :group (GroupBuffer. inits funcs fields (:num-streams cogroup-opts)
                                           (:group-all cogroup-opts)
                                           (:join-nils cogroup-opts)
                                           (:join-requirements cogroup-opts)
                                           udf
-                                          (:key-separate-from-value cogroup-opts))
+                                          key-separate-from-value)
                      :join (JoinBuffer. (first inits) (first funcs) fields (:all-args cogroup-opts)))]
         (update-in flowdef [:pipes pipe] #(Every. % buffer Fields/RESULTS)))
-      (-> flowdef
-          (update-in [:pipes pipe] #(Each. % (PigPenFunction. (first inits) (first funcs) fields) Fields/RESULTS))
-          (add-val [:pipe-fields] pipe field-names)))))
+      (update-in flowdef [:pipes pipe] #(Each. % (PigPenFunction. (first inits) (first funcs) fields) Fields/RESULTS)))))
 
 (defmethod command->flowdef :group
   [{:keys [id keys fields join-types ancestors opts]} flowdef]

                        (Insert. (cfields ["group_all"]) (into-array [1]))
                        (cfields ["group_all" "value"]))]
                (map (:pipes flowdef) ancestors))
        key-separate-from-value (or is-group-all
                                    (every? #(> (count %) 1) (map (:pipe-fields flowdef) ancestors)))
        is-inner (every? #{:required} join-types)
        pipes (map (fn [p k] (if is-inner
                               (Each. p (cfields k) (FilterNull.))

                                       (into-array (map cfields keys))
                                       Fields/NONE
                                       (BufferJoin.)))
        (add-val [:cogroup-opts] id {:group-id          id
                                     :group-type        :group
                                     :join-nils         (true? (:join-nils opts))
                                     :group-all         is-group-all
                                     :num-streams       (count pipes)
                                     :join-requirements (map #(= :required %) join-types)}))))

(defmethod command->flowdef :join
  [{:keys [id keys fields join-types ancestors opts]} flowdef]
