# 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.

""":py:class:`Expr` — the logical expression type used to build DataFusion queries.

An :py:class:`Expr` represents a computation over columns or literals: a

column reference (``col("a")``), a literal (``lit(5)``), an operator

combination (``col("a") + lit(1)``), or the output of a function from

:py:mod:`datafusion.functions`. Expressions are passed to

:py:class:`~datafusion.dataframe.DataFrame` methods such as

:py:meth:`~datafusion.dataframe.DataFrame.select`,

:py:meth:`~datafusion.dataframe.DataFrame.filter`,

:py:meth:`~datafusion.dataframe.DataFrame.aggregate`, and

:py:meth:`~datafusion.dataframe.DataFrame.sort`.

Convenience constructors are re-exported at the package level:

:py:func:`datafusion.col` / :py:func:`datafusion.column` for column references

and :py:func:`datafusion.lit` / :py:func:`datafusion.literal` for scalar

literals.

Examples:

>>> ctx = dfn.SessionContext()

>>> df = ctx.from_pydict({"a": [1, 2, 3]})

>>> df.select((col("a") * lit(10)).alias("ten_a")).to_pydict()

{'ten_a': [10, 20, 30]}

See :ref:`expressions` in the online documentation for details on available

operators and helpers.

"""

# ruff: noqa: PLC0415

from __future__ import annotations

from collections.abc import Callable, Iterable, Sequence

from typing import TYPE_CHECKING, Any, ClassVar

try:

from warnings import deprecated # Python 3.13+

except ImportError:

from typing_extensions import deprecated # Python 3.12

import pyarrow as pa

from ._internal import expr as expr_internal

from ._internal import functions as functions_internal

from .ipc import get_sender_ctx

if TYPE_CHECKING:

from collections.abc import Sequence

from datafusion.common import ( # type: ignore[import]

DataTypeMap,

NullTreatment,

RexType,

)

from datafusion.context import SessionContext

from datafusion.plan import LogicalPlan

# Standard error message for invalid expression types

# Mention both alias forms of column and literal helpers

EXPR_TYPE_ERROR = "Use col()/column() or lit()/literal() to construct expressions"

# The following are imported from the internal representation. We may choose to

# give these all proper wrappers, or to simply leave as is. These were added

# in order to support passing the `test_imports` unit test.

# Tim Saucer note: It is not clear to me what the use case is for exposing

# these definitions to the end user.

Alias = expr_internal.Alias

Analyze = expr_internal.Analyze

Aggregate = expr_internal.Aggregate

AggregateFunction = expr_internal.AggregateFunction

Between = expr_internal.Between

BinaryExpr = expr_internal.BinaryExpr

Case = expr_internal.Case

Cast = expr_internal.Cast

Column = expr_internal.Column

CopyTo = expr_internal.CopyTo

CreateCatalog = expr_internal.CreateCatalog

CreateCatalogSchema = expr_internal.CreateCatalogSchema

CreateExternalTable = expr_internal.CreateExternalTable

@deprecated("CreateExternalTable.location() is deprecated; use locations() instead.")

def _create_external_table_location(self: Any) -> str:

"""Return the first external table location.

Use :meth:`CreateExternalTable.locations` instead.

Examples:

>>> class Command:

... def locations(self) -> list[str]:

... return ["data.csv"]

>>> Command().locations()

['data.csv']

"""

locations = self.locations()

return locations[0] if locations else ""

CreateExternalTable.location = _create_external_table_location

CreateFunction = expr_internal.CreateFunction

CreateFunctionBody = expr_internal.CreateFunctionBody

CreateIndex = expr_internal.CreateIndex

CreateMemoryTable = expr_internal.CreateMemoryTable

CreateView = expr_internal.CreateView

Deallocate = expr_internal.Deallocate

DescribeTable = expr_internal.DescribeTable

Distinct = expr_internal.Distinct

DmlStatement = expr_internal.DmlStatement

DropCatalogSchema = expr_internal.DropCatalogSchema

DropFunction = expr_internal.DropFunction

DropTable = expr_internal.DropTable

DropView = expr_internal.DropView

EmptyRelation = expr_internal.EmptyRelation

Execute = expr_internal.Execute

Exists = expr_internal.Exists

Explain = expr_internal.Explain

Extension = expr_internal.Extension

FileType = expr_internal.FileType

Filter = expr_internal.Filter

Join = expr_internal.Join

ILike = expr_internal.ILike

InList = expr_internal.InList

InSubquery = expr_internal.InSubquery

IsFalse = expr_internal.IsFalse

IsNotTrue = expr_internal.IsNotTrue

IsNull = expr_internal.IsNull

IsTrue = expr_internal.IsTrue

IsUnknown = expr_internal.IsUnknown

IsNotFalse = expr_internal.IsNotFalse

IsNotNull = expr_internal.IsNotNull

IsNotUnknown = expr_internal.IsNotUnknown

JoinConstraint = expr_internal.JoinConstraint

JoinType = expr_internal.JoinType

Like = expr_internal.Like

Limit = expr_internal.Limit

Literal = expr_internal.Literal

Negative = expr_internal.Negative

Not = expr_internal.Not

OperateFunctionArg = expr_internal.OperateFunctionArg

Partitioning = expr_internal.Partitioning

Placeholder = expr_internal.Placeholder

Prepare = expr_internal.Prepare

Projection = expr_internal.Projection

RecursiveQuery = expr_internal.RecursiveQuery

Repartition = expr_internal.Repartition

ScalarSubquery = expr_internal.ScalarSubquery

ScalarVariable = expr_internal.ScalarVariable

SetVariable = expr_internal.SetVariable

SimilarTo = expr_internal.SimilarTo

Sort = expr_internal.Sort

Subquery = expr_internal.Subquery

SubqueryAlias = expr_internal.SubqueryAlias

TableScan = expr_internal.TableScan

TransactionAccessMode = expr_internal.TransactionAccessMode

TransactionConclusion = expr_internal.TransactionConclusion

TransactionEnd = expr_internal.TransactionEnd

TransactionIsolationLevel = expr_internal.TransactionIsolationLevel

TransactionStart = expr_internal.TransactionStart

TryCast = expr_internal.TryCast

Union = expr_internal.Union

HigherOrderFunction = expr_internal.HigherOrderFunction

Lambda = expr_internal.Lambda

LambdaVariable = expr_internal.LambdaVariable

Unnest = expr_internal.Unnest

UnnestExpr = expr_internal.UnnestExpr

Values = expr_internal.Values

WindowExpr = expr_internal.WindowExpr

__all__ = [

"EXPR_TYPE_ERROR",

"Aggregate",

"AggregateFunction",

"Alias",

"Analyze",

"Between",

"BinaryExpr",

"Case",

"CaseBuilder",

"Cast",

"Column",

"CopyTo",

"CreateCatalog",

"CreateCatalogSchema",

"CreateExternalTable",

"CreateFunction",

"CreateFunctionBody",

"CreateIndex",

"CreateMemoryTable",

"CreateView",

"Deallocate",

"DescribeTable",

"Distinct",

"DmlStatement",

"DropCatalogSchema",

"DropFunction",

"DropTable",

"DropView",

"EmptyRelation",

"Execute",

"Exists",

"Explain",

"Expr",

"Extension",

"FileType",

"Filter",

"GroupingSet",

"HigherOrderFunction",

"ILike",

"InList",

"InSubquery",

"IsFalse",

"IsNotFalse",

"IsNotNull",

"IsNotTrue",

"IsNotUnknown",

"IsNull",

"IsTrue",

"IsUnknown",

"Join",

"JoinConstraint",

"JoinType",

"Lambda",

"LambdaVariable",

"Like",

"Limit",

"Literal",

"Literal",

"Negative",

"Not",

"OperateFunctionArg",

"Partitioning",

"Placeholder",

"Prepare",

"Projection",

"RecursiveQuery",

"Repartition",

"ScalarSubquery",

"ScalarVariable",

"SetVariable",

"SimilarTo",

"Sort",

"SortExpr",

"SortKey",

"Subquery",

"SubqueryAlias",

"TableScan",

"TransactionAccessMode",

"TransactionConclusion",

"TransactionEnd",

"TransactionIsolationLevel",

"TransactionStart",

"TryCast",

"Union",

"Unnest",

"UnnestExpr",

"Values",

"Window",

"WindowExpr",

"WindowFrame",

"WindowFrameBound",

"coerce_to_expr",

"coerce_to_expr_list",

"coerce_to_expr_or_none",

"ensure_expr",

"ensure_expr_list",

]

def ensure_expr(value: Expr | Any) -> expr_internal.Expr:

"""Return the internal expression from ``Expr`` or raise ``TypeError``.

This helper rejects plain strings and other non-:class:`Expr` values so

higher level APIs consistently require explicit :func:`~datafusion.col` or

:func:`~datafusion.lit` expressions.

See Also:

:func:`coerce_to_expr` — the opposite behavior: *wraps* non-``Expr``

values as literals instead of rejecting them.

Args:

value: Candidate expression or other object.

Returns:

The internal expression representation.

Raises:

TypeError: If ``value`` is not an instance of :class:`Expr`.

"""

if not isinstance(value, Expr):

raise TypeError(EXPR_TYPE_ERROR)

return value.expr

def ensure_expr_list(

exprs: Iterable[Expr | Iterable[Expr]],

) -> list[expr_internal.Expr]:

"""Flatten an iterable of expressions, validating each via ``ensure_expr``.

Args:

exprs: Possibly nested iterable containing expressions.

Returns:

A flat list of raw expressions.

Raises:

TypeError: If any item is not an instance of :class:`Expr`.

"""

def _iter(

items: Iterable[Expr | Iterable[Expr]],

) -> Iterable[expr_internal.Expr]:

for expr in items:

if isinstance(expr, Iterable) and not isinstance(

expr, Expr | str | bytes | bytearray

):

# Treat string-like objects as atomic to surface standard errors

yield from _iter(expr)

else:

yield ensure_expr(expr)

return list(_iter(exprs))

def coerce_to_expr(value: Any) -> Expr:

"""Coerce a native Python value to an ``Expr`` literal, passing ``Expr`` through.

This is the complement of :func:`ensure_expr`: where ``ensure_expr``

*rejects* non-``Expr`` values, ``coerce_to_expr`` *wraps* them via

:meth:`Expr.literal` so that functions can accept native Python types

(``int``, ``float``, ``str``, ``bool``, etc.) alongside ``Expr``.

Args:

value: An ``Expr`` instance (returned as-is) or a Python literal to wrap.

Returns:

An ``Expr`` representing the value.

"""

if isinstance(value, Expr):

return value

return Expr.literal(value)

def coerce_to_expr_or_none(value: Any | None) -> Expr | None:

"""Coerce a value to ``Expr`` or pass ``None`` through unchanged.

Same as :func:`coerce_to_expr` but accepts ``None`` for optional parameters.

Args:

value: An ``Expr`` instance, a Python literal to wrap, or ``None``.

Returns:

An ``Expr`` representing the value, or ``None``.

"""

if value is None:

return None

return coerce_to_expr(value)

def coerce_to_expr_list(values: Iterable[Any]) -> list[Expr]:

"""Coerce each item in an iterable to ``Expr`` via :func:`coerce_to_expr`.

Args:

values: Iterable of ``Expr`` instances or Python literals to wrap.

Returns:

A list of ``Expr`` instances.

"""

return [coerce_to_expr(value) for value in values]

def _to_raw_expr(value: Expr | str) -> expr_internal.Expr:

"""Convert a Python expression or column name to its raw variant.

Args:

value: Candidate expression or column name.

Returns:

The internal :class:`~datafusion._internal.expr.Expr` representation.

Raises:

TypeError: If ``value`` is neither an :class:`Expr` nor ``str``.

"""

if isinstance(value, str):

return Expr.column(value).expr

if isinstance(value, Expr):

return value.expr

error = (

"Expected Expr or column name, found:"

f" {type(value).__name__}. {EXPR_TYPE_ERROR}."

)

raise TypeError(error)

def expr_list_to_raw_expr_list(

expr_list: list[Expr] | Expr | None,

) -> list[expr_internal.Expr] | None:

"""Convert a sequence of expressions or column names to raw expressions."""

if isinstance(expr_list, Expr | str):

expr_list = [expr_list]

if expr_list is None:

return None

return [_to_raw_expr(e) for e in expr_list]

def sort_or_default(e: Expr | SortExpr) -> expr_internal.SortExpr:

"""Helper function to return a default Sort if an Expr is provided."""

if isinstance(e, SortExpr):

return e.raw_sort

return SortExpr(e, ascending=True, nulls_first=True).raw_sort

def sort_list_to_raw_sort_list(

sort_list: Sequence[SortKey] | SortKey | None,

) -> list[expr_internal.SortExpr] | None:

"""Helper function to return an optional sort list to raw variant."""

if isinstance(sort_list, Expr | SortExpr | str):

sort_list = [sort_list]

if sort_list is None:

return None

raw_sort_list = []

for item in sort_list:

if isinstance(item, SortExpr):

raw_sort_list.append(sort_or_default(item))

else:

raw_expr = _to_raw_expr(item) # may raise ``TypeError``

raw_sort_list.append(sort_or_default(Expr(raw_expr)))

return raw_sort_list

class Expr: # noqa: PLW1641

"""Expression object.

Expressions are one of the core concepts in DataFusion. See

:ref:`Expressions` in the online documentation for more information.

"""

def __init__(self, expr: expr_internal.RawExpr) -> None:

"""This constructor should not be called by the end user."""

self.expr = expr

def to_variant(self) -> Any:

"""Convert this expression into a python object if possible."""

return self.expr.to_variant()

def schema_name(self) -> str:

"""Returns the name of this expression as it should appear in a schema.

This name will not include any CAST expressions.

"""

return self.expr.schema_name()

def canonical_name(self) -> str:

"""Returns a complete string representation of this expression."""

return self.expr.canonical_name()

def variant_name(self) -> str:

"""Returns the name of the Expr variant.

Ex: ``IsNotNull``, ``Literal``, ``BinaryExpr``, etc

"""

return self.expr.variant_name()

def to_bytes(self, ctx: SessionContext | None = None) -> bytes:

"""Serialize this expression to bytes for shipping to another process.

Use this — or :func:`pickle.dumps` — to send an expression to a

worker process for distributed evaluation.

When ``ctx`` is supplied, encoding routes through that session's

installed :class:`LogicalExtensionCodec` (so settings like

:meth:`SessionContext.with_python_udf_inlining` take effect).

When ``ctx`` is ``None``, the default codec is used (Python UDF

inlining on, no user-installed extension codec).

Built-in functions travel inside the returned bytes. Python UDFs

(scalar, aggregate, window) also inline by default, so the worker

does not need to pre-register them; when the encoding session has

:meth:`SessionContext.with_python_udf_inlining` set to ``False``,

Python UDFs travel by name only and must be registered on the

worker. UDFs imported via the FFI capsule protocol always travel

by name only and must be registered on the worker.

.. warning:: Security

Bytes returned here may embed a cloudpickled Python

callable (when the expression carries a Python UDF).

Reconstructing them via :meth:`from_bytes` or

:func:`pickle.loads` executes arbitrary Python on the

receiver. Only accept payloads from trusted sources.

.. warning:: Portability

cloudpickle serializes Python bytecode, which is **not

stable across Python minor versions**. A payload produced

on Python 3.11 will fail to load on Python 3.12. The

wire format stamps the sender's ``(major, minor)``;

:meth:`from_bytes` raises a :class:`ValueError` naming

both versions on mismatch.

cloudpickle captures the UDF callable **by value** —

bytecode and closure cells inlined — but names the

callable resolves via ``import`` are captured **by

reference** (module path only) and must be importable on

the receiver.

**Self-contained — works anywhere:**

.. code-block:: python

# Lambda: bytecode captured inline

udf(lambda x: x * 2, [pa.int64()], pa.int64(),

volatility="immutable")

# Locally-defined function: bytecode captured inline

def double(x):

return x * 2

udf(double, [pa.int64()], pa.int64(), volatility="immutable")

# Closure over a local variable: value captured inline

factor = 3

udf(lambda x: x * factor, [pa.int64()], pa.int64(),

volatility="immutable")

**Requires matching environment on receiver:**

.. code-block:: python

# Top-level import: `foo` must be installed on receiver

from foo import double

udf(double, [pa.int64()], pa.int64(), volatility="immutable")

# Bound method of an imported class: same caveat

from mylib import Transformer

t = Transformer()

udf(t.transform, [pa.int64()], pa.int64(),

volatility="immutable")

Examples:

>>> from datafusion import col, lit

>>> blob = (col("a") + lit(1)).to_bytes()

>>> isinstance(blob, bytes)

True

"""

ctx_arg = ctx.ctx if ctx is not None else None

return self.expr.to_bytes(ctx_arg)

@classmethod

def from_bytes(cls, buf: bytes, ctx: SessionContext | None = None) -> Expr:

"""Reconstruct an expression from serialized bytes.

Accepts output of :meth:`to_bytes` or :func:`pickle.dumps`.

``ctx`` is the :class:`SessionContext` used to resolve any

function references that travel by name (e.g. FFI UDFs, or

Python UDFs sent with inlining disabled via

:meth:`SessionContext.with_python_udf_inlining`). When

``ctx`` is ``None`` the worker context installed via

:func:`datafusion.ipc.set_worker_ctx` is consulted; if no worker

context is installed, the global :class:`SessionContext` is used

(sufficient for built-ins and Python UDFs, plus any UDFs

registered on the global context).

.. warning:: Security

Decoding may invoke ``cloudpickle.loads`` on bytes embedded

in the payload, which executes arbitrary Python code. Treat

``buf`` as code, not data — only decode bytes you produced

yourself or received from a trusted sender.

.. warning:: Portability

cloudpickle payloads are **not portable across Python

minor versions**. The wire format stamps the sender's

``(major, minor)``; if it does not match the current

interpreter, this method raises :class:`ValueError`

naming both versions. Modules the UDF imports must also

be importable on the receiver — see :meth:`to_bytes` for

by-value vs. by-reference details.

Examples:

>>> from datafusion import Expr, col, lit

>>> blob = (col("a") + lit(1)).to_bytes()

>>> Expr.from_bytes(blob).canonical_name()

'a + Int64(1)'

"""

from datafusion.ipc import _resolve_ctx

resolved = _resolve_ctx(ctx)

return cls(expr_internal.RawExpr.from_bytes(resolved.ctx, buf))

def __reduce__(self) -> tuple[Callable[[bytes], Expr], tuple[bytes]]:

"""Pickle protocol hook.

Lets expressions be shipped to worker processes via

:func:`pickle.dumps` / :func:`pickle.loads`. Built-in functions

and Python UDFs (scalar, aggregate, window) travel inside the

pickle bytes; only FFI-capsule UDFs require pre-registration on

the worker. The worker's :class:`SessionContext` for resolving

those references is looked up via

:func:`datafusion.ipc.set_worker_ctx`, falling back to the

global :class:`SessionContext` if none has been installed on

the worker.

.. warning:: Security

:func:`pickle.loads` on the returned tuple executes

arbitrary Python on the receiver, including any

cloudpickled UDF callable embedded in the payload. Only

unpickle expressions from trusted sources.

.. warning:: Portability

Sender and receiver must run the same Python

``(major, minor)`` version; cloudpickle bytecode is not

portable across minor versions. See :meth:`to_bytes` for

details on what travels by value vs. by reference.

Examples:

>>> import pickle

>>> from datafusion import col, lit

>>> e = col("a") * lit(2)

>>> pickle.loads(pickle.dumps(e)).canonical_name()

'a * Int64(2)'

The encoding side honors a driver-side sender context installed

via :func:`datafusion.ipc.set_sender_ctx` — that is how

:meth:`SessionContext.with_python_udf_inlining` propagates

through ``pickle.dumps``. The sender context is read by

``__reduce__``, so :func:`copy.copy` and :func:`copy.deepcopy`

— which also go through ``__reduce__`` — pick it up too.

"""

return (Expr._reconstruct, (self.to_bytes(get_sender_ctx()),))

@classmethod

def _reconstruct(cls, proto_bytes: bytes) -> Expr:

"""Internal entry point used by :meth:`__reduce__` on unpickle.

Examples:

>>> from datafusion import Expr, col, lit

>>> blob = (col("a") + lit(1)).to_bytes()

>>> Expr._reconstruct(blob).canonical_name()

'a + Int64(1)'

"""

return cls.from_bytes(proto_bytes)

def __richcmp__(self, other: Expr, op: int) -> Expr:

"""Comparison operator."""

return Expr(self.expr.__richcmp__(other.expr, op))

def __repr__(self) -> str:

"""Generate a string representation of this expression."""

return self.expr.__repr__()

def __add__(self, rhs: Any) -> Expr:

"""Addition operator.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__add__(rhs.expr))

def __sub__(self, rhs: Any) -> Expr:

"""Subtraction operator.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__sub__(rhs.expr))

def __truediv__(self, rhs: Any) -> Expr:

"""Division operator.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__truediv__(rhs.expr))

def __mul__(self, rhs: Any) -> Expr:

"""Multiplication operator.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__mul__(rhs.expr))

def __mod__(self, rhs: Any) -> Expr:

"""Modulo operator (%).

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__mod__(rhs.expr))

def __and__(self, rhs: Expr) -> Expr:

"""Logical AND."""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__and__(rhs.expr))

def __or__(self, rhs: Expr) -> Expr:

"""Logical OR."""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__or__(rhs.expr))

def __invert__(self) -> Expr:

"""Binary not (~)."""

return Expr(self.expr.__invert__())

def __getitem__(self, key: str | int) -> Expr:

"""Retrieve sub-object.

If ``key`` is a string, returns the subfield of the struct.

If ``key`` is an integer, retrieves the element in the array. Note that the

element index begins at ``0``, unlike

:py:func:`~datafusion.functions.array_element` which begins at ``1``.

If ``key`` is a slice, returns an array that contains a slice of the

original array. Similar to integer indexing, this follows Python convention

where the index begins at ``0`` unlike

:py:func:`~datafusion.functions.array_slice` which begins at ``1``.

"""

if isinstance(key, int):

return Expr(

functions_internal.array_element(self.expr, Expr.literal(key + 1).expr)

)

if isinstance(key, slice):

if isinstance(key.start, int):

start = Expr.literal(key.start + 1).expr

elif isinstance(key.start, Expr):

start = (key.start + Expr.literal(1)).expr

else:

# Default start at the first element, index 1

start = Expr.literal(1).expr

if isinstance(key.stop, int):

stop = Expr.literal(key.stop).expr

else:

stop = key.stop.expr

if isinstance(key.step, int):

step = Expr.literal(key.step).expr

elif isinstance(key.step, Expr):

step = key.step.expr

else:

step = key.step

return Expr(functions_internal.array_slice(self.expr, start, stop, step))

return Expr(self.expr.__getitem__(key))

def __eq__(self, rhs: object) -> Expr:

"""Equal to.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if rhs is None:

return self.is_null()

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__eq__(rhs.expr))

def __ne__(self, rhs: object) -> Expr:

"""Not equal to.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if rhs is None:

return self.is_not_null()

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__ne__(rhs.expr))

def __ge__(self, rhs: Any) -> Expr:

"""Greater than or equal to.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__ge__(rhs.expr))

def __gt__(self, rhs: Any) -> Expr:

"""Greater than.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__gt__(rhs.expr))

def __le__(self, rhs: Any) -> Expr:

"""Less than or equal to.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__le__(rhs.expr))

def __lt__(self, rhs: Any) -> Expr:

"""Less than.

Accepts either an expression or any valid PyArrow scalar literal value.

"""

if not isinstance(rhs, Expr):

rhs = Expr.literal(rhs)

return Expr(self.expr.__lt__(rhs.expr))

__radd__ = __add__

__rand__ = __and__

__rmod__ = __mod__

__rmul__ = __mul__

__ror__ = __or__

__rsub__ = __sub__

__rtruediv__ = __truediv__

@staticmethod

def literal(value: Any) -> Expr:

"""Creates a new expression representing a scalar value.

``value`` must be a valid PyArrow scalar value or easily castable to one.

"""

if isinstance(value, str):

value = pa.scalar(value, type=pa.string_view())

return Expr(expr_internal.RawExpr.literal(value))

@staticmethod

def literal_with_metadata(value: Any, metadata: dict[str, str]) -> Expr:

"""Creates a new expression representing a scalar value with metadata.

Args:

value: A valid PyArrow scalar value or easily castable to one.

metadata: Metadata to attach to the expression.

"""

if isinstance(value, str):

value = pa.scalar(value, type=pa.string_view())

return Expr(expr_internal.RawExpr.literal_with_metadata(value, metadata))

@staticmethod

def string_literal(value: str) -> Expr:

"""Creates a new expression representing a UTF8 literal value.

It is different from `literal` because it is pa.string() instead of

pa.string_view()

This is needed for cases where DataFusion is expecting a UTF8 instead of

UTF8View literal, like in:

https://github.com/apache/datafusion/blob/86740bfd3d9831d6b7c1d0e1bf4a21d91598a0ac/datafusion/functions/src/core/arrow_cast.rs#L179

"""

if isinstance(value, str):

value = pa.scalar(value, type=pa.string())

return Expr(expr_internal.RawExpr.literal(value))

return Expr.literal(value)

@staticmethod

def column(value: str) -> Expr:

"""Creates a new expression representing a column."""

return Expr(expr_internal.RawExpr.column(value))

def alias(self, name: str, metadata: dict[str, str] | None = None) -> Expr:

"""Assign a name to the expression.

Args:

name: The name to assign to the expression.

metadata: Optional metadata to attach to the expression.

Returns:

A new expression with the assigned name.

"""

return Expr(self.expr.alias(name, metadata))

def sort(self, ascending: bool = True, nulls_first: bool = True) -> SortExpr:

"""Creates a sort :py:class:`Expr` from an existing :py:class:`Expr`.

Args:

ascending: If true, sort in ascending order.

nulls_first: Return null values first.

"""

return SortExpr(self, ascending=ascending, nulls_first=nulls_first)

def is_null(self) -> Expr:

"""Returns ``True`` if this expression is null."""

return Expr(self.expr.is_null())

def is_not_null(self) -> Expr:

"""Returns ``True`` if this expression is not null."""

return Expr(self.expr.is_not_null())

def fill_nan(self, value: Any | Expr | None = None) -> Expr:

"""Fill NaN values with a provided value."""

if not isinstance(value, Expr):

value = Expr.literal(value)

return Expr(functions_internal.nanvl(self.expr, value.expr))

def fill_null(self, value: Any | Expr | None = None) -> Expr:

"""Fill NULL values with a provided value."""

if not isinstance(value, Expr):

value = Expr.literal(value)

return Expr(functions_internal.nvl(self.expr, value.expr))

_to_pyarrow_types: ClassVar[dict[type, pa.DataType]] = {

float: pa.float64(),

int: pa.int64(),

str: pa.string(),

bool: pa.bool_(),

}

def cast(self, to: pa.DataType[Any] | type) -> Expr:

"""Cast to a new data type."""

if not isinstance(to, pa.DataType):

try:

to = self._to_pyarrow_types[to]

except KeyError as err:

error_msg = "Expected instance of pyarrow.DataType or builtins.type"

raise TypeError(error_msg) from err

return Expr(self.expr.cast(to))

def try_cast(self, to: pa.DataType[Any] | type) -> Expr:

"""Cast to a new data type, returning NULL on failure.

Like :py:meth:`cast` but produces NULL instead of erroring when the

cast cannot be performed for a given row.

Examples:

>>> ctx = dfn.SessionContext()

>>> df = ctx.from_pydict({"a": ["oops"]})

>>> result = df.select(col("a").try_cast(pa.float64()).alias("c"))

>>> result.collect_column("c")[0].as_py() is None

True

"""

if not isinstance(to, pa.DataType):

try:

to = self._to_pyarrow_types[to]

except KeyError as err:

error_msg = "Expected instance of pyarrow.DataType or builtins.type"

raise TypeError(error_msg) from err

return Expr(self.expr.try_cast(to))

def between(self, low: Any, high: Any, negated: bool = False) -> Expr:

"""Returns ``True`` if this expression is between a given range.

Args:

low: lower bound of the range (inclusive).

high: higher bound of the range (inclusive).

negated: negates whether the expression is between a given range

"""

if not isinstance(low, Expr):

low = Expr.literal(low)

if not isinstance(high, Expr):

high = Expr.literal(high)

return Expr(self.expr.between(low.expr, high.expr, negated=negated))

def rex_type(self) -> RexType:

"""Return the Rex Type of this expression.

A Rex (Row Expression) specifies a single row of data.That specification

could include user defined functions or types. RexType identifies the

row as one of the possible valid ``RexType``.

"""

return self.expr.rex_type()

def types(self) -> DataTypeMap:

"""Return the ``DataTypeMap``.

Returns:

DataTypeMap which represents the PythonType, Arrow DataType, and

SqlType Enum which this expression represents.

"""

return self.expr.types()

def python_value(self) -> Any:

"""Extracts the Expr value into `Any`.

This is only valid for literal expressions.

Returns:

Python object representing literal value of the expression.

"""

return self.expr.python_value()