{"record":{"id":"4518a4c4116672e8","repo":"pathwaycom/pathway","slug":"result-of-async-function-does-not-match-output-sch","errorCode":null,"errorMessage":"result of async function does not match output schema","messagePattern":"result of async function does not match output schema","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/stdlib/utils/async_transformer.py","lineNumber":288,"sourceCode":"        instance_data.buffer.clear()\n\n    def _set_failure(self, key: Pointer, task_id: Pointer) -> None:\n        # TODO: replace None with api.ERROR\n        data = {col: None for col in self._transformer.output_schema.column_names()}\n        self._upsert(key, data, task_id, _AsyncStatus.FAILURE)\n\n    def _upsert(\n        self, key: Pointer, data: dict, task_id: Pointer, status=_AsyncStatus.SUCCESS\n    ) -> None:\n        data = {**data, _ASYNC_STATUS_COLUMN: status.value}\n        self._add_inner(task_id, data)\n\n    def _remove_by_key(self, key: Pointer, task_id: Pointer) -> None:\n        self._remove_inner(task_id, {})\n\n    def _check_result_against_schema(self, result: dict) -> None:\n        if result.keys() != self._transformer.output_schema.keys():\n            raise ValueError(\"result of async function does not match output schema\")\n\n    def on_stop(self) -> None:\n        self._transformer.close()\n\n    def on_subscribe_change(\n        self, key: Pointer, row: list[Any], time: int, is_addition: bool\n    ) -> None:\n        self._put_request((key, row, time, is_addition))\n\n    def on_subscribe_time_end(self, time: int) -> None:\n        self._put_request(time)\n\n    def on_subscribe_end(self) -> None:\n        self._put_request(\"*FINISH*\")\n\n    def _put_request(self, message) -> None:\n        def put_message(message):\n            self._maybe_create_queue()","sourceCodeStart":270,"sourceCodeEnd":306,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/stdlib/utils/async_transformer.py#L270-L306","documentation":"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.","triggerScenarios":"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().","commonSituations":"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.","solutions":["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)}","Define the dict once from the schema: return YourSchema(...).as_dict() style construction or explicit {..} literal mirroring the schema","Audit every return path in invoke() (including exception/fallback branches) so no path returns a different key set"],"exampleFix":"# before\nclass Enrich(pw.AsyncTransformer, output_schema=Schema(answer=str, score=float)):\n    async def invoke(self, q: str) -> dict:\n        return {'answer': await call(q)}  # missing 'score'\n\n# after\nclass Enrich(pw.AsyncTransformer, output_schema=Schema(answer=str, score=float)):\n    async def invoke(self, q: str) -> dict:\n        r = await call(q)\n        return {'answer': r.text, 'score': r.score}","handlingStrategy":"validation","validationCode":"expected = set(YourOutputSchema.column_names())\n# in tests, against a sample result from invoke():\n# assert set(sample_result.keys()) == expected","typeGuard":"def result_matches_schema(result: Any, schema: type) -> bool:\n    return isinstance(result, dict) and set(result.keys()) == set(schema.column_names())","tryCatchPattern":"async def safe_invoke(self, *args):\n    result = await self.invoke(*args)\n    if not result_matches_schema(result, type(self).output_schema):\n        raise ValueError(f'invoke returned {list(result) if isinstance(result, dict) else type(result)}')\n    return result","preventionTips":["Construct the return dict with a literal that mirrors output_schema, and add a unit test comparing key sets","Keep exactly one dict-building helper per transformer so schema edits propagate","Cover fallback/exception return paths in invoke() with the same test"],"tags":["pathway","async","async-transformer","schema-mismatch"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}