Implemented union_stream_t::next_batch_impl.