pathwaycom/pathway · error · ValueError

Unexpected value type. Expected bytes, got {type(data)}

Error message

Unexpected value type. Expected bytes, got {type(data)}

What it means

Raised at runtime by the Pub/Sub output connector when the single value in a change row is not a Python bytes object. Google Pub/Sub messages are raw byte payloads, so the connector refuses to publish str, int, or any other type instead of silently encoding it with an unspecified serialization.

Source

Thrown at python/pathway/io/pubsub/__init__.py:32

class _OutputBuffer:
    MAX_BUFFER_SIZE = 1024

    def __init__(
        self, publisher: pubsub_v1.PublisherClient, project_id: str, topic_id: str
    ) -> None:
        self._publisher = publisher
        self._topic_path = publisher.topic_path(project_id, topic_id)
        self._publish_futures: list = []

    def on_change(
        self, key: Pointer, row: dict[str, Any], time: int, is_addition: bool
    ) -> None:
        if len(row) != 1:
            raise ValueError(f"Unexpected number of columns: {len(row)}")
        data = next(iter(row.values()))
        if not isinstance(data, bytes):
            raise ValueError(f"Unexpected value type. Expected bytes, got {type(data)}")

        diff = 1 if is_addition else -1
        publish_future = self._publisher.publish(
            self._topic_path, data, pathway_time=str(time), pathway_diff=str(diff)
        )
        self._publish_futures.append(publish_future)

    def on_time_end(self, time: int) -> None:
        if self._publish_futures:
            self._flush_publish_futures()

    def _flush_publish_futures(self) -> None:
        for future in self._publish_futures:
            try:
                future.result()
            except Exception:
                logging.exception("Failed to publish message")
        self._publish_futures = []

View on GitHub (pinned to fa2f74a464)

Solutions

1. Encode the payload column to bytes before writing: convert at the source or with an apply/astype into a bytes column.
2. If the column is dt.ANY, normalize it explicitly (e.g. json.dumps(record).encode()) so the type is guaranteed bytes.
3. Use a serialization step (pw.asynchronous_apply or apply returning bytes) instead of forwarding raw ANY values.

Example fix

# before
t = t.select(data="hello world")
pw.io.pubsub.write(t, publisher, project_id, topic_id)

# after
t = t.select(data=pw.apply(lambda s: s.encode("utf-8"), pw.this.msg))
pw.io.pubsub.write(t, publisher, project_id, topic_id)

t = t.select(data=pw.apply(lambda v: v if isinstance(v, bytes) else str(v).encode("utf-8"), pw.this.payload))
pw.io.pubsub.write(t, publisher, project_id, topic_id)

def is_bytes_value(v) -> bool:
    return isinstance(v, bytes)

When it happens

Trigger: The table's single column has dtype ANY (which passes the static write() check) but at runtime holds str/int/float values, e.g. a table built with a literal string value or values from an untyped input connector.

Common situations: Using dt.ANY columns from python connectors or untyped sources and feeding them straight to pubsub.write; forgetting to encode a JSON string with .encode('utf-8'); prototyping with literal values that are str, not bytes.

Related errors

• The column should be of the type 'bytes'
• Unexpected number of columns: {len(row)}
• Unexpected number of columns in table: {len(table._columns)}
• demo.noisy_linear_stream error: nb_rows should be strictly p
• Invalid schema. Time columns must be int or float.

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/784f74217f4b3e19. Report an issue: GitHub.