user_defined.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""Provides the user-defined functions for evaluation of dataframes."""

from __future__ import annotations

import functools
import sys
from abc import ABCMeta, abstractmethod
from enum import Enum
from typing import TYPE_CHECKING, Any, Protocol, TypeGuard, TypeVar, cast, overload

import pyarrow as pa

import datafusion._internal as df_internal
from datafusion import SessionContext
from datafusion.expr import Expr

# Imported at runtime so ``typing.get_type_hints`` resolves the capsule
# overloads; typing_extensions is a runtime dependency below 3.13.
if sys.version_info >= (3, 13):
    from types import CapsuleType as _PyCapsule
else:
    from typing_extensions import CapsuleType as _PyCapsule

if TYPE_CHECKING:
    _R = TypeVar("_R", bound=pa.Array)
    from collections.abc import Callable, Sequence


class Volatility(Enum):
    """Defines how stable or volatile a function is.

    When setting the volatility of a function, you can either pass this
    enumeration or a ``str``. The ``str`` equivalent is the lower case value of the
    name (`"immutable"`, `"stable"`, or `"volatile"`).
    """

    Immutable = 1
    """An immutable function will always return the same output when given the
    same input.

    DataFusion will attempt to inline immutable functions during planning.
    """

    Stable = 2
    """
    Returns the same value for a given input within a single queries.

    A stable function may return different values given the same input across
    different queries but must return the same value for a given input within a
    query. An example of this is the ``Now`` function. DataFusion will attempt to
    inline ``Stable`` functions during planning, when possible. For query
    ``select col1, now() from t1``, it might take a while to execute but ``now()``
    column will be the same for each output row, which is evaluated during
    planning.
    """

    Volatile = 3
    """A volatile function may change the return value from evaluation to
    evaluation.

    Multiple invocations of a volatile function may return different results
    when used in the same query. An example of this is the random() function.
    DataFusion can not evaluate such functions during planning. In the query
    ``select col1, random() from t1``, ``random()`` function will be evaluated
    for each output row, resulting in a unique random value for each row.
    """

    def __str__(self) -> str:
        """Returns the string equivalent."""
        return self.name.lower()


def data_type_or_field_to_field(value: pa.DataType | pa.Field, name: str) -> pa.Field:
    """Helper function to return a Field from either a Field or DataType."""
    if isinstance(value, pa.Field):
        return value
    return pa.field(name, type=value)


def data_types_or_fields_to_field_list(
    inputs: Sequence[pa.Field | pa.DataType] | pa.Field | pa.DataType,
) -> list[pa.Field]:
    """Helper function to return a list of Fields."""
    if isinstance(inputs, pa.DataType):
        return [pa.field("value", type=inputs)]
    if isinstance(inputs, pa.Field):
        return [inputs]

    return [
        data_type_or_field_to_field(v, f"value_{idx}") for (idx, v) in enumerate(inputs)
    ]


class ScalarUDFExportable(Protocol):
    """Type hint for object that has __datafusion_scalar_udf__ PyCapsule."""

    def __datafusion_scalar_udf__(self) -> object: ...  # noqa: D105


def _is_pycapsule(value: object) -> TypeGuard[_PyCapsule]:
    """Return ``True`` when ``value`` is a CPython ``PyCapsule``."""
    return value.__class__.__name__ == "PyCapsule"


class LogicalExtensionCodecExportable(Protocol):
    """Type hint for objects exposing ``__datafusion_logical_extension_codec__``.

    Implement the codec itself exactly as you would for a session that installs
    only yours. A session may hold several codecs, but each payload records the
    codec that wrote it and is only ever handed back to that codec, so there is
    no need to recognise or reject another library's payloads.

    A serialized plan records which codec wrote each payload, as a short id
    taken from your class's module and qualified name. An optional
    ``__datafusion_codec_id__`` attribute pins that id instead. It is not part
    of this protocol and is rarely needed: declare it when renaming your class
    must not stop older plans from decoding, or when one library installs two
    instances that own disjoint slices of the wire format.

    Args:
        session: A handle on the session this codec is being installed on. Take
            the task context provider from it rather than building a session of
            your own, so the decode callbacks resolve names against the session
            that runs the query.

            **Duck-type it; do not check its type.** This argument is the
            reference for every capsule getter in the protocol, and across them
            it is not always a session at all: the provider and catalog getters
            receive the host's logical codec as a bare ``PyCapsule``. Even when
            it *is* a session it is the PyO3 context from
            ``datafusion._internal``, not the
            :py:class:`~datafusion.context.SessionContext` wrapper, so it
            exposes every capsule getter and ``__datafusion_codec_id__`` —
            everything the protocol asks of it — but
            ``isinstance(session, SessionContext)`` is ``False`` even though its
            ``repr`` reads ``datafusion.SessionContext``. Pass it to
            ``ffi_logical_codec_from_pycapsule``, which handles both.

            The two bundle hooks are the exception:
            :py:class:`~datafusion.extensions.SessionComponentsExportable` and
            :py:class:`~datafusion.extensions.SessionPlannerExportable` are
            dispatched from Python and receive the wrapper. See
            :ref:`extension_getter_argument`.

    Returns:
        A ``datafusion_logical_extension_codec`` PyCapsule.

    Examples:
        A :py:class:`~datafusion.context.SessionContext` satisfies this
        protocol, which is how you obtain a real capsule to test against:

        >>> from datafusion import SessionContext
        >>> ctx = SessionContext()
        >>> type(ctx.__datafusion_logical_extension_codec__(ctx)).__name__
        'PyCapsule'
    """

    def __datafusion_logical_extension_codec__(  # noqa: D105
        self, session: Any
    ) -> object: ...


class PhysicalExtensionCodecExportable(Protocol):
    """Type hint for objects exposing ``__datafusion_physical_extension_codec__``.

    See :py:class:`LogicalExtensionCodecExportable` for ``session``, for why a
    codec need not recognise other libraries' payloads, and for
    ``__datafusion_codec_id__``.
    """

    def __datafusion_physical_extension_codec__(  # noqa: D105
        self, session: Any
    ) -> object: ...


class ScalarUDF:
    """Class for performing scalar user-defined functions (UDF).

    Scalar UDFs operate on a row by row basis. See also :py:class:`AggregateUDF` for
    operating on a group of rows.
    """

    @overload
    def __init__(
        self,
        name: str,
        func: Callable[..., _R],
        input_fields: list[pa.Field],
        return_field: pa.Field,
        volatility: Volatility | str,
    ) -> None: ...

    @overload
    def __init__(
        self,
        name: str,
        func: ScalarUDFExportable,
        input_fields: None = ...,
        return_field: None = ...,
        volatility: None = ...,
    ) -> None: ...

    def __init__(
        self,
        name: str,
        func: Callable[..., _R] | ScalarUDFExportable,
        input_fields: list[pa.Field] | None = None,
        return_field: pa.Field | None = None,
        volatility: Volatility | str | None = None,
    ) -> None:
        """Instantiate a scalar user-defined function (UDF).

        See helper method :py:func:`udf` for argument details.
        """
        if hasattr(func, "__datafusion_scalar_udf__"):
            self._udf = df_internal.ScalarUDF.from_pycapsule(func)
            return
        if input_fields is None or return_field is None or volatility is None:
            msg = (
                "`input_fields`, `return_field`, and `volatility` "
                "must be provided when `func` is callable."
            )
            raise TypeError(msg)
        if isinstance(input_fields, pa.DataType):
            input_fields = [input_fields]
        self._udf = df_internal.ScalarUDF(
            name, func, input_fields, return_field, str(volatility)
        )

    @classmethod
    def _from_internal(cls, internal: df_internal.ScalarUDF) -> ScalarUDF:
        """Wrap an already-constructed internal ``ScalarUDF`` handle.

        Used by :py:meth:`SessionContext.udf` and :py:meth:`from_pycapsule`
        to wrap a handle from the session's function registry or an FFI
        capsule without re-running :py:meth:`__init__`.
        """
        wrapper = cls.__new__(cls)
        wrapper._udf = internal
        return wrapper

    @property
    def name(self) -> str:
        """Return the registered name of this UDF.

        For UDFs imported via the FFI capsule protocol, this is the
        name the capsule itself reports — not the ``name`` argument
        passed to the constructor (which is ignored on the FFI path).

        Examples:
            >>> import pyarrow as pa
            >>> from datafusion import udf
            >>> double = udf(
            ...     lambda arr: pa.array([(v.as_py() or 0) * 2 for v in arr]),
            ...     [pa.int64()],
            ...     pa.int64(),
            ...     volatility="immutable",
            ...     name="double",
            ... )
            >>> double.name
            'double'
        """
        return self._udf.name

    def __repr__(self) -> str:
        """Print a string representation of the Scalar UDF."""
        return self._udf.__repr__()

    def __call__(self, *args: Expr) -> Expr:
        """Execute the UDF.

        This function is not typically called by an end user. These calls will
        occur during the evaluation of the dataframe.
        """
        args_raw = [arg.expr for arg in args]
        return Expr(self._udf.__call__(*args_raw))

    @overload
    @staticmethod
    def udf(
        input_fields: Sequence[pa.DataType | pa.Field] | pa.DataType | pa.Field,
        return_field: pa.DataType | pa.Field,
        volatility: Volatility | str,
        name: str | None = None,
    ) -> Callable[..., ScalarUDF]: ...

    @overload
    @staticmethod
    def udf(
        func: Callable[..., _R],
        input_fields: Sequence[pa.DataType | pa.Field] | pa.DataType | pa.Field,
        return_field: pa.DataType | pa.Field,
        volatility: Volatility | str,
        name: str | None = None,
    ) -> ScalarUDF: ...

    @overload
    @staticmethod
    def udf(func: ScalarUDFExportable) -> ScalarUDF: ...

    @overload
    @staticmethod
    def udf(func: _PyCapsule) -> ScalarUDF: ...

    @staticmethod
    def udf(*args: Any, **kwargs: Any):  # noqa: D417
        """Create a new User-Defined Function (UDF).

        This class can be used both as either a function or a decorator.

        Usage:
            - As a function: ``udf(func, input_fields, return_field,
              volatility, name)``.
            - As a decorator: ``@udf(input_fields, return_field, volatility, name)``.
              When used a decorator, do **not** pass ``func`` explicitly.

        In lieu of passing a PyArrow Field, you can pass a DataType for simplicity.
        When you do so, it will be assumed that the nullability of the inputs and
        output are True and that they have no metadata.

        Args:
            func (Callable, optional): Only needed when calling as a function.
                Skip this argument when using `udf` as a decorator. If you have a Rust
                backed ScalarUDF within a PyCapsule, you can pass this parameter
                and ignore the rest. They will be determined directly from the
                underlying function. See the online documentation for more information.
            input_fields (list[pa.Field | pa.DataType]): The data types or Fields
                of the arguments to ``func``. This list must be of the same length
                as the number of arguments.
            return_field (pa.DataType | pa.Field): The field of the return value
                from the function.
            volatility (Volatility | str): See `Volatility` for allowed values.
            name (Optional[str]): A descriptive name for the function.

        Returns:
            A user-defined function that can be used in SQL expressions,
            data aggregation, or window function calls.

        Examples:
            Using ``udf`` as a function:

            >>> import pyarrow.compute as pc
            >>> from datafusion.user_defined import ScalarUDF
            >>> def double_func(x):
            ...     return pc.multiply(x, 2)
            >>> double_udf = ScalarUDF.udf(
            ...     double_func, [pa.int64()], pa.int64(),
            ...     "volatile", "double_it")

            Using ``udf`` as a decorator:

            >>> @ScalarUDF.udf([pa.int64()], pa.int64(), "volatile")
            ... def decorator_double_udf(x):
            ...     return pc.multiply(x, 3)

            Apply to a dataframe:

            >>> ctx = dfn.SessionContext()
            >>> df = ctx.from_pydict({"x": [1, 2, 3]})
            >>> df.select(double_udf(col("x")).alias("result")).to_pydict()
            {'result': [2, 4, 6]}
            >>> df.select(decorator_double_udf(col("x")).alias("result")).to_pydict()
            {'result': [3, 6, 9]}
        """

        def _function(
            func: Callable[..., _R],
            input_fields: Sequence[pa.DataType | pa.Field] | pa.DataType | pa.Field,
            return_field: pa.DataType | pa.Field,
            volatility: Volatility | str,
            name: str | None = None,
        ) -> ScalarUDF:
            if not callable(func):
                msg = "`func` argument must be callable"
                raise TypeError(msg)
            if name is None:
                name = getattr(func, "__qualname__", func.__class__.__name__).lower()
            input_fields = data_types_or_fields_to_field_list(input_fields)
            return_field = data_type_or_field_to_field(return_field, "value")
            return ScalarUDF(
                name=name,
                func=func,
                input_fields=input_fields,
                return_field=return_field,
                volatility=volatility,
            )

        def _decorator(
            input_fields: Sequence[pa.DataType | pa.Field] | pa.DataType | pa.Field,
            return_field: pa.DataType | pa.Field,
            volatility: Volatility | str,
            name: str | None = None,
        ) -> Callable:
            def decorator(func: Callable) -> Callable:
                udf_caller = ScalarUDF.udf(
                    func, input_fields, return_field, volatility, name
                )

                @functools.wraps(func)
                def wrapper(*args: Any, **kwargs: Any) -> Expr:
                    return udf_caller(*args, **kwargs)

                return wrapper

            return decorator

        if not args and "func" in kwargs:
            args = (kwargs.pop("func"),)
        if args and (
            hasattr(args[0], "__datafusion_scalar_udf__") or _is_pycapsule(args[0])
        ):
            return ScalarUDF.from_pycapsule(args[0])

        if args and callable(args[0]):
            # Case 1: Used as a function, require the first parameter to be callable
            return _function(*args, **kwargs)
        # Case 2: Used as a decorator with parameters
        return _decorator(*args, **kwargs)

    @staticmethod
    def from_pycapsule(func: ScalarUDFExportable | _PyCapsule) -> ScalarUDF:
        """Create a Scalar UDF from ScalarUDF PyCapsule object.

        This function will instantiate a Scalar UDF that uses a DataFusion
        ScalarUDF that is exported via the FFI bindings.
        """
        if _is_pycapsule(func):
            return ScalarUDF._from_internal(df_internal.ScalarUDF.from_pycapsule(func))

        func = cast("ScalarUDFExportable", func)
        name = str(func.__class__)
        return ScalarUDF(
            name=name,
            func=func,
            input_fields=None,
            return_field=None,
            volatility=None,
        )


class Accumulator(metaclass=ABCMeta):
    """Defines how an :py:class:`AggregateUDF` accumulates values."""

    @abstractmethod
    def state(self) -> list[pa.Scalar]:
        """Return the current state.

        While this function template expects PyArrow Scalar values return type,
        you can return any value that can be converted into a Scalar. This
        includes basic Python data types such as integers and strings. In
        addition to primitive types, we currently support PyArrow, nanoarrow,
        and arro3 objects in addition to primitive data types. Other objects
        that support the Arrow FFI standard will be given a "best attempt" at
        conversion to scalar objects.
        """

    @abstractmethod
    def update(self, *values: pa.Array) -> None:
        """Evaluate an array of values and update state."""

    @abstractmethod
    def merge(self, states: list[pa.Array]) -> None:
        """Merge a set of states."""

    @abstractmethod
    def evaluate(self) -> pa.Scalar:
        """Return the resultant value.

        While this function template expects a PyArrow Scalar value return type,
        you can return any value that can be converted into a Scalar. This
        includes basic Python data types such as integers and strings. In
        addition to primitive types, we currently support PyArrow, nanoarrow,
        and arro3 objects in addition to primitive data types. Other objects
        that support the Arrow FFI standard will be given a "best attempt" at
        conversion to scalar objects.
        """


class AggregateUDFExportable(Protocol):
    """Type hint for object that has __datafusion_aggregate_udf__ PyCapsule."""

    def __datafusion_aggregate_udf__(self) -> object: ...  # noqa: D105


class AggregateUDF:
    """Class for performing scalar user-defined functions (UDF).

    Aggregate UDFs operate on a group of rows and return a single value. See
    also :py:class:`ScalarUDF` for operating on a row by row basis.
    """

    @overload
    def __init__(
        self,
        name: str,
        accumulator: Callable[[], Accumulator],
        input_types: list[pa.DataType],
        return_type: pa.DataType,
        state_type: list[pa.DataType],
        volatility: Volatility | str,
    ) -> None: ...

    @overload
    def __init__(
        self,
        name: str,
        accumulator: AggregateUDFExportable,
        input_types: None = ...,
        return_type: None = ...,
        state_type: None = ...,
        volatility: None = ...,
    ) -> None: ...

    def __init__(
        self,
        name: str,
        accumulator: Callable[[], Accumulator] | AggregateUDFExportable,
        input_types: list[pa.DataType] | None = None,
        return_type: pa.DataType | None = None,
        state_type: list[pa.DataType] | None = None,
        volatility: Volatility | str | None = None,
    ) -> None:
        """Instantiate a user-defined aggregate function (UDAF).

        See :py:func:`udaf` for a convenience function and argument
        descriptions.
        """
        if hasattr(accumulator, "__datafusion_aggregate_udf__"):
            self._udaf = df_internal.AggregateUDF.from_pycapsule(accumulator)
            return
        if (
            input_types is None
            or return_type is None
            or state_type is None
            or volatility is None
        ):
            msg = (
                "`input_types`, `return_type`, `state_type`, and `volatility` "
                "must be provided when `accumulator` is callable."
            )
            raise TypeError(msg)

        self._udaf = df_internal.AggregateUDF(
            name,
            accumulator,
            input_types,
            return_type,
            state_type,
            str(volatility),
        )

    @classmethod
    def _from_internal(cls, internal: df_internal.AggregateUDF) -> AggregateUDF:
        """Wrap an already-constructed internal ``AggregateUDF`` handle.

        Used by :py:meth:`SessionContext.udaf` and :py:meth:`from_pycapsule`
        to wrap a handle from the session's function registry or an FFI
        capsule without re-running :py:meth:`__init__`.
        """
        wrapper = cls.__new__(cls)
        wrapper._udaf = internal
        return wrapper

    @property
    def name(self) -> str:
        """Return the registered name of this UDAF.

        For UDAFs imported via the FFI capsule protocol, this is the
        name the capsule itself reports — not the ``name`` argument
        passed to the constructor (which is ignored on the FFI path).
        """
        return self._udaf.name

    def __repr__(self) -> str:
        """Print a string representation of the Aggregate UDF."""
        return self._udaf.__repr__()

    def __call__(self, *args: Expr) -> Expr:
        """Execute the UDAF.

        This function is not typically called by an end user. These calls will
        occur during the evaluation of the dataframe.
        """
        args_raw = [arg.expr for arg in args]
        return Expr(self._udaf.__call__(*args_raw))

    @overload
    @staticmethod
    def udaf(
        input_types: pa.DataType | list[pa.DataType],
        return_type: pa.DataType,
        state_type: list[pa.DataType],
        volatility: Volatility | str,
        name: str | None = None,
    ) -> Callable[..., AggregateUDF]: ...

    @overload
    @staticmethod
    def udaf(
        accum: Callable[[], Accumulator],
        input_types: pa.DataType | list[pa.DataType],
        return_type: pa.DataType,
        state_type: list[pa.DataType],
        volatility: Volatility | str,
        name: str | None = None,
    ) -> AggregateUDF: ...

    @overload
    @staticmethod
    def udaf(accum: AggregateUDFExportable) -> AggregateUDF: ...

    @overload
    @staticmethod
    def udaf(accum: _PyCapsule) -> AggregateUDF: ...

    @staticmethod
    def udaf(*args: Any, **kwargs: Any):  # noqa: D417, C901
        """Create a new User-Defined Aggregate Function (UDAF).

        This class allows you to define an aggregate function that can be used in
        data aggregation or window function calls.

        Usage:
            - As a function: ``udaf(accum, input_types, return_type, state_type, volatility, name)``.
            - As a decorator: ``@udaf(input_types, return_type, state_type, volatility, name)``.
              When using ``udaf`` as a decorator, do not pass ``accum`` explicitly.

        If your :py:class:`Accumulator` can be instantiated with no arguments, you
        can simply pass its type as ``accum``. If you need to pass additional
        arguments to its constructor, you can define a lambda or a factory method.
        During runtime the :py:class:`Accumulator` will be constructed for every
        instance in which this UDAF is used.

        Examples:
            >>> import pyarrow.compute as pc
            >>> from datafusion.user_defined import AggregateUDF, Accumulator, udaf
            >>> class Summarize(Accumulator):
            ...     def __init__(self, bias: float = 0.0):
            ...         self._sum = pa.scalar(bias)
            ...     def state(self):
            ...         return [self._sum]
            ...     def update(self, values):
            ...         self._sum = pa.scalar(
            ...             self._sum.as_py() + pc.sum(values).as_py())
            ...     def merge(self, states):
            ...         self._sum = pa.scalar(
            ...             self._sum.as_py() + pc.sum(states[0]).as_py())
            ...     def evaluate(self):
            ...         return self._sum

            Using ``udaf`` as a function:

            >>> udaf1 = AggregateUDF.udaf(
            ...     Summarize, pa.float64(), pa.float64(),
            ...     [pa.float64()], "immutable")

            Wrapping ``udaf`` with a function:

            >>> def sum_bias_10() -> Summarize:
            ...     return Summarize(10.0)
            >>> udaf2 = udaf(sum_bias_10, pa.float64(), pa.float64(), [pa.float64()],
            ...     "immutable")

            Using ``udaf`` with lambda:

            >>> udaf3 = udaf(lambda: Summarize(20.0), pa.float64(), pa.float64(),
            ...     [pa.float64()], "immutable")

            Using ``udaf`` as a decorator:

            >>> @AggregateUDF.udaf(
            ...     pa.float64(), pa.float64(),
            ...     [pa.float64()], "immutable")
            ... def udaf4():
            ...     return Summarize(10.0)

            Apply to a dataframe:

            >>> ctx = dfn.SessionContext()
            >>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0]})
            >>> df.aggregate([], [udaf1(col("a")).alias("total")]).collect_column(
            ...     "total")[0].as_py()
            6.0
            >>> df.aggregate([], [udaf2(col("a")).alias("total")]).collect_column(
            ...     "total")[0].as_py()
            16.0
            >>> df.aggregate([], [udaf3(col("a")).alias("total")]).collect_column(
            ...     "total")[0].as_py()
            26.0
            >>> df.aggregate([], [udaf4(col("a")).alias("total")]).collect_column(
            ...     "total")[0].as_py()
            16.0

        Args:
            accum: The accumulator python function. Only needed when calling as a
                function. Skip this argument when using ``udaf`` as a decorator.
                If you have a Rust backed AggregateUDF within a PyCapsule, you can
                pass this parameter and ignore the rest. They will be determined
                directly from the underlying function. See the online documentation
                for more information.
            input_types: The data types of the arguments to ``accum``.
            return_type: The data type of the return value.
            state_type: The data types of the intermediate accumulation.
            volatility: See :py:class:`Volatility` for allowed values.
            name: A descriptive name for the function.

        Returns:
            A user-defined aggregate function, which can be used in either data
            aggregation or window function calls.
        """  # noqa: E501 W505

        def _function(
            accum: Callable[[], Accumulator],
            input_types: pa.DataType | list[pa.DataType],
            return_type: pa.DataType,
            state_type: list[pa.DataType],
            volatility: Volatility | str,
            name: str | None = None,
        ) -> AggregateUDF:
            if not callable(accum):
                msg = "`func` must be callable."
                raise TypeError(msg)
            if not isinstance(accum(), Accumulator):
                msg = "Accumulator must implement the abstract base class Accumulator"
                raise TypeError(msg)
            if name is None:
                name = accum().__class__.__qualname__.lower()
            if isinstance(input_types, pa.DataType):
                input_types = [input_types]
            return AggregateUDF(
                name=name,
                accumulator=accum,
                input_types=input_types,
                return_type=return_type,
                state_type=state_type,
                volatility=volatility,
            )

        def _decorator(
            input_types: pa.DataType | list[pa.DataType],
            return_type: pa.DataType,
            state_type: list[pa.DataType],
            volatility: Volatility | str,
            name: str | None = None,
        ) -> Callable[..., Callable[..., Expr]]:
            def decorator(accum: Callable[[], Accumulator]) -> Callable[..., Expr]:
                udaf_caller = AggregateUDF.udaf(
                    accum, input_types, return_type, state_type, volatility, name
                )

                @functools.wraps(accum)
                def wrapper(*args: Any, **kwargs: Any) -> Expr:
                    return udaf_caller(*args, **kwargs)

                return wrapper

            return decorator

        if not args and "accum" in kwargs:
            args = (kwargs.pop("accum"),)
        if args and (
            hasattr(args[0], "__datafusion_aggregate_udf__") or _is_pycapsule(args[0])
        ):
            return AggregateUDF.from_pycapsule(args[0])

        if args and callable(args[0]):
            # Case 1: Used as a function, require the first parameter to be callable
            return _function(*args, **kwargs)
        # Case 2: Used as a decorator with parameters
        return _decorator(*args, **kwargs)

    @staticmethod
    def from_pycapsule(func: AggregateUDFExportable | _PyCapsule) -> AggregateUDF:
        """Create an Aggregate UDF from AggregateUDF PyCapsule object.

        This function will instantiate a Aggregate UDF that uses a DataFusion
        AggregateUDF that is exported via the FFI bindings.
        """
        if _is_pycapsule(func):
            return AggregateUDF._from_internal(
                df_internal.AggregateUDF.from_pycapsule(func)
            )

        capsule = cast("AggregateUDFExportable", func)
        name = str(capsule.__class__)
        return AggregateUDF(
            name=name,
            accumulator=capsule,
            input_types=None,
            return_type=None,
            state_type=None,
            volatility=None,
        )


class WindowEvaluator:
    """Evaluator class for user-defined window functions (UDWF).

    It is up to the user to decide which evaluate function is appropriate.

    +------------------------+--------------------------------+------------------+---------------------------+
    | ``uses_window_frame``  | ``supports_bounded_execution`` | ``include_rank`` | function_to_implement     |
    +========================+================================+==================+===========================+
    | False (default)        | False (default)                | False (default)  | ``evaluate_all``          |
    +------------------------+--------------------------------+------------------+---------------------------+
    | False                  | True                           | False            | ``evaluate``              |
    +------------------------+--------------------------------+------------------+---------------------------+
    | False                  | True/False                     | True             | ``evaluate_all_with_rank``|
    +------------------------+--------------------------------+------------------+---------------------------+
    | True                   | True/False                     | True/False       | ``evaluate``              |
    +------------------------+--------------------------------+------------------+---------------------------+
    """  # noqa: W505, E501

    def memoize(self) -> None:
        """Perform a memoize operation to improve performance.

        When the window frame has a fixed beginning (e.g UNBOUNDED
        PRECEDING), some functions such as FIRST_VALUE and
        NTH_VALUE do not need the (unbounded) input once they have
        seen a certain amount of input.

        `memoize` is called after each input batch is processed, and
        such functions can save whatever they need
        """

    def get_range(self, idx: int, num_rows: int) -> tuple[int, int]:  # noqa: ARG002
        """Return the range for the window function.

        If `uses_window_frame` flag is `false`. This method is used to
        calculate required range for the window function during
        stateful execution.

        Generally there is no required range, hence by default this
        returns smallest range(current row). e.g seeing current row is
        enough to calculate window result (such as row_number, rank,
        etc)

        Args:
            idx:: Current index
            num_rows: Number of rows.
        """
        return (idx, idx + 1)

    def is_causal(self) -> bool:
        """Get whether evaluator needs future data for its result."""
        return False

    def evaluate_all(self, values: list[pa.Array], num_rows: int) -> pa.Array:
        """Evaluate a window function on an entire input partition.

        This function is called once per input *partition* for window functions that
        *do not use* values from the window frame, such as
        :py:func:`~datafusion.functions.row_number`,
        :py:func:`~datafusion.functions.rank`,
        :py:func:`~datafusion.functions.dense_rank`,
        :py:func:`~datafusion.functions.percent_rank`,
        :py:func:`~datafusion.functions.cume_dist`,
        :py:func:`~datafusion.functions.lead`,
        and :py:func:`~datafusion.functions.lag`.

        It produces the result of all rows in a single pass. It
        expects to receive the entire partition as the ``value`` and
        must produce an output column with one output row for every
        input row.

        ``num_rows`` is required to correctly compute the output in case
        ``len(values) == 0``

        Implementing this function is an optimization. Certain window
        functions are not affected by the window frame definition or
        the query doesn't have a frame, and ``evaluate`` skips the
        (costly) window frame boundary calculation and the overhead of
        calling ``evaluate`` for each output row.

        For example, the `LAG` built in window function does not use
        the values of its window frame (it can be computed in one shot
        on the entire partition with ``Self::evaluate_all`` regardless of the
        window defined in the ``OVER`` clause)

        .. code-block:: text

            lag(x, 1) OVER (ORDER BY z ROWS BETWEEN 2 PRECEDING AND 3 FOLLOWING)

        However, ``avg()`` computes the average in the window and thus
        does use its window frame.

        .. code-block:: text

            avg(x) OVER (PARTITION BY y ORDER BY z ROWS BETWEEN 2 PRECEDING AND 3 FOLLOWING)
        """  # noqa: W505, E501

    def evaluate(
        self, values: list[pa.Array], eval_range: tuple[int, int]
    ) -> pa.Scalar:
        """Evaluate window function on a range of rows in an input partition.

        This is the simplest and most general function to implement
        but also the least performant as it creates output one row at
        a time. It is typically much faster to implement stateful
        evaluation using one of the other specialized methods on this
        trait.

        Returns a [`ScalarValue`] that is the value of the window
        function within `range` for the entire partition. Argument
        `values` contains the evaluation result of function arguments
        and evaluation results of ORDER BY expressions. If function has a
        single argument, `values[1..]` will contain ORDER BY expression results.
        """

    def evaluate_all_with_rank(
        self, num_rows: int, ranks_in_partition: list[tuple[int, int]]
    ) -> pa.Array:
        """Called for window functions that only need the rank of a row.

        Evaluate the partition evaluator against the partition using
        the row ranks. For example, ``rank(col("a"))`` produces

        .. code-block:: text

            a | rank
            - + ----
            A | 1
            A | 1
            C | 3
            D | 4
            D | 4

        For this case, `num_rows` would be `5` and the
        `ranks_in_partition` would be called with

        .. code-block:: text

            [
                (0,1),
                (2,2),
                (3,4),
            ]

        The user must implement this method if ``include_rank`` returns True.
        """

    def supports_bounded_execution(self) -> bool:
        """Can the window function be incrementally computed using bounded memory?"""
        return False

    def uses_window_frame(self) -> bool:
        """Does the window function use the values from the window frame?"""
        return False

    def include_rank(self) -> bool:
        """Can this function be evaluated with (only) rank?"""
        return False


class WindowUDFExportable(Protocol):
    """Type hint for object that has __datafusion_window_udf__ PyCapsule."""

    def __datafusion_window_udf__(self) -> object: ...  # noqa: D105


class WindowUDF:
    """Class for performing window user-defined functions (UDF).

    Window UDFs operate on a partition of rows. See
    also :py:class:`ScalarUDF` for operating on a row by row basis.
    """

    @overload
    def __init__(
        self,
        name: str,
        func: Callable[[], WindowEvaluator],
        input_types: list[pa.DataType],
        return_type: pa.DataType,
        volatility: Volatility | str,
    ) -> None: ...

    @overload
    def __init__(
        self,
        name: str,
        func: WindowUDFExportable,
        input_types: None = ...,