o $iAo@s8ddlZddlZddlZddlmZmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZmZddlZddlZddlZddlmZmZddlmZddlmZddl m!Z!e r|ddl"Z"ddlZdd l#m$Z$dd l%m&Z&dd l'm(Z(dd l)m*Z*ed ddZ+edddZ,edZ-edZ.edZ/ee0ddfZ1edZ2edZ3e4e5Z6e!GdddeZ7e!Gddde8eZ9edde e8ej:ffZ;ee2ej:fZee e+ge,fe e+ge e,fe0d fZ?eeee/d!fZ@ed!ZAgd"ZBd#ZCd$ee1d%eDfd&d'ZEd(e d)d%ed)fd*d+ZFd,ee8d%e8fd-d.ZGe!d/ed!d%ed0fd1d2ZHe!Gd3d4d4ZIGd5d6d6ZJe!eGd7d0d0ZKd8d9eeKDZLe!eGd:d!d!eKZMe!d;d<eGd=d>d>eMZNe!Gd?d@d@ZOe!dAd<GdBdCdCZPdDeQej:d%ej:fdEdFZRdS)GN) dataclassfields)Enum) TYPE_CHECKINGAnyCallableDictIteratorListOptionalProtocolTupleTypeVarUnion)_check_pyarrow_version_truncated_repr) ObjectRef)log_once) DeveloperAPI) BlockBuilderPandasBlockSchema)SortKey) AggregateFnTT) contravariantU) covariantKeyTypeAggType) pyarrow.Tablepandas.DataFramerpyarrow.lib.Schema)pyarrow.ChunkedArray pyarrow.Array pandas.Series)r%z np.ndarrayr$r#c@seZdZdZdZdS) BlockTypearrowpandasN)__name__ __module__ __qualname__ARROWPANDASr.r.K/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/block.pyr&Dsr&c@seZdZdZdZdZdS) BatchFormatpyarrowr(numpyN)r)r*r+r,r-NUMPYr.r.r.r/r0Jsr0r r!c@s*eZdZdedeeeeffddZdS)_CallableClassProtocol_CallableClassProtocol__argreturncCsdSNr.)selfr5r.r.r/__call__`sz_CallableClassProtocol.__call__N)r)r*r+rrrr r9r.r.r.r/r4_s"r4 BlockMetadata)r(r1r2Nr2schemar6cCs,ddlm}|dupt||r|j S| S)Nrr)ray.data._internal.pandas_blockr isinstancenames)r;rr.r.r/_is_empty_schemaws r?schemasSchemacCs|D] }t|s |SqdS)zReturn the first non-empty schema from an iterator of schemas. Args: schemas: Iterator of schemas to check. Returns: The first non-empty schema, or None if all schemas are empty. N)r?)r@r;r.r.r/_take_first_non_empty_schemas  rBgiven_batch_formatcCs.|dkrt}|tvrtd|dtd|S)NdefaultzThe given batch format z isn't allowed (must be one of z).)DEFAULT_BATCH_FORMATVALID_BATCH_FORMATS ValueError)rCr.r.r/_apply_batch_formatsrHmetas BlockStatscCsdd|DS)NcSsg|]}|qSr.)to_stats).0mr.r.r/ szto_stats..r.)rIr.r.r/rKsrKc@s.eZdZdZddZed ddZdd Zd S) BlockExecStatsaExecution stats for this block. Attributes: wall_time_s: The wall-clock time it took to compute this block. cpu_time_s: The CPU time it took to compute this block. node_id: A unique id for the node that computed this block. max_uss_bytes: An estimate of the maximum amount of physical memory that the process was using while computing this block. cCs>d|_d|_d|_d|_d|_tj|_ d|_ d|_ dS)Nr) start_time_s end_time_s wall_time_s udf_time_s cpu_time_srayruntime_contextget_runtime_context get_node_idnode_id max_uss_bytestask_idxr8r.r.r/__init__s zBlockExecStats.__init__r6_BlockExecStatsBuildercCtSr7)r^r.r.r.r/builderzBlockExecStats.buildercCst|j|j|j|jdS)N)rRrTrSrY)reprrRrTrSrYr\r.r.r/__repr__szBlockExecStats.__repr__N)r6r^)r)r*r+__doc__r] staticmethodr`rcr.r.r.r/rOs    rOc@s"eZdZdZddZd ddZdS) r^zHelper class for building block stats. When this class is created, we record the start time. When build() is called, the time delta is saved as part of the stats. cCst|_t|_dSr7)time perf_counter _start_time process_time _start_cpur\r.r.r/r]s z_BlockExecStatsBuilder.__init__r6rOcCs@t}t}t}|j|_||_||j|_||j|_ |Sr7) rfrgrirOrhrPrQrRrjrT)r8end_timeend_cpustatsr.r.r/builds  z_BlockExecStatsBuilder.buildN)r6rO)r)r*r+rdr]rnr.r.r.r/r^sr^c@s>eZdZUdZeeed<eeed<eeed<ddZdS)rJz#Statistics about the block producednum_rows size_bytes exec_statscCs"|jdur t|jtsJdSdSr7)rpr=intr\r.r.r/ __post_init__s zBlockStats.__post_init__N) r)r*r+rdr rr__annotations__rOrsr.r.r.r/rJs     cCsh|]}|jqSr.)name)rLfr.r.r/ srwcs:eZdZUdZeeeed<ddZfddZ Z S)r:zMetadata about the block. input_filescstdifddtDS)Ncsi|]}||qSr.)__getattribute__)rLkeyr\r.r/ sz*BlockMetadata.to_stats..r.)rJ_BLOCK_STATS_FIELD_NAMESr\r.r\r/rKszBlockMetadata.to_statscs"t|jdurg|_dSdSr7)superrsrxr\ __class__r.r/rss   zBlockMetadata.__post_init__) r)r*r+rdr r strrtrKrs __classcell__r.r.r~r/r:s alpha) stabilitycsneZdZUdZeeed<ddededffdd Z dde ded d dfd d Z e d efd dZ Z S)BlockMetadataWithSchemaNr;metadatarAcs&tj|j|j|j|jd||_dS)N)rxrprorq)r}r]rxrprorqr;)r8rr;r~r.r/r]s z BlockMetadataWithSchema.__init__blockrmrOr6cCs*t|}|j|d}|}t||dS)N)rq)rr;) BlockAccessor for_block get_metadatar;r)rrmaccessormetar;r.r.r/ from_blocks   z"BlockMetadataWithSchema.from_blockcCst|j|j|j|jdS)N)rorprqrx)r:rorprqrxr\r.r.r/r!s z BlockMetadataWithSchema.metadatar7)r)r*r+r;r rArtr:r]Blockrpropertyrrr.r.r~r/r s   rc@seZdZdZdefddZdedeefddZ dxd ed ed ede fd d Z de ede fddZ de ede fddZde eede fddZdeeefde fddZdedede fddZdeede fddZdyd!d"Z #dzdeeee efdeejeeejfffd$d%Zd{d'd(Zde fd)d*Zde fd+d,Zd-eedefd.d/Z defd0d1Z!dee"d2ffd3d4Z# # #d|d5ee ed6ee$de%fd7d8Z&d}d;d<Z'e(d~d>d?Z)e* #dzd@edAee+de fdBdCZ,e*d@eee-fde fdDdEZ.e*d@eee-fde fdFdGZ/e(dHe ddIfdJdKZ0dLedMdNdd:fdOdPZ1dxdQedRedee2fdSdTZ3dQedRedee2fdUdVZ4dQedRedee2fdWdXZ5dQedRedee2fdYdZZ6dQedRedee2fd[d\Z7 #dzdQedRed]ee2dee2fd^d_Z8dd`daZ9dbe edMdNde d:fdcddZ:dedNdfe;dgde fdhdiZe( mddje e dMdNdfe;dgdnede;e e=ff dodpZ?dbe e;e-dMdNde e fdqdrZ@de+fdsdtZAdue edejfdvdwZBd#S)raProvides accessor methods for a specific block. Ideally, we wouldn't need a separate accessor classes for blocks. However, this is needed if we want to support storing ``pyarrow.Table`` directly as a top-level Ray object, without a wrapping class (issue #17186). r6cCt)z2Return the number of rows contained in this block.NotImplementedErrorr\r.r.r/ro4zBlockAccessor.num_rowspublic_row_formatcCr)zIterate over the rows of this block. Args: public_row_format: Whether to cast rows into the public Dict row format (this incurs extra copy conversions). r)r8rr.r.r/ iter_rows8zBlockAccessor.iter_rowsFstartendcopycCr)a(Return a slice of this block. Args: start: The starting index of the slice (inclusive). end: The ending index of the slice (exclusive). copy: Whether to perform a data copy for the slice. Returns: The sliced block result. r)r8rrrr.r.r/sliceAs zBlockAccessor.sliceindicescCr)zReturn a new block containing the provided row indices. Args: indices: The row indices to return. Returns: A new block containing the provided row indices. r)r8rr.r.r/takeNs zBlockAccessor.takecolumnscCr)z array, e.g., `{'data': array}` instead of `array`.Nr)ArrowConversionError!_fallback_to_pandas_block_warningz)Failed to convert batch to Arrow due to: z; falling back to Pandas block)r=npndarrayrGr collectionsabcMappingr&r,$ray.air.util.tensor_extensions.arrowrbatch_to_arrow_blockrloggerwarningbatch_to_pandas_blockr-)clsrrrer.r.r/batch_to_blocks,     zBlockAccessor.batch_to_blockcCddlm}||S)z4Create an Arrow block from user-facing data formats.r)ArrowBlockBuilder)ray.data._internal.arrow_blockr_table_from_pydict)rrrr.r.r/r  z"BlockAccessor.batch_to_arrow_blockcCr)z4Create a Pandas block from user-facing data formats.r)PandasBlockBuilder)r<rr)rrrr.r.r/rrz#BlockAccessor.batch_to_pandas_blockrzBlockAccessor[T]cCstddl}ddl}t||j|jfrddlm}||St||jr.ddl m }||St|t r>ddlm}| |St|t rMtdt|dtd|t|)z,Create a block accessor for the given block.rN)ArrowBlockAccessor)PandasBlockAccessorrz: Standalone Python objects are not allowed in Ray 2.5. To use Python objects in a dataset, wrap them in a dict of numpy arrays, e.g., return `{'item': batch}` instead of just `batch`.zNot a block type: {} ({}))rr(r1r=Table RecordBatchrr DataFramer<rbytes from_byteslistrGr TypeErrorformattype)rr(r1rrr.r.r/rs"       zBlockAccessor.for_block n_samplessort_keyrcCr)z0Return a random sample of items from this block.r)r8rrr.r.r/samplerzBlockAccessor.sampleon ignore_nullscCr)z=Returns a count of the distinct values in the provided columnrr8rrr.r.r/countrzBlockAccessor.countcCr)z2Returns a sum of the values in the provided columnrrr.r.r/sum!rzBlockAccessor.sumcCr)z2Returns a min of the values in the provided columnrrr.r.r/min%rzBlockAccessor.mincCr)z2Returns a max of the values in the provided columnrrr.r.r/max)rzBlockAccessor.maxcCr)z3Returns a mean of the values in the provided columnrrr.r.r/mean-rzBlockAccessor.meanrcCr)zBReturns a sum of diffs (from mean) squared for the provided columnr)r8rrrr.r.r/sum_of_squared_diffs_from_mean1rz,BlockAccessor.sum_of_squared_diffs_from_meancCr)z9Returns new block sorted according to provided `sort_key`r)r8rr.r.r/sort:rzBlockAccessor.sort boundariescCr)z1Return a list of sorted partitions of this block.rr8rrr.r.r/sort_and_partition>sz BlockAccessor.sort_and_partitionrzaggsrcCr)z3Combine rows with the same key into an accumulator.r)r8rzrr.r.r/ _aggregateDrzBlockAccessor._aggregateblockscCr)z9Return a sorted block by merging a list of sorted blocks.r)rrr.r.r/merge_sorted_blocksHsz!BlockAccessor.merge_sorted_blocksTfinalizecCr)z/Aggregate partially combined and sorted blocks.r)rrrrr.r.r/_combine_aggregated_blocksOrz(BlockAccessor._combine_aggregated_blockscCr)zNOTE: PLEASE READ CAREFULLY Returns dataset partitioned using list of boundaries This method requires that - Block being sorted (according to `sort_key`) - Boundaries is a sorted list of tuples rrr.r.r/_find_partitions_sortedYs z%BlockAccessor._find_partitions_sortedcCr)z$Return the block type of this block.rr\r.r.r/rhrzBlockAccessor.block_typekeyscCsL|dkrtjgtjdS|std|gS||}tt|S)a NOTE: THIS METHOD ASSUMES THAT PROVIDED BLOCK IS ALREADY SORTED Compute boundaries of the groups within a block based on provided key (a column or a list of columns) NOTE: In each column, NaNs/None are considered to be the same group. Args: block: sorted block for which grouping of rows will be determined based on provided key keys: list of columns determining the key for every row based on which the block will be grouped Returns: A list of starting indices of each group and an end index of the last group, i.e., there are ``num_groups + 1`` entries and the first and last entries are 0 and ``len(array)`` respectively. r)dtype)rorarrayint32r"_get_group_boundaries_sorted_numpyrvalues)r8rprojected_blockr.r.r/_get_group_boundaries_sortedls  z*BlockAccessor._get_group_boundaries_sortedF)r6r!r7)r6r )NN)rrr6r)r6r)rrr6r)T)Cr)r*r+rdrrroboolr rrrrr rrrr rrr BlockColumnrrrrrrrrrr DataBatchrrprr;rOr:rrrer` classmethodr&rrrrrrrrrrrrrrrr rrrrrrrr.r.r.r/r+s         %        rbetac @seZdZdZdefddZdddeded eefd d Z dddeded eefd d Z dddeded eefddZ dddeded eefddZ dddeded eefddZ dddededed eefddZd efddZd eeeffddZd efddZd efddZd efdd Zd efd!d"Zd#dd$ded%eeded eefd&d'Zd eefd(d)Zd3d+ed ejfd,d-Zd eeed.ffd/d0Z e!ded dfd1d2Z"d#S)4BlockColumnAccessorzbProvides vendor-neutral interface to apply common operations to block's (Pandas/Arrow) columnscolcCs ||_dSr7)_column)r8rr.r.r/r]s zBlockColumnAccessor.__init__T)as_pyrrr6cCr)z4Returns a count of the distinct values in the columnrr8rrr.r.r/rrazBlockColumnAccessor.countcCr_)z)Returns a sum of the values in the columnrrr.r.r/rrazBlockColumnAccessor.sumcCr)z)Returns a min of the values in the columnrrr.r.r/rrazBlockColumnAccessor.mincCr)z)Returns a max of the values in the columnrrr.r.r/rrazBlockColumnAccessor.maxcCr)z*Returns a mean of the values in the columnrrr.r.r/rrazBlockColumnAccessor.meanqcCr)z.Returns requested quantile of the given columnr)r8rrrr.r.r/quantileszBlockColumnAccessor.quantilecCr)zBReturns new column holding only distinct values of the current onerr\r.r.r/uniquerazBlockColumnAccessor.uniquecCrr7rr\r.r.r/ value_countsz BlockColumnAccessor.value_countscCr)a Computes a 64-bit hash value for each row in the column. Provides a unified hashing method across supported backends. Handles complex types like lists or nested structures by producing a single hash per row. These hashes are useful for downstream operations such as deduplication, grouping, or partitioning. Internally, Polars is used to compute row-level hashes even when the original column is backed by Pandas or PyArrow. Returns: A column of 64-bit integer hashes, returned in the same format as the underlying backend (e.g., Pandas Series or PyArrow Array). rr\r.r.r/hashszBlockColumnAccessor.hashcCr)z;Flattens nested lists merging them into top-level containerrr\r.r.r/flattenszBlockColumnAccessor.flattencCrr7rr\r.r.r/dropnarzBlockColumnAccessor.dropnacCr)z Checks whether the column is composed of list-like elements. Returns: True if the column is made up of list-like values; False otherwise. rr\r.r.r/is_composed_of_listssz(BlockColumnAccessor.is_composed_of_listsN)rrrcCr)z9Returns a sum of diffs (from mean) squared for the columnr)r8rrrr.r.r/rsz2BlockColumnAccessor.sum_of_squared_diffs_from_meancCr)z8Converts block column to a list of Python native objectsrr\r.r.r/ to_pylistrazBlockColumnAccessor.to_pylistFzero_copy_onlycCr)z#Converts underlying column to Numpyr)r8r r.r.r/rrazBlockColumnAccessor.to_numpyr$cCr)zAConverts block column into a representation compatible with Arrowrr\r.r.r/_as_arrow_compatibleraz(BlockColumnAccessor._as_arrow_compatiblecCsntddl}t|tjst|tjrddlm}||St||jr-ddl m }||St dt |d)z-Create a column accessor for the given columnrN)ArrowBlockColumnAccessor)PandasBlockColumnAccessorzEExpected either a pandas.Series or pyarrow.Array (ChunkedArray) (got )) rr(r=paArray ChunkedArrayrr Seriesr<r rr)rpdr r r.r.r/ for_columns   zBlockColumnAccessor.for_columnr)#r)r*r+rdrr]rr rrrrrrfloatrrrrr rrrrrrrr rrrrr rerr.r.r.r/rsP         rrcCs@g}g}g}|D]0}t|jtjrt|dr||qt|jtjs3|ddur3||q||qg}t|dkrR|tdd|Djddt|dkri|tdd|Djddt|dkr|tdd|Djddt dgt |jdd ddt|dgg t }|S) NrcSs$g|]}|dd|ddkqSNrr.rLarrr.r.r/rNs$z6_get_group_boundaries_sorted_numpy..)axisc SsHg|] }|dd|ddkt|ddt|ddB@qSr)risfiniterr.r.r/rN"s "c SsNg|]#}|dd|ddkt|dddt|ddd@@qSr)requalrr.r.r/rN/s (r)r issubdtypernumberisnanappendlenvstackanyhstack column_stacknonzeroastyperr)rgeneral_arraysnum_arrays_with_nancat_arrays_with_nonerdiffsrr.r.r/rsV       r)Srloggingrf dataclassesrrenumrtypingrrrrr r r r r rrr2rr1rrUray.data._internal.utilrr ray.typesrray.utilrray.util.annotationsrr( ray.data._internal.block_builderrr<r2ray.data._internal.planner.exchange.sort_task_specrray.data.aggregaterrrrrrrrAr BatchColumn getLoggerr)rr&rr0rrDataBatchColumn CallableClassr4UserDefinedFunctionBlockPartitionBlockPartitionMetadatarFrErr?rBrHrKrOr^rJr|r:rrrrrr.r.r.r/s 4              $dv