{"record":{"id":"de1123f0508d6294","repo":"pathwaycom/pathway","slug":"unexpected-number-of-columns-len-row","errorCode":null,"errorMessage":"Unexpected number of columns: {len(row)}","messagePattern":"Unexpected number of columns: (.+?)","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/pubsub/__init__.py","lineNumber":29,"sourceCode":"from pathway.internals.expression import ColumnReference\nfrom pathway.io._subscribe import subscribe\n\n\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()","sourceCodeStart":11,"sourceCodeEnd":47,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/pubsub/__init__.py#L11-L47","documentation":"Raised at runtime by the Pub/Sub output connector's on_change callback when the change row it receives does not contain exactly one column. The connector serializes a single bytes value per message, so a zero-column or multi-column row is a configuration bug, not a data condition. This mirrors the static check done in pw.io.pubsub.write but fires inside the subscription callback.","triggerScenarios":"A table with more than one column (or a table whose shape changed after write() validated it) reaching the _OutputBuffer.on_change callback; effectively only reachable if the single-column validation in write() was bypassed or the table was mutated between setup and run.","commonSituations":"Almost always a secondary symptom: the real error is passing a multi-column table to pw.io.pubsub.write, which normally fails earlier with 'Unexpected number of columns in table'. Hitting this one means the table was built dynamically and changed shape late.","solutions":["Pass a table with exactly one column to pw.io.pubsub.write; select the payload column before writing: t = t.select(payload=t.data).","Serialize multi-field records yourself into bytes first (e.g. json.dumps of a dict into one bytes column).","If you hit this at runtime, re-check that no select/with_columns added columns to the table after the writer was attached."],"exampleFix":"# before\npw.io.pubsub.write(t, publisher, project_id, topic_id)  # t has columns: id, data\n\n# after\nt = t.select(data=t.data)\npw.io.pubsub.write(t, publisher, project_id, topic_id)","handlingStrategy":"validation","validationCode":"assert len(t.columns) == 1, f\"pubsub writer needs exactly one column, got {len(t.columns)}\"\npw.io.pubsub.write(t, publisher, project_id, topic_id)","typeGuard":"def single_column_table(t) -> bool:\n    return len(t._columns) == 1","tryCatchPattern":null,"preventionTips":["Always project with t.select(payload=...) immediately before attaching the writer.","Treat any runtime 'Unexpected number of columns' as late schema drift — freeze the table shape before write().","Assert the column count in tests before pw.run()."],"tags":["pathway","gcp","pubsub","runtime","schema"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}