pathwaycom/pathway · error · ValueError

Unexpected number of columns: {len(row)}

Error message

Unexpected number of columns: {len(row)}

What it means

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.

Source

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

from pathway.internals.expression import ColumnReference
from pathway.io._subscribe import subscribe


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()

View on GitHub (pinned to fa2f74a464)

Solutions

1. Pass a table with exactly one column to pw.io.pubsub.write; select the payload column before writing: t = t.select(payload=t.data).
2. Serialize multi-field records yourself into bytes first (e.g. json.dumps of a dict into one bytes column).
3. If you hit this at runtime, re-check that no select/with_columns added columns to the table after the writer was attached.

Example fix

# before
pw.io.pubsub.write(t, publisher, project_id, topic_id)  # t has columns: id, data

# after
t = t.select(data=t.data)
pw.io.pubsub.write(t, publisher, project_id, topic_id)

assert len(t.columns) == 1, f"pubsub writer needs exactly one column, got {len(t.columns)}"
pw.io.pubsub.write(t, publisher, project_id, topic_id)

def single_column_table(t) -> bool:
    return len(t._columns) == 1

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Unexpected value type. Expected bytes, got {type(data)}
• Unexpected number of columns in table: {len(table._columns)}
• The column should be of the type 'bytes'
• 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/de1123f0508d6294. Report an issue: GitHub.