pathwaycom/pathway · error · ValueError

Batching is not supported for fully asynchronous UDFs.

Error message

Batching is not supported for fully asynchronous UDFs.

What it means

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.

Source

Thrown at python/pathway/internals/udfs/__init__.py:159

            executor: Defines the executor of the UDF. It determines if the execution is
                synchronous or asynchronous.
                Defaults to ``AutoExecutor()``, meaning that the execution strategy will be
                inferred from the function definition. By default, if the function is a coroutine,
                then it is executed asynchronously. Otherwise it is executed synchronously.
            cache_strategy: Defines the caching mechanism.
                Defaults to None.
            max_batch_size: If set, defines the maximal number of rows that can be passed
                to a UDF at once. Then each argument is a list of values and a UDF has to
                return a list with results with the same length as input lists. The result
                at position `i` has to be the result for input at position `i`.
        """
        self.return_type = return_type
        self.deterministic = deterministic
        self.propagate_none = propagate_none
        self.executor = self._prepare_executor(executor)
        self.cache_strategy = cache_strategy
        if isinstance(self.executor, FullyAsyncExecutor) and max_batch_size is not None:
            raise ValueError("Batching is not supported for fully asynchronous UDFs.")
        self.max_batch_size = max_batch_size
        self.func = self._wrap_function()

    def _get_config(self) -> dict[str, Any]:
        return {
            "return_type": self.return_type,
            "deterministic": self.deterministic,
            "propagate_none": self.propagate_none,
            "executor": self.executor,
            "cache_strategy": self.cache_strategy,
        }

    def _get_return_type(self) -> Any:
        return_type = self.return_type
        if inspect.isclass(self.__wrapped__):
            sig_return_type: Any = self.__wrapped__
        else:
            try:

View on GitHub (pinned to fa2f74a464)

Solutions

1. Remove max_batch_size when using fully_async_executor
2. If batching is required, use async_executor (not fully_async) so the framework controls batching
3. Make batch parameters conditional in shared UDF factories based on executor type

Example fix

// before
@pw.udf(executor=pw.udfs.fully_async_executor(), max_batch_size=100)
def f(x: int) -> int: ...

// after
@pw.udf(executor=pw.udfs.fully_async_executor())
def f(x: int) -> int: ...

from pathway.internals.udfs import FullyAsyncExecutor

def assert_batching_allowed(executor, max_batch_size):
    if isinstance(executor, FullyAsyncExecutor):
        assert max_batch_size is None, 'fully async UDFs cannot use max_batch_size'
    return max_batch_size

from pathway.internals.udfs import FullyAsyncExecutor

def is_fully_async(executor) -> bool:
    return isinstance(executor, FullyAsyncExecutor)

try:
    udf = pw.udf(executor=exec, max_batch_size=100)(fn)
except ValueError as e:
    if 'fully asynchronous' in str(e):
        udf = pw.udf(executor=exec)(fn)  # drop batching
    else:
        raise

When it happens

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

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

Related errors

• Cannot perform {name} when column of type {dtypes_repr} is i
• A batch UDF has to return a list but is annotated as returni
• The function is a coroutine. You can't use SyncExecutor.
• Using column of type {dtype.typehint} is not allowed here. C
• cache name `{name}` used more than once

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