   [_ group-fn stream & args]
   `(.. ~(eval stream)
        ~(remove nil? (list 'groupBy
-                           `(reify KeyValueMapper
-                              (apply [_ k# v#]
-                                (apply key-value (~group-fn k# v#))))
+                           `(key-value-map-fn ~group-fn)
                            args))))
 
 (defmethod eval-op :group-by-key
   [_ group-fn stream & args]
   `(.. ~(eval stream)
        ~(remove nil? (list 'groupByKey
-                            `(reify KeyValueMapper
-                               (apply [_ k# v#]
-                                 (apply key-value (~group-fn k# v#))))
+                            `(key-value-map-fn ~group-fn)
                             args))))
 
 (defmethod eval-op :join

  [_ join-fn map-fn left right]
  `(.. ~(eval left)
       (join ~(eval right)
             `(reify KeyValueMapper
                (apply [_ k# v#]
                  (~map-fn k# v#)))
             `(reify ValueJoiner
                (apply [_ left# right#]
                  (~join-fn left# right#))))))

(defmethod eval-op :flat-map
  [_ map-fn stream-or-table]
  `(.. ~(eval stream-or-table)
       (flatMap (key-value-mapper map-fn))))

(defmethod eval-op :flat-map-values
  [_ map-fn stream-or-table]
