added optimized reducer join streaming