async def execute_or_delegate_operator(
    operator_uri, request_params, exhaust=False
):
    """Executes the operator with the given name.

    Args:
        operator_uri: the URI of the operator
        request_params: a dictionary of parameters for the operator
        exhaust (False): whether to immediately exhaust generator operators

    Returns:
        an :class:`ExecutionResult`
    """
    prepared = await prepare_operator_executor(operator_uri, request_params)
    if isinstance(prepared, ExecutionResult):
        raise prepared.to_exception()
    else:
        operator, executor, ctx, inputs = prepared

    execution_options = operator.resolve_execution_options(ctx)
    if (
        not execution_options.allow_immediate_execution
        and not execution_options.allow_delegated_execution
    ):
        raise RuntimeError(
            "This operation does not support immediate OR delegated execution"
        )

    should_delegate = (
        operator.resolve_delegation(ctx) or ctx.requesting_delegated_execution
    )
    if should_delegate:
        if not execution_options.allow_delegated_execution:
            logger.warning(
                "This operation does not support delegated "
                "execution; it will be executed immediately"
            )
            should_delegate = False
    else:
        if not execution_options.allow_immediate_execution:
            logger.warning(
                "This operation does not support immediate "
                "execution; it will be delegated"
            )
            should_delegate = True

    if should_delegate:
        try:
            from .delegated import DelegatedOperationService

            ctx.request_params["delegated"] = True
            metadata = {"inputs_schema": None, "outputs_schema": None}

            if inputs is not None:
                try:
                    metadata["inputs_schema"] = inputs.to_json()
                except Exception as e:
                    logger.warning(
                        f"Failed to resolve inputs schema for the operation: {str(e)}"
                    )

            op = DelegatedOperationService().queue_operation(
                operator=operator.uri,
                context=ctx.serialize(),
                delegation_target=ctx.delegation_target,
                label=operator.name,
                metadata=metadata,
            )

            execution = ExecutionResult(
                op.__dict__, executor, None, delegated=True
            )
            execution.result["context"] = (
                execution.result["context"].serialize()
                if execution.result["context"]
                else None
            )
            return execution
        except Exception as error:
            return ExecutionResult(
                executor=executor,
                error=traceback.format_exc(),
                error_message=str(error),
            )
    else:
        try:
            result = await do_execute_operator(operator, ctx, exhaust=exhaust)
        except Exception as error:
            return ExecutionResult(
                executor=executor,
                error=traceback.format_exc(),
                error_message=str(error),
            )

        return ExecutionResult(result=result, executor=executor)
Executes the operator with the given name.

    Args:
        operator_uri: the URI of the operator
        request_params: a dictionary of parameters for the operator
        exhaust (False): whether to immediately exhaust generator operators

    Returns:
        an :class:`ExecutionResult`
    
fiftyone/operators/executor.py