refactored to express all pipes as a pipe_with, and added the reduce function as argument
unpiped reduce function to support pipe_accumulate and pipe_accumulate_matching
accumulate_matching without expr