{"record":{"id":"784f74217f4b3e19","repo":"pathwaycom/pathway","slug":"unexpected-value-type-expected-bytes-got-type-d","errorCode":null,"errorMessage":"Unexpected value type. Expected bytes, got {type(data)}","messagePattern":"Unexpected value type\\. Expected bytes, got (.+?)","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/pubsub/__init__.py","lineNumber":32,"sourceCode":"\nclass _OutputBuffer:\n    MAX_BUFFER_SIZE = 1024\n\n    def __init__(\n        self, publisher: pubsub_v1.PublisherClient, project_id: str, topic_id: str\n    ) -> None:\n        self._publisher = publisher\n        self._topic_path = publisher.topic_path(project_id, topic_id)\n        self._publish_futures: list = []\n\n    def on_change(\n        self, key: Pointer, row: dict[str, Any], time: int, is_addition: bool\n    ) -> None:\n        if len(row) != 1:\n            raise ValueError(f\"Unexpected number of columns: {len(row)}\")\n        data = next(iter(row.values()))\n        if not isinstance(data, bytes):\n            raise ValueError(f\"Unexpected value type. Expected bytes, got {type(data)}\")\n\n        diff = 1 if is_addition else -1\n        publish_future = self._publisher.publish(\n            self._topic_path, data, pathway_time=str(time), pathway_diff=str(diff)\n        )\n        self._publish_futures.append(publish_future)\n\n    def on_time_end(self, time: int) -> None:\n        if self._publish_futures:\n            self._flush_publish_futures()\n\n    def _flush_publish_futures(self) -> None:\n        for future in self._publish_futures:\n            try:\n                future.result()\n            except Exception:\n                logging.exception(\"Failed to publish message\")\n        self._publish_futures = []","sourceCodeStart":14,"sourceCodeEnd":50,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/pubsub/__init__.py#L14-L50","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Encode the payload column to bytes before writing: convert at the source or with an apply/astype into a bytes column.","If the column is dt.ANY, normalize it explicitly (e.g. json.dumps(record).encode()) so the type is guaranteed bytes.","Use a serialization step (pw.asynchronous_apply or apply returning bytes) instead of forwarding raw ANY values."],"exampleFix":"# before\nt = t.select(data=\"hello world\")\npw.io.pubsub.write(t, publisher, project_id, topic_id)\n\n# after\nt = t.select(data=pw.apply(lambda s: s.encode(\"utf-8\"), pw.this.msg))\npw.io.pubsub.write(t, publisher, project_id, topic_id)","handlingStrategy":"type-guard","validationCode":"t = t.select(data=pw.apply(lambda v: v if isinstance(v, bytes) else str(v).encode(\"utf-8\"), pw.this.payload))\npw.io.pubsub.write(t, publisher, project_id, topic_id)","typeGuard":"def is_bytes_value(v) -> bool:\n    return isinstance(v, bytes)","tryCatchPattern":null,"preventionTips":["Encode strings at the boundary: .encode('utf-8') or astype(bytes).","Avoid forwarding dt.ANY columns to byte-oriented sinks without normalization.","Serialize structured records with json.dumps(...).encode() into a bytes column."],"tags":["pathway","gcp","pubsub","type-mismatch","bytes","runtime"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}