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

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

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

;;

(defn predicate
  [pred-fn]
  (reify Predicate
    (test [_ 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 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
