pathwaycom/pathway · error · ValueError

result of async function does not match output schema

Error message

result of async function does not match output schema

What it means

AsyncTransformer wraps a user-defined async invoke() method and pushes each returned row into a table with the schema declared via __init_subclass__(output_schema=...). Before emitting, _check_result_against_schema compares the keys of the dict returned by invoke() with the column names of output_schema; an exact key-set match is required. Any missing column, extra column, or non-dict return (tuple/list results have no matching keys) triggers this ValueError.

Source

Thrown at python/pathway/stdlib/utils/async_transformer.py:288

        instance_data.buffer.clear()

    def _set_failure(self, key: Pointer, task_id: Pointer) -> None:
        # TODO: replace None with api.ERROR
        data = {col: None for col in self._transformer.output_schema.column_names()}
        self._upsert(key, data, task_id, _AsyncStatus.FAILURE)

    def _upsert(
        self, key: Pointer, data: dict, task_id: Pointer, status=_AsyncStatus.SUCCESS
    ) -> None:
        data = {**data, _ASYNC_STATUS_COLUMN: status.value}
        self._add_inner(task_id, data)

    def _remove_by_key(self, key: Pointer, task_id: Pointer) -> None:
        self._remove_inner(task_id, {})

    def _check_result_against_schema(self, result: dict) -> None:
        if result.keys() != self._transformer.output_schema.keys():
            raise ValueError("result of async function does not match output schema")

    def on_stop(self) -> None:
        self._transformer.close()

    def on_subscribe_change(
        self, key: Pointer, row: list[Any], time: int, is_addition: bool
    ) -> None:
        self._put_request((key, row, time, is_addition))

    def on_subscribe_time_end(self, time: int) -> None:
        self._put_request(time)

    def on_subscribe_end(self) -> None:
        self._put_request("*FINISH*")

    def _put_request(self, message) -> None:
        def put_message(message):
            self._maybe_create_queue()

View on GitHub (pinned to fa2f74a464)

Solutions

1. Make invoke() return a dict whose keys are exactly the output_schema column names: {c: value for c, value in zip(YourSchema.column_names(), result_tuple)}
2. Define the dict once from the schema: return YourSchema(...).as_dict() style construction or explicit {..} literal mirroring the schema
3. Audit every return path in invoke() (including exception/fallback branches) so no path returns a different key set

Example fix

# before
class Enrich(pw.AsyncTransformer, output_schema=Schema(answer=str, score=float)):
    async def invoke(self, q: str) -> dict:
        return {'answer': await call(q)}  # missing 'score'

# after
class Enrich(pw.AsyncTransformer, output_schema=Schema(answer=str, score=float)):
    async def invoke(self, q: str) -> dict:
        r = await call(q)
        return {'answer': r.text, 'score': r.score}

expected = set(YourOutputSchema.column_names())
# in tests, against a sample result from invoke():
# assert set(sample_result.keys()) == expected

def result_matches_schema(result: Any, schema: type) -> bool:
    return isinstance(result, dict) and set(result.keys()) == set(schema.column_names())

async def safe_invoke(self, *args):
    result = await self.invoke(*args)
    if not result_matches_schema(result, type(self).output_schema):
        raise ValueError(f'invoke returned {list(result) if isinstance(result, dict) else type(result)}')
    return result

When it happens

Trigger: invoke() returns {'answer': ...} but output_schema declares columns ('answer', 'confidence'); invoke() returns a tuple or a string instead of a dict; a column is renamed in the schema but not in the returned dict; returning None on an error path inside invoke().

Common situations: Iterating on an LLM/HTTP enrichment UDF where the schema gains a column (e.g. adding 'latency') but the dict-building code is not updated; refactoring invoke() to return dataclasses or tuples; early-return paths in invoke() that skip keys.

Related errors

• Input table has a column {column!r} but it is not present on
• Column {arg_name!r} is present on the argument list of the i
• You can't use a column of type {instance_dtype} as instance
• Value {val} is not of type {dtype}.
• wrong schema of debug data

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