{"record":{"id":"54666ae73e509a43","repo":"pathwaycom/pathway","slug":"expected-is-upsert-to-be-of-type-bool-got-i","errorCode":null,"errorMessage":"Expected 'is_upsert' to be of type 'bool', got '{is_upsert_type.typehint}'","messagePattern":"Expected 'is_upsert' to be of type 'bool', got '(.+?)'","errorType":"exception","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"python/pathway/internals/table.py","lineNumber":2955,"sourceCode":"        ... id | pet | age | is_upsert | __time__\n        ...  1 | cat |  3  |   True    |     2\n        ...  2 | dog | 11  |   True    |     2\n        ...  1 | cat | 4   |   True    |     4\n        ...  2 | dog | 0   |  False    |     4\n        ... '''\n        ... )\n        >>> t2 = t1.stream_to_table(pw.this.is_upsert)\n        >>> pw.debug.compute_and_print_update_stream(t2, include_id=False)\n        pet | age | is_upsert | __time__ | __diff__\n        cat | 3   | True      | 2        | 1\n        dog | 11  | True      | 2        | 1\n        cat | 3   | True      | 4        | -1\n        dog | 11  | True      | 4        | -1\n        cat | 4   | True      | 4        | 1\n        \"\"\"\n        is_upsert_type = self.eval_type(is_upsert)\n        if is_upsert_type != dt.BOOL:\n            raise TypeError(\n                f\"Expected 'is_upsert' to be of type 'bool', got '{is_upsert_type.typehint}'\"\n            )\n        self._validate_expression(is_upsert)\n        is_upsert_column = self._eval(is_upsert)\n        assert self._universe == is_upsert_column.universe\n        context = clmn.StreamToTableContext(self._id_column, is_upsert_column)\n        return self._table_with_context(context)\n\n    @trace_user_frame\n    @contextualized_operator\n    @check_arg_types\n    def from_streams(self, deletion_stream: Table) -> Table[TSchema]:\n        \"\"\"\n        Converts streams of changes (updates and deletions) into a table.\n\n        This method reconstructs the current state of the table from such streams by applying the updates\n        and deletions in order. It is a stateful operation: the operator keeps track of the latest value for each id.\n        If there are multiple events for a single id in a single batch in the input streams, the order of applying","sourceCodeStart":2937,"sourceCodeEnd":2973,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/internals/table.py#L2937-L2973","documentation":"Raised by Table.stream_to_table() when the is_upsert expression's dtype is not exactly dt.BOOL. stream_to_table collapses an update stream into a table using this boolean flag to decide whether each message is an insertion or a deletion, so a strictly bool-typed expression is mandatory.","triggerScenarios":"t.stream_to_table(pw.this.flag) where flag is int (0/1), str ('true'), Optional[bool], or Any-typed; passing a raw column from a connector that does not declare bool.","commonSituations":"Kafka/message-queue connectors delivering 0/1 or 'true'/'false' flags; schemas declared with int or Any for the upsert column; older Pathway versions being more lenient with type coercion.","solutions":["Cast the column to bool: t.stream_to_table(pw.this.flag.cast(bool))","Fix the schema so the column is declared bool: flag: bool in the connector's Schema class","If the flag is int 0/1, convert explicitly: t.stream_to_table(pw.this.flag == 1)","Verify dtype before the call: t.eval_type(pw.this.flag) must print bool"],"exampleFix":"# before\nt2 = t.stream_to_table(pw.this.flag)  # flag: int -> TypeError\n\n# after\nt2 = t.stream_to_table(pw.this.flag == 1)\n# or\nt2 = t.stream_to_table(pw.this.flag.cast(bool))","handlingStrategy":"type-guard","validationCode":"import pathway as pw\n\ndef upsert_is_bool(t, expr) -> bool:\n    return t.eval_type(expr) == pw.schema_dtypes().get('bool') or str(t.eval_type(expr)) == 'bool'\n# simpler: t.eval_type(expr) must equal dt.BOOL","typeGuard":"def is_bool_expr(t, expr) -> bool:\n    from pathway.internals.dtype import dtypes as dt\n    return t.eval_type(expr) == dt.BOOL","tryCatchPattern":"try:\n    t2 = t.stream_to_table(pw.this.flag)\nexcept TypeError as e:\n    if \"'is_upsert'\" in str(e):\n        t2 = t.stream_to_table(pw.this.flag.cast(bool))","preventionTips":["Declare the upsert column as bool in the connector schema","Cast int 0/1 flags with == 1 or .cast(bool) before stream_to_table","Verify eval_type prints exactly bool"],"tags":["pathway","typing","stream-to-table","bool"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}