o 1 i0p@s4ddlZddlZddlZddlmZmZddlmZddlm 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,fd 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.Z/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/data/block.pyr&Dsr&c@seZdZdZdZdZdS) BatchFormatpyarrowr(numpyN)r)r*r+r,r-ZNUMPYr.r.r.r/r0Jsr0r r!c@s*eZdZdedeeeeffddZdS)_CallableClassProtocol_CallableClassProtocol__argreturncCsdSNr.)selfr4r.r.r/__call__`sz_CallableClassProtocol.__call__N)r)r*r+rrrr r8r.r.r.r/r3_s"r3 BlockMetadata)r(r1r2Nr2schemar5cCs,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  rAgiven_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)rBr.r.r/_apply_batch_formatsrGmetas BlockStatscCsdd|DS)NcSsg|]}|qSr.)to_stats).0mr.r.r/ szto_stats..r.)rHr.r.r/rJsrJc@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_srayZruntime_contextZget_runtime_contextZ get_node_idnode_idZ max_uss_bytesZtask_idxr7r.r.r/__init__s zBlockExecStats.__init__r5_BlockExecStatsBuildercCtSr6)rXr.r.r.r/builderzBlockExecStats.buildercCst|j|j|j|jdS)N)rQrSrRrU)reprrQrSrRrUrVr.r.r/__repr__szBlockExecStats.__repr__N)r5rX)r)r*r+__doc__rW staticmethodrZr]r.r.r.r/rNs    rNc@s"eZdZdZddZd ddZdS) rXzHelper 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|_dSr6)time perf_counter _start_time process_time _start_cpurVr.r.r/rWs z_BlockExecStatsBuilder.__init__r5rNcCs@t}t}t}|j|_||_||j|_||j|_ |Sr6) r`rarcrNrbrOrPrQrdrS)r7end_timeZend_cpustatsr.r.r/builds  z_BlockExecStatsBuilder.buildN)r5rN)r)r*r+r^rWrgr.r.r.r/rXsrXc@s>eZdZUdZeeed<eeed<eeed<ddZdS)rIz#Statistics about the block producednum_rows size_bytes exec_statscCs"|jdur t|jtsJdSdSr6)rir<intrVr.r.r/ __post_init__s zBlockStats.__post_init__N) r)r*r+r^r rk__annotations__rNrlr.r.r.r/rIs     cCsh|]}|jqSr.)name)rKfr.r.r/ srpcs:eZdZUdZeeeed<ddZfddZ Z S)r9zMetadata about the block. input_filescstdifddtDS)Ncsi|]}||qSr.)__getattribute__)rKkeyrVr.r/ sz*BlockMetadata.to_stats..r.)rI_BLOCK_STATS_FIELD_NAMESrVr.rVr/rJszBlockMetadata.to_statscs"t|jdurg|_dSdSr6)superrlrqrV __class__r.r/rls   zBlockMetadata.__post_init__) r)r*r+r^r r strrmrJrl __classcell__r.r.rwr/r9s alpha)Z stabilitycsneZdZUdZeeed<ddededffdd Z dde ded d dfd d Z e d efd dZ Z S)BlockMetadataWithSchemaNr:metadatar@cs&tj|j|j|j|jd||_dS)N)rqrirhrj)rvrWrqrirhrjr:)r7r}r:rwr.r/rWs z BlockMetadataWithSchema.__init__blockrfrNr5cCs*t|}|j|d}|}t||dS)N)rj)r}r:) BlockAccessor for_block get_metadatar:r|)r~rfaccessormetar:r.r.r/ from_blocks   z"BlockMetadataWithSchema.from_blockcCst|j|j|j|jdS)N)rhrirjrq)r9rhrirjrqrVr.r.r/r}!s z BlockMetadataWithSchema.metadatar6)r)r*r+r:r r@rmr9rWBlockrpropertyr}rzr.r.rwr/r| s   r|c@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). r5cCt)z2Return the number of rows contained in this block.NotImplementedErrorrVr.r.r/rh4zBlockAccessor.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)r7rr.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)r7rrrr.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)r7rr.r.r/takeNs zBlockAccessor.takecolumnscCr)z array, e.g., `{'data': array}` instead of `array`.Nr)ArrowConversionErrorZ!_fallback_to_pandas_block_warningz)Failed to convert batch to Arrow due to: z; falling back to Pandas block)r<npndarrayrFr collectionsabcMappingr&r,Z$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_blockr~zBlockAccessor[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<TableZ RecordBatchrrZ DataFramer;rbytes from_byteslistrFr TypeErrorformattype)r~r(r1rrr.r.r/rs"       zBlockAccessor.for_block n_samplessort_keyrcCr)z0Return a random sample of items from this block.r)r7rrr.r.r/samplerzBlockAccessor.sampleon ignore_nullscCr)z=Returns a count of the distinct values in the provided columnrr7rrr.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)r7rrrr.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)r7rr.r.r/sort:rzBlockAccessor.sort boundariescCr)z1Return a list of sorted partitions of this block.rr7rrr.r.r/sort_and_partition>sz BlockAccessor.sort_and_partitionrsaggsrcCr)z3Combine rows with the same key into an accumulator.r)r7rsrr.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.rrVr.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)rhrarrayZint32r"_get_group_boundaries_sorted_numpyrvalues)r7rZprojected_blockr.r.r/_get_group_boundaries_sortedls  z*BlockAccessor._get_group_boundaries_sortedF)r5r!r6)r5r )NN)rrr5r)r5r)rrr5r)T)Cr)r*r+r^rkrhboolr rrrrr rryrr rrr BlockColumnrrrrrrrrrr DataBatchrrirr:rNr9rrr_rZ classmethodr&rrrrrrrrrrrrrrrr rr|rrrrrr.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 Zd4d"eed efd#d$Zd!dd%ded&eeded eefd'd(Zd eefd)d*Zd5d,ed ejfd-d.Zd e eed/ffd0d1Z!e"ded dfd2d3Z#d!S)6BlockColumnAccessorzbProvides vendor-neutral interface to apply common operations to block's (Pandas/Arrow) columnscolcCs ||_dSr6)_column)r7rr.r.r/rWs zBlockColumnAccessor.__init__T)as_pyrrr5cCr)z4Returns a count of the distinct values in the columnrr7rrr.r.r/rr[zBlockColumnAccessor.countcCrY)z)Returns a sum of the values in the columnrrr.r.r/rr[zBlockColumnAccessor.sumcCr)z)Returns a min of the values in the columnrrr.r.r/rr[zBlockColumnAccessor.mincCr)z)Returns a max of the values in the columnrrr.r.r/rr[zBlockColumnAccessor.maxcCr)z*Returns a mean of the values in the columnrrr.r.r/rr[zBlockColumnAccessor.meanqcCr)z.Returns requested quantile of the given columnr)r7rrrr.r.r/quantileszBlockColumnAccessor.quantilecCr)zBReturns new column holding only distinct values of the current onerrVr.r.r/uniquer[zBlockColumnAccessor.uniquecCrr6rrVr.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. :return: A column of 64-bit integer hashes, returned in the same format as the underlying backend (e.g., Pandas Series or PyArrow Array). rrVr.r.r/hashszBlockColumnAccessor.hashcCr)z;Flattens nested lists merging them into top-level containerrrVr.r.r/flattenszBlockColumnAccessor.flattencCrr6rrVr.r.r/dropnarzBlockColumnAccessor.dropnaNtypescCr)a Checks whether the column is composed of list-like elements. :param types: Optional tuple of backend-specific types to check against. If not provided, defaults to list-like types appropriate for the underlying backend (e.g., PyArrow list types). :return: True if the column is made up of list-like values; False otherwise. r)r7rr.r.r/is_composed_of_listss z(BlockColumnAccessor.is_composed_of_lists)rrrcCr)z9Returns a sum of diffs (from mean) squared for the columnr)r7rrrr.r.r/rsz2BlockColumnAccessor.sum_of_squared_diffs_from_meancCr)z8Converts block column to a list of Python native objectsrrVr.r.r/ to_pylistr[zBlockColumnAccessor.to_pylistFzero_copy_onlycCr)z#Converts underlying column to Numpyr)r7rr.r.r/rr[zBlockColumnAccessor.to_numpyr$cCr)zAConverts block column into a representation compatible with ArrowrrVr.r.r/_as_arrow_compatibler[z(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<paZArrayZ ChunkedArrayrrZSeriesr;rrr)rpdrrr.r.r/ for_columns   zBlockColumnAccessor.for_columnr6r)$r)r*r+r^rrWrr rrrrrrfloatrrrryr rrrrr rrrrrrrrrr_rr.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.rKarrr.r.r/rMs$z6_get_group_boundaries_sorted_numpy..)Zaxisc SsHg|] }|dd|ddkt|ddt|ddB@qSr)risfiniter r.r.r/rM#s "c SsNg|]#}|dd|ddkt|dddt|ddd@@qSr)requalr r.r.r/rM0s (r)rZ issubdtypernumberisnanappendlenZvstackanyZhstackZ column_stackZnonzeroZastyperk)rZgeneral_arraysZnum_arrays_with_nanZcat_arrays_with_noner Zdiffsrr.r.r/rsV       r)Srloggingr` dataclassesrrenumrtypingrrrrr r r r r rrr2rr1rrTZray.data._internal.utilrrZ ray.typesrZray.utilrZray.util.annotationsrr(Z ray.data._internal.block_builderrr;rZ2ray.data._internal.planner.exchange.sort_task_specrZray.data.aggregaterrrrrrrr@rZ BatchColumn getLoggerr)rr&ryr0rrZDataBatchColumnZ CallableClassr3ZUserDefinedFunctionZBlockPartitionZBlockPartitionMetadatarErDrr>rArGrJrNrXrIrur9r|rrrrr.r.r.r/s 4              $dw