Refactored DistinctPipe