    def process_step(self):
        """
        Called for each element.
        """
        self._time_profiler.record(self.uid, Event.queue_in)
        data = self.read_row()
        if data is None or not self.side_by_to_next(data):
            return None
        process_data = [data.get(key) for key in self._node_repr.inputs]

        if any((item is Empty() for item in process_data)):
            return None

        self._time_profiler.record(self.uid, Event.process_in)
        succ, outputs, msg = self._call(process_data)
        assert succ, msg
        if isinstance(outputs, Generator):
            outputs = self._get_from_generator(outputs, len(self._node_repr.outputs))
        self._time_profiler.record(self.uid, Event.process_out)

        size = len(self._node_repr.outputs)
        if size > 1:
            output_map = dict((self._node_repr.outputs[i], outputs[i])
                              for i in range(size))
        elif size == 0:
            # ignore the op result
            # eg: ignore the milvus result
            # .map('vec', (), ops.ann_insert.milvus()),
            output_map = {}
        else:
            # Use one col to store all op result.
            output_map = {}
            output_map[self._node_repr.outputs[0]] = outputs

        self._time_profiler.record(self.uid, Event.queue_out)
        self.data_to_next(output_map)
Called for each element.
towhee/runtime/nodes/_map.py