Method stream_to_table

python/pathway/internals/table.py:2911–2962

Converts a stream of changes (updates and deletions) into a table. In the Pathway Live Data Framework, a stream is a sequence of row changes, where each row has an id and a boolean column (e.g., "is_upsert") indicating whether the row is an update (``True``) or a de

(self, is_upsert: expr.ColumnExpression)

    @check_arg_types
    @contextualized_operator
    def stream_to_table(self, is_upsert: expr.ColumnExpression) -> Table[TSchema]:
        """
        Converts a stream of changes (updates and deletions) into a table.

        In the Pathway Live Data Framework, a stream is a sequence of row changes, where each row
        has an id and a boolean column
        (e.g., "is_upsert") indicating whether the row is an update (``True``) or a deletion (``False``).

        This method reconstructs the current state of the table from such a stream by applying the updates
        and deletions in order. It is a stateful operation: the operator keeps track of the latest value for each id.
        If there are multiple events for a single id in a single batch in a stream, the order of applying
        the actions is not specified.
        For deletions, only ids are important. The values in columns are ignored.

        Args:
            is_upsert: An expression that evaluates to a boolean value. ``True`` means the row
                is an upsert (insert or update), ``False`` means the row is a deletion.

        Returns:
            Table: A table with the same columns as the original stream, representing the current state.

        Example:

        >>> import pathway as pw
        >>> t1 = pw.debug.table_from_markdown(
        ...     '''
        ... id | pet | age | is_upsert | __time__
        ...  1 | cat |  3  |   True    |     2
        ...  2 | dog | 11  |   True    |     2
        ...  1 | cat | 4   |   True    |     4
        ...  2 | dog | 0   |  False    |     4
        ... '''
        ... )
        >>> t2 = t1.stream_to_table(pw.this.is_upsert)
        >>> pw.debug.compute_and_print_update_stream(t2, include_id=False)
        pet | age | is_upsert | __time__ | __diff__
        cat | 3   | True      | 2        | 1
        dog | 11  | True      | 2        | 1
        cat | 3   | True      | 4        | -1
        dog | 11  | True      | 4        | -1
        cat | 4   | True      | 4        | 1
        """
        is_upsert_type = self.eval_type(is_upsert)
        if is_upsert_type != dt.BOOL:
            raise TypeError(
                f"Expected 'is_upsert' to be of type 'bool', got '{is_upsert_type.typehint}'"
            )
        self._validate_expression(is_upsert)
        is_upsert_column = self._eval(is_upsert)
        assert self._universe == is_upsert_column.universe
        context = clmn.StreamToTableContext(self._id_column, is_upsert_column)
        return self._table_with_context(context)

    @trace_user_frame
    @contextualized_operator