&`iK2 UddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl mZmZmZddlmZddlmZmZmZmZmZmZmZmZmZmZmZmZddl Z!ddl"Z#ddl$Z$ddl%Z$ddl&Z&ddl'm(Z(ddl)m*Z*m+Z+m,Z,ddl-m.Z.ddl/Z/er0ddl"Z"ddl0m1Z1dd l2m3Z3dd l4m5Z5dd l6m7Z7m8Z8m9Z9m:Z:dd l;mm?Z?ej@eAZBdZCdeCzZDdeDzZEeFZGdZHdZIedeJefZKdaLeKeMd<GddZNeNZOdeKfdZPdZQ ddeRdeeRde,deed deeRdeddeeRdeeReSeeRffdZTd eddeRfd!ZUdeRfd"ZVd#ZWd$eSd%eSddfd&ZXd'eSdeSfd(ZYd)eeSeeSfdeJfd*ZZd+edeSfd,Z[d-ddd.d/eSd0eSd1eJd2eeSd3eRdeSf d4Z\ dd7eJd8eeSd9eeSd:eeSfd;Z]d<Z^d=Z_d>Z` dd?d@dAeeedBeeeSdCfdDeeeReeReRfeeReReRffddCf dEZadFeSfdGZbdFeSfdHZcdIdJdedKfdLZddMe!jede,dedKfdNZfdOedPddQfdRZgdSedQdedTfdUZhdVedTdedTfdWZidXedYdedTfdZZjdOedPd[eeeRekfd\d]deRfd^Zld_eSdefd`ZmedaZnedbZoGdcddeZp ddfeend?eeengeeofdgeJdheRdieRdeeoddff djZqGdkdlZrGdmdne$jsjtZuGdodpe$jsjvZwddqdrdsdtegefdueSdveeeSdweRdxeRdef dyZxdzeRdeSfd{Zydddd|d}eeRd~eeRdeeRdeeeReeRffdZzdeJfdZ{dedeJfdZ|dZ}deRfdZ~GddZdeeedfdeeedffdZdIe#jde#jfdZde#jde#jdeJfdZdeeRdeeRdeeRdeeSefdeeSeff dZe.d'eSdeeSfdZdS)N)EmptyFullQueue) ModuleType) TYPE_CHECKINGAnyCallableDict GeneratorIterableIteratorListOptionalTupleTypeVarUnion)call_with_retry)DEFAULT_READ_OP_MIN_NUM_BLOCKS WARN_PREFIX DataContext) DeveloperAPI)ComputeStrategy) RefBundle)SortKey)BlockBlockMetadataWithSchemaSchemaUserDefinedFunction) DatasourceReader)PlacementGroupilocalexample_pyarrow_datasetc6eZdZdZdZdZdZdZdZdZ dS) _OrderedNullSentinelzSentinel value that sorts greater than any other non-null value. NOTE: Semantic of this sentinel is closely mirroring that one of ``np.nan`` for the purpose of consistency in handling of ``None``s and ``np.nan``s. cdS)NFselfothers k/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/util.py__eq__z_OrderedNullSentinel.__eq__VsucJt|tpt|SN) isinstancer&is_nullr)s r,__lt__z_OrderedNullSentinel.__lt__Ys %!566H'%..Hr.c,||Sr0r3r)s r,__le__z_OrderedNullSentinel.__le__`s{{5!!!r.c.|| Sr0)r6r)s r,__gt__z_OrderedNullSentinel.__gt__e;;u%%%%r.c.|| Sr0r5r)s r,__ge__z_OrderedNullSentinel.__ge__hr9r.c t|Sr0)idr*s r,__hash__z_OrderedNullSentinel.__hash__ks $xxr.N) __name__ __module__ __qualname____doc__r-r3r6r8r;r?r(r.r,r&r&Ns{III""" &&&&&&r.r&returncRt ddlman#t$rdaYnwxYwtS)Nr)datasetF)r$pyarrowrFModuleNotFoundErrorr(r.r,_lazy_import_pyarrow_datasetrIrsS % ; ; ; ; ; ; ;" % % % %    % s  cLtjjdSr0)ray_private arrow_utils_check_pyarrow_versionr(r.r,rNrN~sL3355555r. parallelismtarget_max_block_sizectxdatasource_or_legacy_readermem_sizeplacement_groupr! avail_cpusc d}tj}||r|}|[tj|sG|Et dt ||z }t dt ||jz }d} |dkr<|dkrtd|j F|j tkr6|j tkr&t d|j |_ |tj}|pt#|}t t%|j |||dz}||j kr d |j } nH||krd |jt&z d } n.||kr"|jd } n|jt&z d } d | } nd|d} td|d|d|d|| |fS)aReturns parallelism to use and the min safe parallelism to avoid OOMs. This detects parallelism using the following heuristics, applied in order: 1) We start with the default value of 200. This can be overridden by setting the `read_op_min_num_blocks` attribute of :class:`~ray.data.context.DataContext`. 2) Min block size. If the parallelism would make blocks smaller than this threshold, the parallelism is reduced to avoid the overhead of tiny blocks. 3) Max block size. If the parallelism would make blocks larger than this threshold, the parallelism is increased to avoid OOMs during processing. 4) Available CPUs. If the parallelism cannot make use of all the available CPUs in the cluster, the parallelism is increased until it can. Args: parallelism: The user-requested parallelism, or -1 for auto-detection. target_max_block_size: The target max block size to produce. We pass this separately from the DatasetContext because it may be set per-op instead of per-Dataset. ctx: The current Dataset context to use for configs. datasource_or_legacy_reader: The datasource or legacy reader, to be used for data size estimation. mem_size: If passed, then used to compute the parallelism according to target_max_block_size. placement_group: The placement group that this Dataset will execute inside, if any. avail_cpus: Override avail cpus detection (for testing only). Returns: Tuple of detected parallelism (only if -1 was specified), the reason for the detected parallelism (only if -1 was specified), and the estimated inmemory size of the dataset. Nrz6`parallelism` must either be -1 or a positive integer.zy``DataContext.min_parallelism`` is deprecated in Ray 2.10. Please specify ``DataContext.read_op_min_num_blocks`` instead.z1DataContext.get_current().read_op_min_num_blocks=zOoutput blocks of size at least DataContext.get_current().target_min_block_size=z MiB unlimitedzNoutput blocks of size at most DataContext.get_current().target_max_block_size=z9parallelism at least twice the available number of CPUs ()zAutodetected parallelism=z# based on estimated_available_cpus=z and estimated_data_size=.)sysmaxsizeestimate_inmemory_data_sizenpisnanmaxinttarget_min_block_size ValueErrormin_parallelismrread_op_min_num_blocksloggerwarningrKutilget_current_placement_group_estimate_avail_cpusminMiBrPdebug) rOrPrQrRrSrTrUmin_safe_parallelismmax_reasonable_parallelismreason display_vals r,_autodetect_parallelismrus~V!$7.JJLL""  ! -"1c(5J*J&K&KLL%(C3;T0T,U,U%V%V" FQ "  UVV V   +#'EEE*.LLL NNQ   *-). This can lead to slowdowns during the data reading phase due to excessive task creation. Reduce the parallelism to match with the available CPU slots in the cluster, or set parallelism to -1 for Ray Data to automatically determine the parallelism. You can ignore this message if the cluster is expected to autoscale.)rKavailable_resourcesr{rirjr)requested_parallelismnum_read_tasksavailable_cpu_slotss r,_warn_on_high_parallelismr"s13377qAA 014 4 4 d " " S S:O S S" S S S   4 4 " "r.modulepackagec  tj|dS#t$r'td|jjd|d|d|d wxYw)aRCheck if a required dependency is installed. If `module` can't be imported, this function raises an `ImportError` instructing the user to install `package` from PyPI. Args: obj: The object that has a dependency. module: The name of the module to import. package: The name of the package on PyPI. `z` depends on 'z-', but Ray Data couldn't import it. Install 'z' by running `pip install z`.N) importlib import_module ImportError __class__r@)objrrs r, _check_importr5s '''''     Q & Q Qf Q Q#) Q QEL Q Q Q    s 1A pathc\tj|}|jtkr|j|jz}nl|jtkr\tj tj j dz dz }||j|jzz }t| }|S)zReturns the resolved path if the given path follows a Ray-specific custom scheme. Othewise, returns the path unchanged. The supported custom schemes are: "local", "example". examplesdata)urllibparseurlparsescheme _LOCAL_SCHEMEnetlocr_EXAMPLE_SCHEMEpathlibPath__file__parentstrresolve)r parsed_uriexample_data_paths r,_resolve_custom_schemerIs &&t,,JM)) :?2  o - -#L229@:MPVV J$5 $GH4<<>>"" Kr.pathsct|tr|g}t|tjrt|g}n_t|trt d|Drt dt|dkrt dtd|D}|dkr%|t|krt d||t|kS)zReturns True if the given paths are in local scheme. Note: The paths must be in same scheme, i.e. it's invalid and will raise error if paths are mixed with different schemes. c3BK|]}t|t VdSr0)r1r).0ps r, z#_is_local_scheme..bs/+R+Rq 1c0B0B,B+R+R+R+R+R+Rr.z6paths must be a path string or a list of path strings.rzMust provide at least one path.c3pK|]1}tj|jtkV2dSr0)rrrrr)rrs r,rz#_is_local_scheme..fs: T Tdfl##D))0MA T T T T T Tr.zHThe paths must all be local-scheme or not local-scheme, but found mixed ) r1rrrlistanyrflensum)rnums r,_is_local_schemerYs  %%&&<U  t $ $<+R+RE+R+R+R(R(R<QRRR Uq:;;; T Te T T T T TC Qww3U## '$ ' '    #e** r.rcdt|}t|dkr |dddz}|S)zGUtility to return a truncated object representation for error messages.Nz...)rr)rmsgs r,_truncated_reprros3 c((C 3xx#~~$3$i% Jr.T) insert_after directive skip_matchesmessagepatternrrrc d|vrtd||j}|sd}|dkr|r|}d}n|}||} |} | dkr|r| t |z} n |d| ddz} |d| }|| d}| dz} | dkrnR|s|| | z t |zd}||} | dktd|d|d |t tt| } t | dkr| } nAt ttt| } t | dksJd t | dt | d z z}|d}|0|d |d }| dd|zd z}||zd z|z}n|| dd|zz}||dkz rd|z}| |dkz r|dz}|||g}d ||_dS)N zomessage shouldn't contain any newlines, since this function will insert its own linebreaks when text wrapping: rXrYrWrzPattern z not found after z skips in docstring  z.. z:: z z )rfrCstripfindrrfindrfilterbool splitlinesreversedlstripreplacejoin)rrrrrrdocheadtailiskip_matches_leftoffset after_lineslinesindentbasepartss r,_insert_doc_at_patternrws w @6= @ @   +    C "}}} IIg  (2gg 2S\\)bqb--1=D=D  "  A%%! 9AJW5778 '""A2gg"7\  vdDOO$5$56677K ;!VD(4??+<+<"="=>>?? u::>>>> CaMMCa(9(9$:$:: ;FmmD!!G,,Y,,,//$v (?@@-')G37??4???w&()#7" W./#F"7D !E''%..CKKKr.F Examples:if_more_than_readdatasource_metadataextra_conditiondelegatec| d}|r||z n'|sd|z nd}| |d|dz }|||dzz }|dz|z  fd }|S) zAnnotate the function with an indication that it's a consumption API, and that it will trigger Dataset execution. zN will trigger execution of the lazy transformations performed on this dataset.zThis operationz.If this dataset consists of more than a read, Nz or if the zC can't be determined from the metadata provided by the datasource, z, zthen this operationc0t|d|S)Nnoterrrrr)rrrrs r,wrapz_consumption_api..wraps0 %      r.r() rrrrrrr conditionrrs `` @r,_consumption_apirs   ;T/  ;"T)D  * 80888 I  & 4/ /I33d: Kr.ct|dkrEt|dkr2t|drt|dSt|i|S)rrWr)rcallablerargskwargss r,ConsumptionAPIrsb 4yyA~~#f++**xQ/@/@*!!!$q'*** T ,V , ,,r.c d}|S)Annotate the function with an indication that it's a all to all API, and that it is an operation that requires all inputs to be materialized in-memory to execute. c.t|dddd|S)NzXThis operation requires all inputs to be materialized in object store for it to execute.rFrrr)rs r,rz_all_to_all_api..wraps4 B   r.r()rrrs r,_all_to_all_apirs    Kr.ct|dkr(t|dkrt|dsJt|dS)rrWr)rrrrs r, AllToAllAPIrsT t99>>c&kkQ..8DG3D3D.. D ?  T!W % %%r.fnrfn_constructor_argscomputer concurrencycddlm}m}ddlm}t ||rd}nd}|t d|d|^|r,|d kst ||rt d |d |d |s,|d kst ||rt d|d |d|S|tdt |trt|dvstd|Dst d|d|st d|dt|dkr||d|dS||d|d|dSt |tr|r ||S||St d|d|r |ddS|S)aaGet `ComputeStrategy` based on the function or class, and concurrency information. Args: fn: The function or generator to apply to a record batch, or a class type that can be instantiated to create such a callable. fn_constructor_args: Positional arguments to pass to ``fn``'s constructor. compute: Either "tasks" (default) to use Ray Tasks or an :class:`~ray.data.ActorPoolStrategy` to use an autoscaling actor pool. concurrency: The number of Ray workers to use concurrently. Returns: The `ComputeStrategy` for execution. r)ActorPoolStrategyTaskPoolStrategy) CallableClassTFNzj``fn_constructor_args`` can only be specified if providing a callable class instance for ``fn``, but got: r]tasksz!You specified the callable class z as your UDF with the compute z, but Ray Data can't schedule callable classes with the task pool strategy. To fix this error, pass an ActorPoolStrategy to compute or None to use the default compute strategy.actorszYou specified the function z, but Ray Data can't schedule regular functions with the actor pool strategy. To fix this error, pass a TaskPoolStrategy to compute or None to use the default compute strategy.zThe argument ``concurrency`` is deprecated in Ray 2.51. Please specify argument ``compute`` instead. For more information, see https://docs.ray.io/en/master/data/transforming-data.html#stateful-transforms.)rZc3@K|]}t|tVdSr0)r1rd)rcs r,rz'get_compute_strategy.._s=99'( 1c""999999r.zG``concurrency`` is expected to be set as a tuple of integers, but got: zS``concurrency`` is set as a tuple of integers, but ``fn`` is not a callable class: zD. Use ``concurrency=n`` to control maximum number of workers to use.rZrW)min_sizemax_size)rr initial_size)sizezU``concurrency`` is expected to be set as an integer or a tuple of integers, but got: ) ray.data._internal.computerrray.data.blockrr1rfrirjtuplerallrd)rrrrrrris_callable_classs r,get_compute_strategyr s$*ONNNNNNN,,,,,,"m$$  "  *F@BFFF     w  *W6F"G"G =B=====  #  x  :g7H#I#I =b=====     #   k5 ) )& ;v--S99,799966-!9*5999 %  @02@@@;1$$(((^k!n)((^(^!,Q  S ) )   :((k::::''[9999>/:>>>   &$$a$??? ?##%% %r.scL|d|ddzS)zCapitalize the first letter of a string Args: s: String to capitalize Returns: Capitalized string rrWN)upperr s r,capfirstrs# Q4::<.s(5518A;;555555r._)rsplitrs r, capitalizers/ 7755 555 5 55r.dfpandas.DataFrame)rrcddlm}m}m}||}|}||j||fS)Nr BlockAccessorBlockExecStatsrstats) rrrr for_blockto_arrowbuilder from_blockbuild)rrrrblockrs r,pandas_df_to_arrow_blockr$szVUUUUUUUUU  # #B ' ' 0 0 2 2E  " " $ $E 4)4U%++--PPP PPr.ndarraycddlm}m}m}t j||}|d|i}||j|| fS)Nrrrr) rrrrr _set_currentr batch_to_blockr!r")r%rQrrrrr#s r,ndarray_to_blockr)sVUUUUUUUUUS!!!  " " $ $E  ( (&'): ; ;E 4)4U%++--PPP PPr.table)z pyarrow.Tablerrcddlm}m}|}|j||S)Nr)rrr)rrrr r!r")r*rrrs r,get_table_block_metadata_schemar,sOGFFFFFFF  " " $ $E - " -e5;;== I I IIr.block_metadata_with_schemasrcg}|D]5}|j,|j |jdkr||j6t|S)a\For the input list of BlockMetadata, return a unified schema of the corresponding blocks. If the metadata have no valid schema, returns None. Args: block_metadata_with_schemas: List of BlockMetadata to unify Returns: A unified schema of the input list of schemas, or None if no valid schemas are provided. Nrschemanum_rowsappendunify_schemas_with_validation)r-schemas_to_unifyms r,unify_block_metadata_schemar6sX& (.. 8 QZ%71:>>  # #AH - - - ()9 : ::r.r4c|rPddlm} ddln#t$rdYnwxYw(t fd|Dr ||dS|dSdS)Nr) unify_schemasc3BK|]}t|jVdSr0)r1r)rr pas r,rz0unify_schemas_with_validation..s/!U!Uq*Q ":":!U!U!U!U!U!Ur.T) promote_types).ray.data._internal.arrow_ops.transform_pyarrowr8rGrr)r4r8r:s @r,r3r3s #PPPPPP     BBB  >c!U!U!U!UDT!U!U!UUU> =!1FFF F "" 4s   ref_bundlesrcg}|D]O}|jF||dkr||jPt|S)Nrr/)r=r4rs r,unify_ref_bundles_schemar?sk33 = $ OO   %):):Q)>)>  # #FM 2 2 2 ()9 : ::r.desiredsort_keyrc |}|}dt|}}tt|D]}||kr|cS||}||||} ||} | t } |} ||dur}t jt| dz dd} | t| t j| | d| z z}| t| t j| | d| z z}| t j| | d z}| t j| | d z}|ddur|n|S) avFor the given block, find the index where the desired value should be added, to maintain sorted order. We do this by iterating over each column, starting with the primary sort key, and binary searching for the desired value in the column. Each binary search shortens the "range" of indices (represented by ``left`` and ``right``, which are indices of rows) where the desired value could be inserted. Args: table: The block to search in. desired: A single tuple representing the boundary to partition at. ``len(desired)`` must be less than or equal to the number of columns being sorted. sort_key: The sort key to use for sorting, providing the columns to be sorted and their directions. Returns: The index where the desired value should be inserted to maintain sorted order. rNTrWrYright)sidesorterleft)rD) get_columnsget_descendingrrangeto_numpy NULL_SENTINELraarange searchsorted) r*r@rAcolumns descendingrFrCrcol_namecol_vals desired_valprevleftrEs r,find_partition_indexrTs2""$$G((**JSZZ%D 3w<< (T(T 5==LLL1:?++--d5j9aj   'K a=D Ys8}}q0"b99FH / ! DH /! EEboh &QQQQDrx7SSSSEEqMT))55t3r. class_namecddlm}|d}t|dkrt d|dd|dd}|d}t |||S)zGet Python attribute from the provided class name. The caller needs to make sure the provided class name includes full module name, and can be imported successfully. r)rr]rZzCannot create object from NrY)rrrrrfrgetattr)rUrr module_nameattribute_names r,get_attribute_from_class_namerZFs ('''''   S ! !E 5zzA~~CjCCCDDD((5":&&K2YN ==--~ > >>r.TUcbeZdZdZdZ d dedeejffd Z d fd Z d fd Z xZ S) _InterruptibleQueueznExtension of Python's `queue.Queue` providing ability to get interrupt its method callers in other threadsg?Nrinterrupted_eventct||ptj|_dS)N)r_)super__init__ threadingEvent_interrupted_event)r*rr_rs r,rbz_InterruptibleQueue.__init__as; ***"3"Hy7H7Hr.Tc|r|"t||S |jrt  td|jS#t $rYnwxYwaNT)r#timeout)rar{reis_setInterruptedError INTERRUPTION_CHECK_FREQUENCY_SECr)r*r#rhrs r,r{z_InterruptibleQueue.getgs /+77;;ug.. . &--// )&((( ww{{(M#     s'A99 BBc |r|%t|||dS |jrt  t|d|jdS#t $rYnwxYwdrg)raputrerirjrkr)r*itemr#rhrs r,rmz_InterruptibleQueue.put~s + GGKKeW - - - F &--// )&(((  d.S     s)A?? B  B r0)TN) r@rArBrCrkrdrrcrdrbr{rm __classcell__rs@r,r^r^[s''(+$MQIII080IIIIIII .r.r^rW base_iteratorpreserve_ordering num_workers buffer_sizec# Ktjdddkrtdtj|rdn dzzdfdt Dfdt Dfd}fd  tj|d d }| fd t D}|D]}| } t| dkrxg | D]P} | } t| tr| | tur | L| VQ r fd| D} t| dkxdS#wxYw)a=Returns a generator (iterator) mapping items from the provided iterator applying provided transformation in parallel (using a thread-pool). NOTE: There are some important constraints that needs to be carefully understood before using this method 1. If `preserve_ordering` is True a. This method would unroll input iterator eagerly (irrespective of the speed of resulting generator being consumed). This is necessary as we can not guarantee liveness of the algorithm AND preserving of the original ordering at the same time. b. Resulting ordering of the output will "match" ordering of the input, ie that: iterator = [A1, A2, ... An] output iterator = [map(A1), map(A2), ..., map(An)] 2. If `preserve_ordering` is False a. No more than `num_workers * (queue_buffer_size + 1)` elements will be fetched from the iterator b. Resulting ordering of the output is unspecified (and is non-deterministic) Args: base_iterator: Iterator yielding elements to map fn: Transformation to apply to each element preserve_ordering: Whether ordering has to be preserved num_workers: The number of threads to use in the threadpool (defaults to 1) buffer_size: Number of objects to be buffered in its input/output queues (per queue; defaults to 2). Total number of objects held in memory could be calculated as: num_workers * buffer_size * 2 (input and output) Returns: An generator (iterator) of the elements corresponding to the source elements mapped by provided transformation (while *preserving the ordering*) rirWz&Size of threadpool must be at least 1.rYc0g|]}tSr(r^)rrinput_queue_buf_sizer_s r, z"make_async_gen..s5  02CDDr.c0g|]}tSr(rw)rrrtr_s r,ryz"make_async_gen..s2@AK):;;r.c tD]#\}}|z|$tD]%}|zt&dS#t$rYdSt $rN}t d|tD]}||Yd}~dSd}~wwxYw)Nz#Caught exception in filling worker!exc_info) enumeratermrISENTINELrj Exceptionrirjr) idxrne output_queuerq input_queuesnum_input_queuesrs output_queuess r,_run_filling_workerz+make_async_gen.._run_filling_workers/ $'}55 ? ? TS#33488>>>>[)) C CS#33488BBBB C C     DD $ $ $ NN@1N M M M !) 7 7 $ $   #### $ $ $ $ $ $  $sA(A-- C: CAC  Ccb t|jt}|D]}|||tdS#t$rYdSt $r<}t d|||Yd}~dSd}~wwxYw)Nz(Caught exception in transforming worker!r|)iterr{rrmrjrrirj) input_queuerinput_queue_iterresultrrs r,_run_transforming_workerz0make_async_gen.._run_transforming_workers  $KOX>> "-.. ) )  ((((   X & & & & &    DD   NNEPQN R R R   Q           sAA B.) B.21B))B.zmap_tp_filling_worker-T)targetnamedaemonc ng|]1}tjdd||z|fd2S)zmap_tp_transforming_worker--T)rrrr)rcThread)rrrgen_idrrrs r,ryz"make_async_gen..2sn###  +=v====s%556 c8JK    ###r.cg|]}|v| Sr(r()rq empty_queuess r,ryz"make_async_gen..cs*+++!<:O:OA:O:O:Or.N)randomrandintrfrcrdrIrstartrr{r1rrr2set)rqrrrrsrtrfilling_worker_threadtransforming_worker_threadstremaining_output_queuesrrnrrrrxrr_rrs`` `` @@@@@@@@r,make_async_genrs`^Ay ) )FQABBB")).!& +a;>'((L EJ;EWEWM $$$$$$$$$:     0&," .f . . !!!########%%###)  2 "/)**Q..L!8   #''))dI..J8## '' 5555JJJJ ++++6+++'?)**Q..\ s B F--Gc XeZdZ ddejdededefdZdZd e d e fd Z d Z d Z dS)RetryingContextManager fcontext max_attempts max_backoff_sc>||_||_||_||_dSr0)_f _data_context _max_attempts_max_backoff_s)r*rrrrs r,rbzRetryingContextManager.__init__vs)$)+r.cVd|jjd|jdS)N)rr@handlerunwrapr>s r,__repr__zRetryingContextManager.__repr__s.H4>*HH 0C0C0E0EHHHHr. operation descriptioncRt|||jj|j|jSz"Execute an operation with retries.)rmatchrr)rrretried_io_errorsrrr*rrs r,_retry_operationz'RetryingContextManager._retry_operations4 #$6+-     r.cB||jjdS)Nzenter file context)rr __enter__r>s r,rz RetryingContextManager.__enter__s$$TW%68LMMMr.cFfdddS)Nc<jSr0)r__exit__)exc_type exc_valuer* tracebacksr,z1RetryingContextManager.__exit__..sDG$$Xy)DDr.zexit file contextr)r*rrrs````r,rzRetryingContextManager.__exit__s@  D D D D D D D      r.Nrr)r@rArBrG NativeFilerrdrbrr rrrrr(r.r,rrus  , ,   , , ,  , , , ,III (     NNN     r.rc eZdZdfd ZedeefdZdZe dd d d eed e d e fdZ dZ e dZ xZS)RetryingPyFileSystemrRetryingPyFileSystemHandlerct|tstdsJt|dS)Nz-handler must be a RetryingPyFileSystemHandler)r1rrfrarb)r*rrs r,rbzRetryingPyFileSystem.__init__sJ'#>?? OMNN N NN !!!!!r.rDc|jjSr0)r_retryable_errorsr>s r,retryable_errorsz%RetryingPyFileSystem.retryable_errorss |--r.c4|jSr0)rrr>s r,rzRetryingPyFileSystem.unwraps|""$$$r.rrfspyarrow.fs.FileSystemrrrcjt|tr|St||||}||Sr0)r1rr)clsrrrrrs r,rzRetryingPyFileSystem.wrapsE b. / / I-  ,   s7||r.c |j|jffSr0)rrr>s r, __reduce__zRetryingPyFileSystem.__reduce__s00r.c ||Sr0r()rstates r, __setstate__z!RetryingPyFileSystem.__setstate__ssE{r.)rrr)r@rArBrbpropertyrrrr classmethodrdrrrrorps@r,rrs"""""" .$s)...X.%%%    # s)      [ 111[r.rc PeZdZdZeddfdddeededefd Zd e d efd Z d Z dedefdZ dede fdZdefdZd)dede fdZdefdZdZddde fdZdeefdZdZd Zdedefd!Zdedefd"Z d*dedd$fd%Zdedd$fd&Z d*dedd$fd'Zdedd$fd(Zd#S)+rzWrapper for filesystem objects that adds retry functionality for file operations. This class wraps any filesystem object and adds automatic retries for common file operations that may fail transiently. rrrrrrrc|t|tr Jd||_||_||_||_dS)aInitialize the retrying filesystem wrapper. Args: fs: The underlying filesystem to wrap context: DataContext for retry settings max_attempts: Maximum number of retry attempts max_backoff_s: Maximum backoff time in seconds z"Cannot wrap a RetryingPyFileSystemN)r1r_fsrrr)r*rrrrs r,rbz$RetryingPyFileSystemHandler.__init__sZ $   0 0 / 0 0 !1)+r.rrcHt|||j|j|jSr)rrrrrs r,rz,RetryingPyFileSystemHandler._retry_operations1 #(+-     r.c|jSr0)rr>s r,rz"RetryingPyFileSystemHandler.unwraps xr.srcdestcJfdddS)z Copy a file.c:jSr0)r copy_filerr*rsr,rz7RetryingPyFileSystemHandler.copy_file..sDH&&sD11r.zcopy file from  to rr*rrs```r,rz%RetryingPyFileSystemHandler.copy_filesB$$ 1 1 1 1 1 13TS3T3Td3T3T   r.r recursivecDfddS)z&Create a directory and subdirectories.c<jS)N)r)r create_dir)rrr*sr,rz8RetryingPyFileSystemHandler.create_dir..sDH'' 'BBr.zcreate directory r)r*rrs```r,rz&RetryingPyFileSystemHandler.create_dirs:$$ B B B B B B & & &   r.c@fddS)z1Delete a directory and its contents, recursively.c8jSr0)r delete_dirrr*sr,rz8RetryingPyFileSystemHandler.delete_dir..sDH''--r.zdelete directory rr*rs``r,rz&RetryingPyFileSystemHandler.delete_dirs6$$ - - - - -/I4/I/I   r.Fmissing_dir_okcDfddS)z+Delete a directory's contents, recursively.c<jS)N)rrdelete_dir_contents)rrr*sr,rzARetryingPyFileSystemHandler.delete_dir_contents..sDH00n0UUr.zdelete directory contents r)r*rrs```r,rz/RetryingPyFileSystemHandler.delete_dir_contentss:$$ U U U U U U / / /   r.c@fddS)zDelete a file.c8jSr0)r delete_filersr,rz9RetryingPyFileSystemHandler.delete_file..sDH((..r.z delete file rrs``r,rz'RetryingPyFileSystemHandler.delete_files6$$ . . . . .0Et0E0E   r.c6fddS)Nc<jddS)N/T)accept_root_dirrr>sr,rzFRetryingPyFileSystemHandler.delete_root_dir_contents.. sDH00d0KKr.zdelete root dir contentsrr>s`r,delete_root_dir_contentsz4RetryingPyFileSystemHandler.delete_root_dir_contents s*$$ K K K K &   r.r+rDc6|j|S)z'Test if this filesystem equals another.)requalsr)s r,rz"RetryingPyFileSystemHandler.equalssxu%%%r.rc@fddS)zGet info for the given files.c8jSr0r get_file_info)rr*sr,rz;RetryingPyFileSystemHandler.get_file_info..sDH**511r.get file info for r)r*rs``r,rz)RetryingPyFileSystemHandler.get_file_infos6$$ 1 1 1 1 1 ( ( (   r.c@fddS)Nc8jSr0r)selectorr*sr,rzDRetryingPyFileSystemHandler.get_file_info_selector..sDH**844r.rr)r*rs``r,get_file_info_selectorz2RetryingPyFileSystemHandler.get_file_info_selectors6$$ 4 4 4 4 4 + + +   r.cdS)Nrr(r>s r, get_type_namez)RetryingPyFileSystemHandler.get_type_name!s%%r.cJfdddS)z"Move / rename a file or directory.c:jSr0)rmoversr,rz2RetryingPyFileSystemHandler.move..'sDHMM#t,,r.z move from rrrs```r,r z RetryingPyFileSystemHandler.move$sB$$ , , , , , ,.J3.J.JD.J.J   r.c@fddS)zNormalize filesystem path.c8jSr0)rnormalize_pathrsr,rz.-sDH++D11r.znormalize path rrs``r,rz*RetryingPyFileSystemHandler.normalize_path*s6$$ 1 1 1 1 13KT3K3K   r.Nzpyarrow.NativeFilecDfddS)zOpen an output stream for appending. Compression is disabled in this method because it is handled in the PyFileSystem abstract class. c>jdSN) compressionmetadata)ropen_append_streamrrr*sr,rz@RetryingPyFileSystemHandler.open_append_stream..;'DH// !0r.zopen append stream for rr*rrs```r,rz.RetryingPyFileSystemHandler.open_append_stream0J$$       -d , ,    r.c@fddS)zOpen an input stream for sequential reading. Compression is disabled in this method because it is handled in the PyFileSystem abstract class. c<jdS)N)r)ropen_input_streamrsr,rz?RetryingPyFileSystemHandler.open_input_stream..MsDH..t.FFr.zopen input stream for rrs``r,rz-RetryingPyFileSystemHandler.open_input_streamCs8$$ F F F F F +T + +   r.cDfddS)zOpen an output stream for sequential writing." Compression is disabled in this method because it is handled in the PyFileSystem abstract class. c>jdSr)ropen_output_streamrsr,rz@RetryingPyFileSystemHandler.open_output_stream..\rr.zopen output stream for rrs```r,r z.RetryingPyFileSystemHandler.open_output_streamQrr.c@fddS)z-Open an input file for random access reading.c8jSr0)ropen_input_filersr,rz=RetryingPyFileSystemHandler.open_input_file..gsDH,,T22r.zopen input file rrs``r,r#z+RetryingPyFileSystemHandler.open_input_fileds6$$ 2 2 2 2 24Mt4M4M   r.)Fr0)r@rArBrCrrrrdrbr rrrrrrrrrrrrr r rrrr r#r(r.r,rrs',egg ,, #,s), ,  ,,,,. (      S      s t     s       T            &3&&&&& 49       &&&  3     3 3             &          "         & C ,@      r.rrr)rrriterable_factoryrrrrc # K|dksJd|dd}t|D]} |}t|D]\}} ||kr |dz }| VdS#t$r |dupt fd|D} | ro|dz|krft d|dzz|t jz} t d|dzd |d | d tj | n dYd d wwxYwdS) aIterate through an iterable with retries. If the iterable raises an exception, this function recreates and re-iterates through the iterable, while skipping the items that have already been yielded. Args: iterable_factory: A no-argument function that creates the iterable. match: A list of strings to match in the exception message. If ``None``, any error is retried. description: An imperitive description of the function being retried. For example, "open the file". max_attempts: The maximum number of attempts to retry. max_backoff_s: The maximum number of seconds to backoff. rWz%`max_attempts` must be positive. Got r]rNc3:K|]}|tvVdSr0)r)rrrs r,rz%iterate_with_retry..s./W/Wg3q660A/W/W/W/W/W/Wr.rZz Retrying z attempts to z after z seconds.) rIr~rrrnrrirptimesleep) r$rrrrnum_items_yieldedattemptiterable item_indexrn is_retryablebackoffrs @r,iterate_with_retryr/ks, 1   UlUUU   &&"" "''))H$-h$7$7   D 111!Q&! FF " " " D=WC/W/W/W/WQV/W/W/W,W,WL "! l : :qWq[1MBBV]__T 0 00 00$000 7####T!$#### """s/A D(BDD num_bytesc|dkrt|dz d}n1|dkrt|dz d}nt|dz d}|S)NgeAGBg.AMBg@@KB)round)r0 num_bytes_strs r,#convert_bytes_to_human_readable_strr7spC S11555 c   S11555  S11555 r.)num_rows_per_filemin_rows_per_filemax_rows_per_filer8r9r:c|4ddl}|dtd|td|}||dkrtd||dkrtd||||krtd |d |d ||fS) a{Helper method to validate and handle rows per file arguments. Args: num_rows_per_file: Deprecated parameter for number of rows per file min_rows_per_file: New parameter for minimum rows per file max_rows_per_file: New parameter for maximum rows per file Returns: A tuple of (effective_min_rows_per_file, effective_max_rows_per_file) Nrzk`num_rows_per_file` is deprecated and will be removed in a future release. Use `min_rows_per_file` instead.r) stacklevelz~Cannot specify both `num_rows_per_file` and `min_rows_per_file`. Use `min_rows_per_file` as `num_rows_per_file` is deprecated.z,max_rows_per_file must be a positive integerz,min_rows_per_file must be a positive integerzmin_rows_per_file (z,) cannot be greater than max_rows_per_file (r\)warningswarnDeprecationWarningrf)r8r9r:r=s r,_validate_rows_per_file_argsr@s $  /      (P .$):a)?)?GHHH$):a)?)?GHHH %  )  1 1 1 7"3 7 7"3 7 7 7   / //r.cx t|totj|S#t$rYdSwxYw)z+Returns true if provide value is ``np.nan``F)r1floatrarb TypeErrorvalues r,is_nanrFsF%'';BHUOO; uus (+ 99rEc(|dupt|S)zYThis generalization of ``is_nan`` util qualifying both None and np.nan as null valuesN)rFrDs r,r2r2s D= )F5MM)r.ct|t|krdSt||D],\}}t|rt|s ||ksdS-dS)NFT)rziprF)keys1keys2k1k2s r, keys_equalrNsk 5zzSZZueU##B r  rRxx55 4r.ctj}tjj}||vsJd|d|||dS)zsReturn the total object store memory on the current node. This function incurs an RPC. Use it cautiously. zExpected node 'z' to be in resources: object_store_memory)rKget_runtime_context get_node_idrLrtotal_resources_per_node)node_idrSs r,get_total_obj_store_mem_on_noderUst %''3355G"|1JJLL++++RRR8PRR ,++ #G ,-B CCr.ceZdZdZdeefdZdZdZdZ dee fdZ d Z de ejejffd Zd Zde fd Zeejdefd ZdS)MemoryProfilera$A context manager that polls the USS of the current process. This class approximates the max USS by polling memory and subtracting the amount of shared memory from the resident set size (RSS). It's not a perfect estimate (it can underestimate, e.g., if you use Torch tensors), but estimating the USS is much cheaper than computing the actual USS. .. warning:: This class only works with Linux. If you use it on another platform, `estimate_max_uss` always returns ``None``. Example: .. testcode:: with MemoryProfiler(poll_interval_s=1.0) as profiler: for i in range(10): ... # Your code here print(f"Max USS: {profiler.estimate_max_uss()}") profiler.reset() poll_interval_sc||_tjtj|_d|_tj|_ d|_ d|_ dS)z Args: poll_interval_s: The interval to poll the USS of the process. If `None`, this class won't poll the USS. N) _poll_interval_spsutilProcessosgetpid_process_max_ussrcLock _max_uss_lock_uss_poll_thread_stop_uss_poll_event)r*rXs r,rbzMemoryProfiler.__init__sP!0ry{{33  &^-- $$(!!!r.cd|jdS)NzMemoryProfiler(poll_interval_s=r\)rZr>s r,rzMemoryProfiler.__repr__*sI1FIIIIr.c~|r(|j!|\|_|_|Sr0)_can_estimate_ussrZ_start_uss_poll_threadrcrdr>s r,rzMemoryProfiler.__enter__-sG  ! ! # # .(=(I++-- %) r.c@|j|dSdSr0)rc_stop_uss_poll_thread)r*rexc_valexc_tbs r,rzMemoryProfiler.__exit__6s+  ,  & & ( ( ( ( ( - ,r.rDc8|s |jJdS|j5|j||_n,t |j||_dddn #1swxYwY|jJ|jS)zGet an estimate of the max USS of the current process. Returns: An estimate of the max USS of the process in bytes, or ``None`` if an estimate isn't available. N)rgr`rb _estimate_ussrcr>s r,estimate_max_usszMemoryProfiler.estimate_max_uss:s%%'' =(((4   I I}$ $ 2 2 4 4 #DM43E3E3G3G H H  I I I I I I I I I I I I I I I }(((}sABBBcT|j5d|_ddddS#1swxYwYdSr0)rbr`r>s r,resetzMemoryProfiler.resetNsv   ! ! DM ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s !!cjJsJtjfd}tj|d}||fS)Nchsj5j_n,t j_dddn #1swxYwYjdSdSr0)rirbr`rnrcwaitrZ)r* stop_eventsr,poll_ussz7MemoryProfiler._start_uss_poll_thread..poll_ussXs '')) 7'QQ},(,(:(:(<(< (+DM4;M;M;O;O(P(P QQQQQQQQQQQQQQQ  5666 !'')) 7 7 7 7 7sAA77A;>A;T)rr)rZrgrcrdrr)r*rvthreadrus` @r,rhz%MemoryProfiler._start_uss_poll_threadRs$000%%'''''_&&  7 7 7 7 7 7!$??? z!!r.c||j4|j|jdSdSr0)rdrrcrr>s r,rjz$MemoryProfiler._stop_uss_poll_threadfsC  $ 0  % ) ) + + +  ! & & ( ( ( ( ( 1 0r.c~|sJ|j}|j|jz Sr0)rgr_ memory_inforssshared)r*rzs r,rnzMemoryProfiler._estimate_ussks?%%'''''m//11 !333r.c0tjdkS)NLinux)platformsystemr(r.r,rgz MemoryProfiler._can_estimate_ussss  G++r.N)r@rArBrCrrBrbrrrrdrorqrrcrrdrhrjrn staticmethod functoolscacherrgr(r.r,rWrWs".))))) JJJ)))(3-(!!!"i.> .O(P""""())) 4s4444_,t,,,_\,,,r.rWr.cTtttt|S)zUnzips a list of tuples into a tuple of lists Args: data: A list of tuples to unzip. Returns: A tuple of lists, where each list corresponds to one element of the tuples in the input list. )rmaprrI)rs r,unziprzs  T3:&& ' ''r.c|}fdg}g}t|j}|D]}||jdkrhdt jjd|d}||||<||||{||| |}|r| |}|S)zESort DataFrame by columns and rows, and also handle unhashable types.ct|ttjfrt fd|DSt|t r:t t fd|DS|S)Nc3.K|]}|VdSr0r()rr to_sortables r,rz0_sort_df..to_sortable..s+33AQ333333r.c38K|]\}}||fVdSr0r()rkvrs r,rz0_sort_df..to_sortable..s4JJ1KKNN 3JJJJJJr.)r1rrar%rdictsorteditems)rrs r,rz_sort_df..to_sortables a$ + , , 4333333333 3 a   LJJJJ JJJJJKK Kr.object __sort_proxy_r__)rN) copyrrNdtypeuuiduuid4hexrr2 sort_valuesdrop)r sort_cols temp_colsrNcoltemp_col sorted_dfrs @r,_sort_dfrs BIIRZ  G " " c7=H $ $Btz||'7AA#AAAHc7;;{33BxL   X & & &   X & & & &   S ! ! ! !y))I6NN9N55 r.actualexpectedc`t|t|krdSt|dkrdS tjt |dt |dddS#t $rYdSwxYw)aCheck if two DataFrames have the same rows. Unlike the built-in pandas equals method, this function ignores indices and the order of rows. This is useful for testing Ray Data because its interface doesn't usually guarantee the order of rows. FrT)r) check_dtype)rpdtestingassert_frame_equalr reset_indexAssertionError)rrs r, rows_samers 6{{c(mm##u 6{{at %% V   ( (d ( 3 3 X   * * * 5 5 &   t uusA$B B-,B-num_cpusnum_gpusmemoryray_remote_argscX|}|||d<|||d<|||d<|S)azConvert the given resources to Ray remote args. Args: num_cpus: The number of CPUs to be added to the Ray remote args. num_gpus: The number of GPUs to be added to the Ray remote args. memory: The memory to be added to the Ray remote args. ray_remote_args: The Ray remote args to be merged. Returns: The converted arguments. Nrrr)r)rrrrs r,"merge_resources_to_ray_remote_argsrsK"&**,,O&. #&. # $*! r.cddl}d} |j|j}nB#tt f$r.ddl}|j|j}|r|dddkrd}YnwxYw|S)NrrWsnappy) rGCodecdetectrrfrCrrsuffix)rr:rrrs r,infer_compressionrsK #hood++0  "###d##*  #fQRRjH,,"K # s(rss     $$$$$$$$$$                             ------UUUUUUUUUU------  8MMM::::::AAAAAAJJJJJJ 76666666777777  8 $ $  Sj Sj 688 4z) * #*###B%$&&  j    666LP"26 $t)t)t)#C=t) t)"*%0F*G!H t) sm t) ./ t) t) 3Xc] "#t)t)t)t)n*:!;D(((((   & #      (       E#tCy.1d,#Q!Q!Q!Q! Q!  Q! } Q!Q! Q!Q!Q!Q!j$)-%)"  (((!#(c](sm ((((V---*&&&487;OS k&k&k&!(3-0k&eC!223 4k&%U38_eCcM6J JKL k&  k&k&k&k&\        6# 6 6 6 6QQ -.QQQQ Q Z Q) Q -. Q Q Q QJ 4 5JJJJJ;!%&?!@; h;;;;4x( h( ;k" ; h ; ; ; ;G4 4 5G4 5e$ %G4G4 G4G4G4G4T?c?c????" GCLL GCLL:::::%:::B Z Z A;Z (1+ +,Z Z  Z  Z q$} Z Z Z Z z! ! ! ! ! ! ! ! H#####7:2###Li i i i i '*">i i i `"& /"/"/"r8|,/"/" DI  /"  /"  /" /"/"/"/"d33(,'+'+ 303030}30 }30 } 30  8C=(3- '( 30303030lT*3*4****  D D D D Du,u,u,u,u,u,u,u,p (U38_% (%S 3*? ( ( ( (  ",    Fblblt0smsm SM#s(^   #s(^ 6CHSMr.