o `۷i@sddlmZddlZddlmZmZddlmZmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZmZmZddlZddlmZddlmZddlmZdd lm Z m!Z!e rxdd l"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z)ddl*m+Z+edZ,edZ-ee-ee,fZ.e ddGddde Z/Gdddeee,Z0Gddde0dZ1e ddeddGdddeZ2e ddeddddGd d!d!e2Z3e ddeddddGd"d#d#e2Z4e ddeddddGd$d%d%e2Z5e ddeddddGd&d'd'e2Z6eddGd(d)d)Z7Gd*d+d+Z8e ddeddddGd,d-d-e2Z9d\d4d5Z:e!ddd]d6d7Z;d^d8d9Ze ddeddddGdBdCdCe2Z?e ddeddddGdDdEdEe2Z@e ddeddddGdFdGdGe2ZAe!dHddadKdLZBe!dHddbdOdPZCe dddcdQdRZDe!ddddSdddWdXZEgdYZFdedZd[ZGdS)f) annotationsN)ABCabstractmethod) dataclassfield)Enum) TYPE_CHECKINGAnyCallableDictGenericListOptionalTupleTypeTypeVarUnion) BatchColumn)DataType) DeveloperAPI PublicAPI_ArrayNamespace_DatetimeNamespace_ListNamespace_StringNamespace_StructNamespaceT).UDFExpralpha) stabilityc@s\eZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZdZdZdS) OperationaEnumeration of supported operations in expressions. This enum defines all the binary operations that can be performed between expressions, including arithmetic, comparison, and boolean operations. Attributes: ADD: Addition operation (+) SUB: Subtraction operation (-) MUL: Multiplication operation (*) DIV: Division operation (/) FLOORDIV: Floor division operation (//) GT: Greater than comparison (>) LT: Less than comparison (<) GE: Greater than or equal comparison (>=) LE: Less than or equal comparison (<=) EQ: Equality comparison (==) NE: Not equal comparison (!=) AND: Logical AND operation (&) OR: Logical OR operation (|) NOT: Logical NOT operation (~) IS_NULL: Check if value is null IS_NOT_NULL: Check if value is not null IN: Check if value is in a list NOT_IN: Check if value is not in a list addsubmuldivmodfloordivgtltgeleeqneandornotis_null is_not_nullinnot_inN)__name__ __module__ __qualname____doc__ADDSUBMULDIVMODFLOORDIVGTLTGELEEQNEANDORNOTIS_NULL IS_NOT_NULLINNOT_INrPrPJ/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/data/expressions.pyr%)s*r%c@seZdZdZd!ddZed"d d Zed#d d Zed$ddZed%ddZ ed&ddZ ed'ddZ ed(ddZ ed)ddZ d S)* _ExprVisitorzd?Zdd@dAZ ddBdCZ!ddDdEZ"ddHdIZ#ddJdKZ$ddMdNZ%ddPdQZ&ddRdSZ'ddTdUZ(ddVdWZ)ddXdYZ*ddZd[Z+dd\d]Z,dd^d_Z-dd`daZ.ddbdcZ/ddddeZ0ddfdgZ1ddhdiZ2ddjdkZ3ddldmZ4ddndoZ5ddqdrZ6ddsdtZ7eddvdwZ8eddydzZ9edd|d}Z:edddZ;edddZ>> from ray.data.expressions import col, lit >>> # Create an expression tree: (col("x") + 5) * col("y") >>> expr = (col("x") + lit(5)) * col("y") >>> # This creates a BinaryExpr with operation=MUL >>> # left=BinaryExpr(op=ADD, left=ColumnExpr("x"), right=LiteralExpr(5)) >>> # right=ColumnExpr("y") Note: This class should not be instantiated directly. Use the concrete subclasses like ColumnExpr, LiteralExpr, etc. r data_typerU str | NonecCsdS)zGet the name associated with this expression. Returns: The name for expressions that have one (ColumnExpr, AliasExpr), None otherwise. NrPrirPrPrQrzsz Expr.nameotherr boolcCst)z4Compare two expression ASTs for structural equality.)NotImplementedErrorrirrPrPrQstructurally_equalsszExpr.structurally_equalsrwcCs t|S)adConvert this Ray Data expression to a PyArrow compute expression. Returns: A PyArrow compute expression equivalent to this Ray Data expression. Raises: ValueError: If the expression contains operations not supported by PyArrow. TypeError: If the expression type cannot be converted to PyArrow. )rvrjrrPrPrQ to_pyarrows zExpr.to_pyarrowstrcCsddlm}||S)u8Return a tree-structured string representation of the expression. Returns: A multi-line string showing the expression tree structure using box-drawing characters for visual clarity. Example: >>> from ray.data.expressions import col, lit >>> expr = (col("x") + lit(5)) * col("y") >>> print(expr) MUL ├── left: ADD │ ├── left: COL('x') │ └── right: LIT(5) └── right: COL('y') r)_TreeReprVisitor)>ray.data._internal.planner.plan_expression.expression_visitorsrrj)rirrPrPrQ__repr__s  z Expr.__repr__rr%rTcCst|ts t|}t|||S)aiCreate a binary expression with the given operation. Args: other: The right operand expression or literal value op: The operation to perform Returns: A new BinaryExpr representing the operation Note: If other is not an Expr, it will be automatically converted to a LiteralExpr. )rVrrYr[)rirrrPrPrQ_bins  z Expr._bincC||tjS)zAddition operator (+).)rr%r=rrPrPrQ__add__4z Expr.__add__cCt||tjS)z/Reverse addition operator (for literal + expr).)rYrr%r=rrPrPrQ__radd__8z Expr.__radd__cCr)zSubtraction operator (-).)rr%r>rrPrPrQ__sub__<rz Expr.__sub__cCr)z2Reverse subtraction operator (for literal - expr).)rYrr%r>rrPrPrQ__rsub__@rz Expr.__rsub__cCr)zMultiplication operator (*).)rr%r?rrPrPrQ__mul__Drz Expr.__mul__cCr)z5Reverse multiplication operator (for literal * expr).)rYrr%r?rrPrPrQ__rmul__Hrz Expr.__rmul__cCrzModulation operator (%).)rr%rArrPrPrQ__mod__Lrz Expr.__mod__cCrr)rYrr%rArrPrPrQ__rmod__Prz Expr.__rmod__cCr)zDivision operator (/).)rr%r@rrPrPrQ __truediv__TrzExpr.__truediv__cCr)z/Reverse division operator (for literal / expr).)rYrr%r@rrPrPrQ __rtruediv__XrzExpr.__rtruediv__cCr)zFloor division operator (//).)rr%rBrrPrPrQ __floordiv__\rzExpr.__floordiv__cCr)z6Reverse floor division operator (for literal // expr).)rYrr%rBrrPrPrQ __rfloordiv__`rzExpr.__rfloordiv__cCr)zGreater than operator (>).)rr%rCrrPrPrQ__gt__erz Expr.__gt__cCr)zLess than operator (<).)rr%rDrrPrPrQ__lt__irz Expr.__lt__cCr)z$Greater than or equal operator (>=).)rr%rErrPrPrQ__ge__mrz Expr.__ge__cCr)z!Less than or equal operator (<=).)rr%rFrrPrPrQ__le__qrz Expr.__le__cCr)zEquality operator (==).)rr%rGrrPrPrQ__eq__urz Expr.__eq__cCr)zNot equal operator (!=).)rr%rHrrPrPrQ__ne__yrz Expr.__ne__cCr)zLogical AND operator (&).)rr%rIrrPrPrQ__and__~rz Expr.__and__cCr)zLogical OR operator (|).)rr%rJrrPrPrQ__or__rz Expr.__or__cC ttj|S)zLogical NOT operator (~).)r]r%rKrrPrPrQ __invert__ zExpr.__invert__cCr)z&Check if the expression value is null.)r]r%rLrrPrPrQr5rz Expr.is_nullcCr)z*Check if the expression value is not null.)r]r%rMrrPrPrQr6rzExpr.is_not_nullvaluesUnion[List[Any], 'Expr']cC t|ts t|}||tjS)z5Check if the expression value is in a list of values.)rVrrYrr%rNrirrPrPrQr z Expr.is_incCr)z9Check if the expression value is not in a list of values.)rVrrYrr%rOrrPrPrQr8rz Expr.not_inrzcCt|j||ddS)aRename the expression. This method allows you to assign a new name to an expression result. This is particularly useful when you want to specify the output column name directly within the expression rather than as a separate parameter. Args: name: The new name for the expression Returns: An AliasExpr that wraps this expression with the specified name Example: >>> from ray.data.expressions import col, lit >>> # Create an expression with a new aliased name >>> expr = (col("price") * col("quantity")).alias("total") >>> # Can be used with Dataset operations that support named expressions F)rrS_name _is_renamer_rrirzrPrPrQaliass z Expr.aliasrscCttj|S)z'Round values up to the nearest integer.)_create_pyarrow_compute_udfryceilrrPrPrQrrz Expr.ceilcCr)z)Round values down to the nearest integer.)rryfloorrrPrPrQrrz Expr.floorcCr)z>> from ray.data.expressions import col >>> import ray >>> ds = ray.data.from_items([{"x": 5}, {"x": -3}]) >>> ds = ds.with_column("neg_x", col("x").negate()) >>> # Result: neg_x = [-5, 3] )rrynegate_checkedrrPrPrQnegate z Expr.negatecCr)aCompute the sign of the expression. Returns: A UDFExpr that returns -1 for negative values, 0 for zero, and 1 for positive values. Example: >>> from ray.data.expressions import col >>> import ray >>> ds = ray.data.from_items([{"x": 5}, {"x": -3}, {"x": 0}]) >>> ds = ds.with_column("sign_x", col("x").sign()) >>> # Result: sign_x = [1, -1, 0] )rrysignrrPrPrQrrz Expr.signexponentcCsttj||S)aORaise the expression to the given power. Args: exponent: The exponent to raise the expression to. Returns: A UDFExpr that computes the power operation. Example: >>> from ray.data.expressions import col, lit >>> import ray >>> ds = ray.data.from_items([{"x": 2}, {"x": 3}]) >>> ds = ds.with_column("x_squared", col("x").power(2)) >>> # Result: x_squared = [4, 9] >>> ds = ds.with_column("x_cubed", col("x").power(3)) >>> # Result: x_cubed = [8, 27] )rrypower)rirrPrPrQr"sz Expr.powercCr)aCompute the absolute value of the expression. Returns: A UDFExpr that computes the absolute value. Example: >>> from ray.data.expressions import col >>> import ray >>> ds = ray.data.from_items([{"x": 5}, {"x": -3}]) >>> ds = ds.with_column("abs_x", col("x").abs()) >>> # Result: abs_x = [5, 3] )rry abs_checkedrrPrPrQabs6rzExpr.abs'_ArrayNamespace'cCddlm}||S)z,Access array operations for this expression.rr),ray.data.namespace_expressions.arr_namespacer)rirrPrPrQarrE zExpr.arr'_ListNamespace'cCr)aAccess list operations for this expression. Returns: A _ListNamespace that provides list-specific operations for both PyArrow ``List`` and ``FixedSizeList`` columns. Example: >>> from ray.data.expressions import col >>> import ray >>> ds = ray.data.from_items([ ... {"items": [1, 2, 3]}, ... {"items": [4, 5]} ... ]) >>> ds = ds.with_column("num_items", col("items").list.len()) >>> ds = ds.with_column("first_item", col("items").list[0]) >>> ds = ds.with_column("slice", col("items").list[1:3]) rr)-ray.data.namespace_expressions.list_namespacer)rirrPrPrQrLs z Expr.list'_StringNamespace'cCr)a`Access string operations for this expression. Returns: A _StringNamespace that provides string-specific operations. Example: >>> from ray.data.expressions import col >>> import ray >>> ds = ray.data.from_items([ ... {"name": "Alice"}, ... {"name": "Bob"} ... ]) >>> ds = ds.with_column("upper_name", col("name").str.upper()) >>> ds = ds.with_column("name_len", col("name").str.len()) >>> ds = ds.with_column("starts_a", col("name").str.starts_with("A")) rr)/ray.data.namespace_expressions.string_namespacer)rirrPrPrQrcs zExpr.str'_StructNamespace'cCr)aAccess struct operations for this expression. Returns: A _StructNamespace that provides struct-specific operations. Example: >>> from ray.data.expressions import col >>> import ray >>> import pyarrow as pa >>> ds = ray.data.from_arrow(pa.table({ ... "user": pa.array([ ... {"name": "Alice", "age": 30} ... ], type=pa.struct([ ... pa.field("name", pa.string()), ... pa.field("age", pa.int32()) ... ])) ... })) >>> ds = ds.with_column("age", col("user").struct["age"]) # doctest: +SKIP rr)/ray.data.namespace_expressions.struct_namespacer )rir rPrPrQstructys z Expr.struct'_DatetimeNamespace'cCr)z/Access datetime operations for this expression.rr)+ray.data.namespace_expressions.dt_namespacer)rirrPrPrQdtrzExpr.dtcCs|SrmrPrrPrPrQ_unaliassz Expr._unaliasN)rUrrr rUr)rUrwrUr)rr rr%rUrT)rr rUrT)rr rUrT)rrrUrTrzrrUrT)rUrs)rr rUrs)rUr)rUr)rUr)rUr)rUr)>r9r:r;r<__annotations__propertyrzrrrrrrrrrrrrrrrrrrrrrrrrrrr5r6rr8rrrrrrrrrrrrrrrrrrrrrrrrrrPrPrPrQrs                                                      rF)rr0reprc@sTeZdZUdZded<eddddZded <edd d ZdddZ dddZ dS)rWaExpression that references a column by name. This expression type represents a reference to an existing column in the dataset. When evaluated, it returns the values from the specified column. Args: name: The name of the column to reference Example: >>> from ray.data.expressions import col >>> # Reference the "age" column >>> age_expr = col("age") # Creates ColumnExpr(name="age") rrcCttSrmrobjectrPrPrPrQzColumnExpr.Fdefault_factoryinitrrrUcC|jS)zGet the column name.rrrPrPrQrzzColumnExpr.namerzcCr)NT)rrrrPrPrQ_renameszColumnExpr._renamerr rcCt|to |j|jkSrm)rVrWrzrrPrPrQrszColumnExpr.structurally_equalsNrrzrr) r9r:r;r<rrrrrzr rrPrPrPrQrWs   rWc@s>eZdZUdZded<eddZded<dd Zdd dZdS)rYaQExpression that represents a constant scalar value. This expression type represents a literal value that will be broadcast to all rows when evaluated. The value can be any Python object. Args: value: The constant value to represent Example: >>> from ray.data.expressions import lit >>> import numpy as np >>> # Create a literal value >>> five = lit(5) # Creates LiteralExpr(value=5) >>> name = lit("John") # Creates LiteralExpr(value="John") >>> numpy_val = lit(np.int32(42)) # Creates LiteralExpr with numpy type r r}F)rrrcCst|j}t|d|dS)Nr)r infer_dtyper}r __setattr__)riinferred_dtyperPrPrQ __post_init__s zLiteralExpr.__post_init__rrUrcCs*t|to|j|jkot|jt|juSrm)rVrYr}rgrrPrPrQrs  zLiteralExpr.structurally_equalsNr) r9r:r;r<rrrrrrPrPrPrQrYs rYc@sLeZdZUdZded<ded<ded<eddd d Zd ed <dddZdS)r[aExpression that represents a binary operation between two expressions. This expression type represents an operation with two operands (left and right). The operation is specified by the `op` field, which must be one of the supported operations from the Operation enum. Args: op: The operation to perform (from Operation enum) left: The left operand expression right: The right operand expression Example: >>> from ray.data.expressions import col, lit, Operation >>> # Manually create a binary expression (usually done via operators) >>> expr = BinaryExpr(Operation.ADD, col("x"), lit(5)) >>> # This is equivalent to: col("x") + lit(5) r%rrrrcCrrmrrPrPrPrQrrzBinaryExpr.Frrrrr rUrcCs2t|to|j|juo|j|jo|j|jSrm)rVr[rrrrrrPrPrQrs    zBinaryExpr.structurally_equalsNrr9r:r;r<rrrrrPrPrPrQr[s r[c@sDeZdZUdZded<ded<edddd Zd ed <dddZdS)r]agExpression that represents a unary operation on a single expression. This expression type represents an operation with one operand. Common unary operations include logical NOT, IS NULL, IS NOT NULL, etc. Args: op: The operation to perform (from Operation enum) operand: The operand expression Example: >>> from ray.data.expressions import col >>> # Check if a column is null >>> expr = col("age").is_null() # Creates UnaryExpr(IS_NULL, col("age")) >>> # Logical not >>> expr = ~(col("active")) # Creates UnaryExpr(NOT, col("active")) r%rrrcCtSrm)rrrPrPrPrQr"rzUnaryExpr.Frrrrr rUrcCs$t|to|j|juo|j|jSrm)rVr]rrrrrPrPrQr$s   zUnaryExpr.structurally_equalsNrrrPrPrPrQr]s r]c@s`eZdZUdZded<dZded<eedZded <ed d d d Z d ed<ddZ dddZ d S)_CallableClassSpecaSpecification for a callable class UDF. This dataclass captures the class type and constructor arguments needed to instantiate a callable class UDF on an actor. It consolidates the callable class metadata that was previously spread across multiple fields. Attributes: cls: The original callable class type args: Positional arguments for the constructor kwargs: Keyword arguments for the constructor _cached_key: Pre-computed key that survives serialization rgclsrPTuple[Any, ...]args)rDict[str, Any]kwargsNF)defaultcomparerzOptional[Tuple] _cached_keycCs|jdurA|jjd|jj}z||jtt|jf}t |Wnt y7|t |jt |jf}Ynwt |d|dSdS)zPre-compute and cache the key at construction time. This ensures the same key survives serialization, since the cached key tuple (containing the already-computed repr strings) gets pickled and unpickled as-is. Nr~r)rrr:r;rtuplesortedritemshashrfrrr )riclass_idkeyrPrPrQr@s   z _CallableClassSpec.__post_init__rUrcCr)aReturn the pre-computed hashable key for UDF instance lookup. The key uniquely identifies a UDF by its class and constructor arguments. This ensures that the same class with different constructor args (e.g., Multiplier(2) vs Multiplier(3)) are treated as distinct UDFs. Returns: A hashable tuple that uniquely identifies this UDF configuration. )rrrPrPrQmake_keyWs z_CallableClassSpec.make_key)rUr) r9r:r;r<rrrdictrrrr!rPrPrPrQr,s  rc@sJeZdZdZdd d ZedddZedddZdddZd ddZdS)!_CallableClassUDFaA wrapper that makes callable class UDFs appear as regular functions. This class wraps callable class UDFs for use in expressions. It provides an `init()` method that should be called at actor startup via `init_fn` to instantiate the underlying class before any blocks are processed. Key responsibilities: 1. Store the callable class and constructor arguments 2. Provide init() for actor startup initialization 3. Handle async bridging for coroutine/async generator UDFs 4. Reuse the same instance across all calls (actor semantics) Example: >>> @udf(return_dtype=DataType.int32()) ... class AddOffset: ... def __init__(self, offset=1): ... self.offset = offset ... def __call__(self, x): ... return pc.add(x, self.offset) >>> >>> add_five = AddOffset(5) # Creates _CallableClassUDF internally >>> expr = add_five(col("value")) # Creates UDFExpr with fn=_CallableClassUDF rrg ctor_argsr ctor_kwargsrrrcCs2||_||_||_||_d|_t|||d|_dS)a!Initialize the _CallableClassUDF wrapper. Args: cls: The original callable class ctor_args: Constructor positional arguments ctor_kwargs: Constructor keyword arguments return_dtype: The return data type for schema inference N)rrr)_cls _ctor_args _ctor_kwargs _return_dtype _instancer_callable_class_spec)rirr$r%rrPrPrQ__init__}s z_CallableClassUDF.__init__rUrcCs|jjS)z2Return the original class name for error messages.)r&r9rrPrPrQr9sz_CallableClassUDF.__name__rcCr)zReturn the callable class spec for this UDF. Used for deduplication when the same UDF appears multiple times in an expression tree. )r+rrPrPrQcallable_class_specsz%_CallableClassUDF.callable_class_specNonecCs(|jdur|j|ji|j|_dSdS)zInitialize the UDF instance. Called at actor startup via init_fn. This ensures the callable class is instantiated before any blocks are processed, matching the behavior of map_batches callable classes. N)r*r&r'r(rrPrPrQrs z_CallableClassUDF.initrr rcOsB|jdurtd|jjdddlm}||jg|Ri|S)aMCall the UDF instance. Args: *args: Evaluated expression arguments (PyArrow arrays, etc.) **kwargs: Evaluated expression keyword arguments Returns: The result of calling the UDF instance Raises: RuntimeError: If init() was not called before __call__ Nz_CallableClassUDF 'zr' was not initialized. init() must be called before __call__. This typically happens via init_fn at actor startup.r)$_call_udf_instance_with_async_bridge)r* RuntimeErrorr&r9ray.data.util.expression_utilsr/)rirrr/rPrPrQ__call__s  z_CallableClassUDF.__call__N)rrgr$rr%rrrr)rUr)rUr.)rr rr rUr ) r9r:r;r<r,rr-rr2rPrPrPrQr#ds     r#c@sBeZdZUdZded<ded<ded<edd d ZdddZdS)r"aExpression that represents a user-defined function call. This expression type wraps a UDF with schema inference capabilities, allowing UDFs to be used seamlessly within the expression system. UDFs operate on batches of data, where each column argument is passed as a PyArrow Array containing multiple values from that column across the batch. Args: fn: The user-defined function to call. For callable classes, this is an _CallableClassUDF instance that handles lazy instantiation internally. args: List of argument expressions (positional arguments) kwargs: Dictionary of keyword argument expressions Example: >>> from ray.data.expressions import col, udf >>> import pyarrow as pa >>> import pyarrow.compute as pc >>> from ray.data.datatype import DataType >>> >>> @udf(return_dtype=DataType.int32()) ... def add_one(x: pa.Array) -> pa.Array: ... return pc.add(x, 1) >>> >>> # Use in expressions >>> expr = add_one(col("value")) >>> # Callable class example >>> @udf(return_dtype=DataType.int32()) ... class AddOffset: ... def __init__(self, offset=1): ... self.offset = offset ... def __call__(self, x: pa.Array) -> pa.Array: ... return pc.add(x, self.offset) >>> >>> # Use callable class >>> add_five = AddOffset(5) >>> expr = add_five(col("value")) Callable[..., BatchColumn]fnz List[Expr]rzDict[str, Expr]rrUOptional[_CallableClassSpec]cCst|jtr |jjSdS)zReturn callable_class_spec if fn is an _CallableClassUDF, else None. This property maintains backward compatibility with code that checks for callable_class_spec. N)rVr4r#r-rrPrPrQr-s zUDFExpr.callable_class_specrr rcsttsdStjtr tjtsdSjjjjkrdSnjjkr(dStjtjkoXtddtjjDoXj j koXtfddj DS)NFcss|] \}}||VqdSrm)r).0abrPrPrQ sz.UDFExpr.structurally_equals..c3s&|]}j|j|VqdSrm)rr)r6krrirPrQr9s  ) rVr"r4r#r-lenrallziprkeysrrPr;rQrs&    zUDFExpr.structurally_equalsN)rUr5r)r9r:r;r<rrr-rrPrPrPrQr"s (  r"r4r3rrrUCallable[..., UDFExpr]cs&dfdd }t||_|S)awCreate a callable that generates UDFExpr when called with expressions. Args: fn: The user-defined function to wrap. Can be a regular function or an _CallableClassUDF instance (for callable classes). return_dtype: The return data type of the UDF Returns: A callable that creates UDFExpr instances when called with expressions rUr"csxg}|D]}t|tr||q|t|qi}|D]\}}t|tr-|||<qt|||<qt||dS)N)r4rrr)rVrappendrYrr")rr expr_argsarg expr_kwargsr:vr4rrPrQ udf_callable-s     z*_create_udf_callable..udf_callableN)rUr") functoolsupdate_wrapper _original_fn)r4rrGrPrFrQ_create_udf_callables rKcdfdd }|S) a Decorator to convert a UDF into an expression-compatible function. This decorator allows UDFs to be used seamlessly within the expression system, enabling schema inference and integration with other expressions. IMPORTANT: UDFs operate on batches of data, not individual rows. When your UDF is called, each column argument will be passed as a PyArrow Array containing multiple values from that column across the batch. Under the hood, when working with multiple columns, they get translated to PyArrow arrays (one array per column). Args: return_dtype: The data type of the return value of the UDF Returns: A callable that creates UDFExpr instances when called with expressions Example: >>> from ray.data.expressions import col, udf >>> import pyarrow as pa >>> import pyarrow.compute as pc >>> import ray >>> >>> # UDF that operates on a batch of values (PyArrow Array) >>> @udf(return_dtype=DataType.int32()) ... def add_one(x: pa.Array) -> pa.Array: ... return pc.add(x, 1) # Vectorized operation on the entire Array >>> >>> # UDF that combines multiple columns (each as a PyArrow Array) >>> @udf(return_dtype=DataType.string()) ... def format_name(first: pa.Array, last: pa.Array) -> pa.Array: ... return pc.binary_join_element_wise(first, last, " ") # Vectorized string concatenation >>> >>> # Callable class UDF >>> @udf(return_dtype=DataType.int32()) ... class AddOffset: ... def __init__(self, offset=1): ... self.offset = offset ... def __call__(self, x: pa.Array) -> pa.Array: ... return pc.add(x, self.offset) >>> >>> # Use in dataset operations >>> ds = ray.data.from_items([ ... {"value": 5, "first": "John", "last": "Doe"}, ... {"value": 10, "first": "Jane", "last": "Smith"} ... ]) >>> >>> # Single column transformation (operates on batches) >>> ds_incremented = ds.with_column("value_plus_one", add_one(col("value"))) >>> >>> # Multi-column transformation (each column becomes a PyArrow Array) >>> ds_formatted = ds.with_column("full_name", format_name(col("first"), col("last"))) >>> >>> # Callable class usage >>> add_five = AddOffset(5) >>> ds_with_offset = ds.with_column("value_plus_five", add_five(col("value"))) >>> >>> # Can also be used in complex expressions >>> ds_complex = ds.with_column("doubled_plus_one", add_one(col("value")) * 2) func_or_class*Union[Callable[..., BatchColumn], Type[T]]rU DecoratedcsNttr"ttr"Gfddd}j|_j|_j|_|StS)Ncs*eZdZdZfddZfddZdS)z.decorator..ExpressionAwareCallableClassaIntercepts callable class instantiation to delay until actor execution. Allows natural syntax like: add_five = AddOffset(5) ds.with_column("result", add_five(col("x"))) When instantiated, creates an _CallableClassUDF that is completely self-contained - it handles lazy instantiation and async bridging internally. From the planner's perspective, this is just a regular callable function. cst||d|_dS)N)rr$r%r)r# _expr_udf)rirrrMrrPrQr,s  zEudf..decorator..ExpressionAwareCallableClass.__init__cst|j|i|Srm)rKrP)ri call_args call_kwargsrrPrQr2szEudf..decorator..ExpressionAwareCallableClass.__call__N)r9r:r;r<r,r2rPrQrPrQExpressionAwareCallableClasss  rT)rVrg issubclassr r9r;r:rK)rMrTr)rMrQ decorators! zudf..decoratorN)rMrNrUrOrPrrVrPrrQudfMs?2rXcstfdd}|S)aWrap a PyArrow compute function to auto-convert inputs to PyArrow format. This wrapper ensures that pandas Series and numpy arrays are converted to PyArrow Arrays before being passed to the function, enabling PyArrow compute functions to work seamlessly with any block format. Args: fn: The PyArrow compute function to wrap Returns: A wrapped function that handles format conversion csddlddlddlfddfdd|D}fdd|D}dd|D}d d|D}td d |DpKtd d |D}|i|}|rbt|jjfrb| }|S) NrcsXt|jjfr |dfSt|jrj|dfSt|jr(|dfS|dfS)z+Convert a value to PyArrow Array if needed.FT)rVArray ChunkedArraySeries from_pandasndarrayr)val)nprpdrPrQto_arrows  z@_create_pyarrow_wrapper..arrow_wrapper..to_arrowcsg|]}|qSrPrP)r6rCrarPrQ zB_create_pyarrow_wrapper..arrow_wrapper..csi|] \}}||qSrPrPr6r:rErbrPrQ zB_create_pyarrow_wrapper..arrow_wrapper..cSsg|]}|dqSrrPr6rErPrPrQrcrdcSsi|] \}}||dqSrhrPrerPrPrQrfrgcss|]}|dVqdS)NrPrirPrPrQr9szA_create_pyarrow_wrapper..arrow_wrapper..) numpypandasrranyrrVrYrZ to_pandas)rr args_resultskwargs_resultsconverted_argsconverted_kwargsinput_was_pandasrr4)r_rr`rarQ arrow_wrappers z._create_pyarrow_wrapper..arrow_wrapper)rHwraps)r4rurPrtrQ_create_pyarrow_wrappers#rwcrL) aEDecorator for PyArrow compute functions with automatic format conversion. This decorator wraps PyArrow compute functions to automatically convert pandas Series and numpy arrays to PyArrow Arrays, ensuring the function works seamlessly regardless of the underlying block format (pandas, arrow, or items). Used internally by namespace methods (list, str, struct) that wrap PyArrow compute functions. Args: return_dtype: The data type of the return value Returns: A callable that creates UDFExpr instances with automatic conversion funcr3rUr@cst|}t|Srm)rwrK)rx wrapped_fnrrPrQrV s zpyarrow_udf..decoratorN)rxr3rUr@rPrWrPrrQ pyarrow_udfsrzpc_funcCallable[..., pyarrow.Array]DataType | NoneCallable[..., 'UDFExpr']csd fdd }|S) z>Create an expression UDF backed by a PyArrow compute function.rSrT positionalr rrUrscs*tp|jddfdd }||S)Nrr pyarrow.ArrayrUcs|gRiSrmrP)r)rr{rrPrQrXrz9_create_pyarrow_compute_udf..wrapper..udf)rrrUr)rzr)rSrrrXr{r)rrrQwrappersz,_create_pyarrow_compute_udf..wrapperN)rSrTrr rr rUrsrP)r{rrrPrrQrsrc@sHeZdZUdZded<dZded<eddd d Zd ed <dddZdS)rbz0Expression that represents a download operation.ruri_column_nameNz'pyarrow.fs.FileSystem' filesystemcCrrm)rbinaryrPrPrPrQr*rzDownloadExpr.Frrrrr rUrcCr rm)rVrbrrrPrPrQr,s  z DownloadExpr.structurally_equalsr) r9r:r;r<rrrrrrPrPrPrQrb#s  rbc@sVeZdZUdZded<ded<ded<edd d Zdd dZdddZdddZ dS)r_z6Expression that represents an alias for an expression.rrSrrrrrUcCr)zGet the alias name.rrrPrPrQrz<rzAliasExpr.namerzrTcCst|jj|j||jdS)N)rr)r_rSrrrrPrPrQrAszAliasExpr.aliascCrrm)rSrrPrPrQrGszAliasExpr._unaliasrr cCs0t|to|j|jo|j|jko|j|jkSrm)rVr_rSrrzrrrPrPrQrJs    zAliasExpr.structurally_equalsNrrrr) r9r:r;r<rrrzrrrrPrPrPrQr_3s    r_c@s4eZdZUdZeddddZded<dd d ZdS)rdaExpression that represents all columns from the input. This is a special expression used in projections to indicate that all existing columns should be preserved at this position in the output. It's typically used internally by operations like with_column() and rename_columns() to maintain existing columns. Example: When with_column("new_col", expr) is called, it creates: Project(exprs=[star(), expr.alias("new_col")]) This means: keep all existing columns, then add/overwrite "new_col" cCrrmrrPrPrPrQrerzStarExpr.Frrrrr rUrcCs t|tSrm)rVrdrrPrPrQrgs zStarExpr.structurally_equalsNr)r9r:r;r<rrrrrPrPrPrQrdSs rdbetarzrcCt|S)a Reference an existing column by name. This is the primary way to reference columns in expressions. The returned expression will extract values from the specified column when evaluated. Args: name: The name of the column to reference Returns: A ColumnExpr that references the specified column Example: >>> from ray.data.expressions import col >>> # Reference columns in an expression >>> expr = col("price") * col("quantity") >>> >>> # Use with Dataset.with_column() >>> import ray >>> ds = ray.data.from_items([{"price": 10, "quantity": 2}]) >>> ds = ds.with_column("total", col("price") * col("quantity")) )rW)rzrPrPrQcolksrr}r cCr)a Create a literal expression from a constant value. This creates an expression that represents a constant scalar value. The value will be broadcast to all rows when the expression is evaluated. Args: value: The constant value to represent. Can be any Python object (int, float, str, bool, etc.) Returns: A LiteralExpr containing the specified value Example: >>> from ray.data.expressions import col, lit >>> # Create literals of different types >>> five = lit(5) >>> pi = lit(3.14159) >>> name = lit("Alice") >>> flag = lit(True) >>> >>> # Use in expressions >>> expr = col("age") + lit(1) # Add 1 to age column >>> >>> # Use with Dataset.with_column() >>> import ray >>> ds = ray.data.from_items([{"age": 25}, {"age": 30}]) >>> ds = ds.with_column("age_plus_one", col("age") + lit(1)) )rY)r}rPrPrQlitsrcCstS)a9 References all input columns from the input. This is a special expression used in projections to preserve all existing columns. It's typically used with operations that want to add or modify columns while keeping the rest. Returns: A StarExpr that represents all input columns. )rdrPrPrPrQstars r)rrr!Optional['pyarrow.fs.FileSystem']cCs t||dS)a Create a download expression that downloads content from URIs. This creates an expression that will download bytes from URIs stored in a specified column. When evaluated, it will fetch the content from each URI and return the downloaded bytes. Args: uri_column_name: The name of the column containing URIs to download from filesystem: PyArrow filesystem to use for reading remote files. If None, the filesystem is auto-detected from the path scheme. Returns: A DownloadExpr that will download content from the specified URI column Example: >>> from ray.data.expressions import download >>> import ray >>> # Create dataset with URIs >>> ds = ray.data.from_items([ ... {"uri": "s3://bucket/file1.jpg", "id": "1"}, ... {"uri": "s3://bucket/file2.jpg", "id": "2"} ... ]) >>> # Add downloaded bytes column >>> ds_with_bytes = ds.with_column("bytes", download("uri")) rr)rbrrPrPrQdownloads r)r%rrWrYr[r]r"rbr_rdrzrXrrrrrrrr rcCs|dkr ddlm}|S|dkrddlm}|S|dkr$ddlm}|S|dkr0dd lm}|S|d krs  4        08@L  $ " "7f  R/  s7       " (