{"record":{"id":"caa842295fd33fe7","repo":"pathwaycom/pathway","slug":"batching-is-not-supported-for-fully-asynchronous-u","errorCode":null,"errorMessage":"Batching is not supported for fully asynchronous UDFs.","messagePattern":"Batching is not supported for fully asynchronous UDFs\\.","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/internals/udfs/__init__.py","lineNumber":159,"sourceCode":"            executor: Defines the executor of the UDF. It determines if the execution is\n                synchronous or asynchronous.\n                Defaults to ``AutoExecutor()``, meaning that the execution strategy will be\n                inferred from the function definition. By default, if the function is a coroutine,\n                then it is executed asynchronously. Otherwise it is executed synchronously.\n            cache_strategy: Defines the caching mechanism.\n                Defaults to None.\n            max_batch_size: If set, defines the maximal number of rows that can be passed\n                to a UDF at once. Then each argument is a list of values and a UDF has to\n                return a list with results with the same length as input lists. The result\n                at position `i` has to be the result for input at position `i`.\n        \"\"\"\n        self.return_type = return_type\n        self.deterministic = deterministic\n        self.propagate_none = propagate_none\n        self.executor = self._prepare_executor(executor)\n        self.cache_strategy = cache_strategy\n        if isinstance(self.executor, FullyAsyncExecutor) and max_batch_size is not None:\n            raise ValueError(\"Batching is not supported for fully asynchronous UDFs.\")\n        self.max_batch_size = max_batch_size\n        self.func = self._wrap_function()\n\n    def _get_config(self) -> dict[str, Any]:\n        return {\n            \"return_type\": self.return_type,\n            \"deterministic\": self.deterministic,\n            \"propagate_none\": self.propagate_none,\n            \"executor\": self.executor,\n            \"cache_strategy\": self.cache_strategy,\n        }\n\n    def _get_return_type(self) -> Any:\n        return_type = self.return_type\n        if inspect.isclass(self.__wrapped__):\n            sig_return_type: Any = self.__wrapped__\n        else:\n            try:","sourceCodeStart":141,"sourceCodeEnd":177,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/internals/udfs/__init__.py#L141-L177","documentation":"ValueError raised by the UDF base class constructor when a FullyAsyncExecutor is combined with max_batch_size. Fully asynchronous UDFs fetch their own results (results arrive via a separate mechanism), so the framework cannot batch rows for them.","triggerScenarios":"Creating pathway.udfs(...) or a UDF decorator with executor=pathway.udfs.fully_async_executor() (or a custom FullyAsyncExecutor subclass) and also passing max_batch_size=N. The check runs in __init__ immediately at decoration/construction time.","commonSituations":"Copying a batched sync UDF config to an async one; enabling batch parameters globally on shared UDF factory code that also builds fully-async UDFs; upgrading a UDF to fully-async without removing batching flags.","solutions":["Remove max_batch_size when using fully_async_executor","If batching is required, use async_executor (not fully_async) so the framework controls batching","Make batch parameters conditional in shared UDF factories based on executor type"],"exampleFix":"// before\n@pw.udf(executor=pw.udfs.fully_async_executor(), max_batch_size=100)\ndef f(x: int) -> int: ...\n\n// after\n@pw.udf(executor=pw.udfs.fully_async_executor())\ndef f(x: int) -> int: ...","handlingStrategy":"validation","validationCode":"from pathway.internals.udfs import FullyAsyncExecutor\n\ndef assert_batching_allowed(executor, max_batch_size):\n    if isinstance(executor, FullyAsyncExecutor):\n        assert max_batch_size is None, 'fully async UDFs cannot use max_batch_size'\n    return max_batch_size","typeGuard":"from pathway.internals.udfs import FullyAsyncExecutor\n\ndef is_fully_async(executor) -> bool:\n    return isinstance(executor, FullyAsyncExecutor)","tryCatchPattern":"try:\n    udf = pw.udf(executor=exec, max_batch_size=100)(fn)\nexcept ValueError as e:\n    if 'fully asynchronous' in str(e):\n        udf = pw.udf(executor=exec)(fn)  # drop batching\n    else:\n        raise","preventionTips":["Never pair fully_async_executor with max_batch_size","Use async_executor when batching is required","Centralize executor/batch configuration in one validated place"],"tags":["pathway","udf","async","batching"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}