<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:

def _stream(self, doc, source, new_data, rollover=None, setter=None):
        ''' Internal implementation to handle special-casing stream events
        on ``ColumnDataSource`` columns.

        Normally any changes to the ``.data`` dict attribute on a
        ``ColumnDataSource`` triggers a notification, causing all of the data
        to be synchronized between server and clients.

        The ``.stream`` method on column data sources exists to provide a
        more efficient way to perform streaming (i.e. append-only) updates
        to a data source, without having to perform a full synchronization,
        which would needlessly re-send all the data.

        To accomplish this, this function bypasses the wrapped methods on
        ``PropertyValueDict`` and uses the unwrapped versions on the dict
        superclass directly. It then explicitly makes a notification, adding
        a special ``ColumnsStreamedEvent`` hint to the message containing
        only the small streamed data that BokehJS needs in order to
        efficiently synchronize.

        .. warning::
            This function assumes the integrity of ``new_data`` has already
            been verified.

        '''
        old = self._saved_copy()

        # TODO (bev) Currently this reports old differently for array vs list
        # For arrays is reports the actual old value. For lists, the old value
        # is actually the already updated value. This is because the method
        # self._saved_copy() makes a shallow copy.
        for k, v in  new_data.items():
            if isinstance(self[k], np.ndarray) or isinstance(new_data[k], np.ndarray):
                data = np.append(self[k], new_data[k])
                if rollover and len(data) > rollover:
                    data = data[-rollover:]
                super(PropertyValueDict, self).__setitem__(k, data)
            else:
                L = self[k]
                L.extend(new_data[k])
                if rollover is not None:
                    del L[:-rollover]

        from ...document.events import ColumnsStreamedEvent

        self._notify_owners(old,
                            hint=ColumnsStreamedEvent(doc, source, new_data, rollover, setter))