Method to_stream

python/pathway/internals/table.py:2857–2905

Converts a table to a stream of changes. If in a given batch there is: - an insert or an update for a given key, a row with ``True`` in the ``update_column_name`` column is produced - a delete for a given key, a row with ``False`` in the ``update_column_name`` column

(self, upsert_column_name: str = "is_upsert")

    @contextualized_operator
    @check_arg_types
    def to_stream(self, upsert_column_name: str = "is_upsert") -> Table:
        """Converts a table to a stream of changes.

        If in a given batch there is:
        - an insert or an update for a given key, a row with ``True`` in the
        ``update_column_name`` column is produced
        - a delete for a given key, a row with ``False`` in the ``update_column_name`` column is produced.

        The values in all other columns are kept. This is a stateless operation.

        Args:
            upsert_column_name: name of the boolean column that will be added to the table
                and contain information about the type of action.

        Returns:
            Table: An append only table with an additional column informing about the action type.

        Example:

        >>> import pathway as pw
        >>> t1 = pw.debug.table_from_markdown('''
        ... id | age | owner | pet | __time__ | __diff__
        ...  1 | 10  | Alice | dog |     2    |     1
        ...  2 | 9   | Bob   | cat |     2    |     1
        ...  1 | 10  | Alice | dog |     4    |    -1
        ...  1 | 11  | Alice | dog |     4    |     1
        ...  2 | 9   | Bob   | cat |     4    |    -1
        ...  2 | 10  | Bob   | cat |     4    |     1
        ...  1 | 11  | Alice | dog |     6    |    -1
        ...  1 | 12  | Alice | dog |     6    |     1
        ...  2 | 10  | Bob   | cat |     6    |    -1
        ... ''')
        >>> t2 = t1.to_stream()
        >>> pw.debug.compute_and_print_update_stream(t2, include_id=False)
        age | owner | pet | is_upsert | __time__ | __diff__
        9   | Bob   | cat | True      | 2        | 1
        10  | Alice | dog | True      | 2        | 1
        10  | Bob   | cat | True      | 4        | 1
        11  | Alice | dog | True      | 4        | 1
        10  | Bob   | cat | False     | 6        | 1
        12  | Alice | dog | True      | 6        | 1
        """
        context = clmn.TableToStreamContext(self._id_column)
        columns = {
            name: self._wrap_column_in_context(context, column, name)
            for name, column in self._columns.items()
        }
        columns[upsert_column_name] = context.is_upsert_column
        return Table(_columns=columns, _context=context)

    @trace_user_frame
    @desugar