pathwaycom/pathway · error · ValueError

The function is a coroutine. You can't use SyncExecutor.

Error message

The function is a coroutine. You can't use SyncExecutor.

What it means

ValueError raised in _prepare_executor when the wrapped function is a coroutine function (async def) but the executor is SyncExecutor. Sync executors call the function directly and cannot await coroutines.

Source

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

            else:
                if not isinstance(wrapped_sig_return_type, dt.List):
                    raise ValueError(
                        f"A batch UDF has to return a list but is annotated as returning {sig_return_type}"
                    )
                return wrapped_sig_return_type.wrapped

        return return_type

    def _wrap_function(self) -> Callable:
        func = self.executor._wrap(self.__wrapped__)
        if self.cache_strategy is not None:
            func = with_cache_strategy(func, self.cache_strategy)
        return func

    def _prepare_executor(self, executor: Executor) -> Executor:
        is_coroutine = inspect.iscoroutinefunction(self.__wrapped__)
        if is_coroutine and isinstance(executor, SyncExecutor):
            raise ValueError("The function is a coroutine. You can't use SyncExecutor.")
        if isinstance(executor, AutoExecutor):
            return async_executor() if is_coroutine else udfs.sync_executor()
        return executor

    def __call__(self, *args, **kwargs) -> expr.ColumnExpression:
        return self.executor._apply_expression_type(
            self.func,
            return_type=self._get_return_type(),
            propagate_none=self.propagate_none,
            deterministic=self.deterministic,
            max_batch_size=self.max_batch_size,
            **self.executor.additional_expression_args(),
            args=args,
            kwargs=kwargs,
        )


class UDFFunction(UDF):

View on GitHub (pinned to fa2f74a464)

Solutions

1. Use an async executor: pw.udfs.async_executor(), or omit executor to let AutoExecutor pick
2. Convert the coroutine back to a sync def if you must keep sync_executor
3. In shared UDF factories, select executor based on inspect.iscoroutinefunction(func)

Example fix

// before
@pw.udf(executor=pw.udfs.sync_executor())
async def f(x: int) -> int:
    return await fetch(x)

// after
@pw.udf(executor=pw.udfs.async_executor())
async def f(x: int) -> int:
    return await fetch(x)

import inspect
from pathway.internals.udfs import SyncExecutor

def assert_executor_matches(fn, executor):
    if inspect.iscoroutinefunction(fn):
        assert not isinstance(executor, SyncExecutor), 'async def needs an async executor'
    return executor

import inspect

def is_coroutine_fn(fn) -> bool:
    return inspect.iscoroutinefunction(fn)

When it happens

Trigger: @pw.udf(executor=pw.udfs.sync_executor()) decorating an async def function; passing an explicit SyncExecutor (or subclass) for a coroutine UDF. Note AutoExecutor would auto-select an async executor, so this only fires when sync_executor was requested explicitly.

Common situations: Converting a sync UDF to async def while keeping an explicit sync_executor in shared decorator config; copy-pasted decorator parameters; custom executor subclassing SyncExecutor used with async functions.

Related errors

• Cannot perform {name} when column of type {dtypes_repr} is i
• Batching is not supported for fully asynchronous UDFs.
• Using column of type {dtype.typehint} is not allowed here. C
• A batch UDF has to return a list but is annotated as returni
• cache name `{name}` used more than once

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