 ;;
 
 (defn predicate
-  [f]
+  [pred-fn]
   (reify Predicate
     (test [_ k v]
-      (boolean (f k v)))))
+      (boolean (pred-fn k v)))))
+
+(defn key-value-mapper
+  [map-fn]
+  (reify KeyValueMapper
+    (apply [_ k v]
+      (apply key-value (map-fn k v)))))
 
 ;; kstream/ktable ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;

(defmethod eval-op :map
  [_ map-fn stream]
  `(.. ~(eval stream)
       (map (reify KeyValueMapper
              (apply [_ k# v#]
                (apply key-value (~map-fn k# v#)))))))

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

  [_ group-fn stream & args]
  `(.. ~(eval stream)
       ~(remove nil? (list 'groupBy
                           `(key-value-map-fn ~group-fn)
                           args))))

(defmethod eval-op :group-by-key
  [_ group-fn stream & args]
  `(.. ~(eval stream)
       ~(remove nil? (list 'groupByKey
                            `(key-value-map-fn ~group-fn)
                            args))))

(defmethod eval-op :join
