&`i+ddlZddlZddlmZmZmZmZddlZddlm Z ddl m Z ddl m Z mZmZmZmZddlmZejeZdedeefd Zd eed eedeefd Zd eed eedeeefd Zdede ded eedeeeeeefe fdff dZdeededeefdZdeeee efdeeededeeee effdZdeeee efdeedeeeee eeeffdZ d!deeee efdeededeedeeeee eeeff dZ de defd Z!dS)"N)IterableListTupleUnion)trace_deallocation)cached_remote_fn)Block BlockAccessorBlockExecStats BlockMetadataBlockPartition) ObjectRefblocks_with_metadatareturnctt}g}|D]W\}}|j/tj||}||_n|j}||X|S)z@Calculate the number of rows for a list of blocks with metadata.)r _get_num_rowsnum_rowsraygetremoteappend)r get_num_rows block_rowsblockmetadatars l/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/split.py_calculate_blocks_rowsrs$M22LJ/$$x   $w|22599::H (H  (H(#### num_rows_per_block split_indicesc>t|fd|DS)zTGenerate valid split indices by apply min(index, total_num_rows) to every index.c0g|]}t|S)min).0index total_rowss r z+_generate_valid_indices..,s# > > >uCz " " > > >r)sum)rr r's @r_generate_valid_indicesr*%s- '((J > > > > > > >>rcg}d}g}d}d}|t|krq||}||}||z |kr|||z |dz }J||g}|||z }|dz }|t|kqt|t|kr7||g}t|t|k7|S)a:Given num rows per block and valid split indices, generate per block split indices. Args: num_rows_per_block: num of rows per block. split_indices: The (global) indices at which to split the blocks. Returns: Per block split indices indicates each input block's split point(s). r)lenr) rr per_block_split_indicescurrent_input_block_idcurrent_block_split_indicescurrent_block_global_offsetcurrent_index_id split_indexcurrent_block_rows r!_generate_per_block_split_indicesr5/s= !"$"# S// / /#$45 ./EF 4 48I I I ' . .99     !  &&'BCCC&(##'9:P'QQ#!# S// / / % & &-?)@)@ @ @&&'BCCC&(# % & &-?)@)@ @ @ #"rblock_idrmeta.ctg}g}tj|}d}||j|D]}td|d|t j} |||} tj| } t| | |j | } || || |}||fg} | |t| S)aPSplit the provided block at the given indices. Args: block_id: the id of this block in the block list. block: block to be split. meta: metadata of the block, we expect meta.num is valid. split_indices: the indices where the block should be split. Returns: returns block_id, split blocks metadata, and a list of blocks in the following form. We return blocks in this way so that the owner of blocks could be the caller(driver) instead of worker itself. Tuple(block_id, split_blocks_meta), block0, block1 ... rzslicing block :)r size_bytes input_files exec_stats)r for_blockrrloggerdebugr builderslicer r:r;buildextendtuple)r6rr7r split_meta split_blocksblock_accessor prev_indexr&stats split_blockaccessor_metaresultss r_split_single_blockrN[s?(JL",U33NJ'''   :j::5::;;;&(($**:u==  *;77&&((**,,({{}}     %   K((( *%&G NN<   >>rblock_split_indicesrcnd}g}|D]-}|dks||kr||kr|||}.|S)zdrop split indices that creates empty block split. This could happen when there are duplicated indices, or index equal to 0 (start of the block) or num_block_rows (end of the block). r)r)rOrrHoptimized_indicesr&s r_drop_empty_block_splitrSsc J$ A::(**  J     ''' rr.owned_by_consumercztt}dgt|z}g}g}g}t|D]\}} ||\} } | j} t | | } t| dkr | | fg||<C|ddt| z|| | | } || d|| dd|| |rbtj |}t||D]=\\}} }t| t|ksJt|| ||<>|r|D]}t|dn|D]}t|dd tj|S) z5Split all the input blocks based on the split indicesNrSPREAD)scheduling_strategy num_returnsr,zsplit._split_all_blocksF)free)rrNr- enumeraterrSoptionsrrrrzipr itertoolschain from_iterable)rr.rTsplit_single_blockall_blocks_split_results per_block_split_metadata_futuresper_block_split_block_refsblocks_splittedr6rO block_refr7 block_row object_refsper_block_split_metadata block_refsbs r_split_all_blocksrls9 **=>>6:VcBV>W>W5W')$!#O)23J)K)K..%%0:DM 56I9UU " # #q ( (4=d2C1D $X . .-44$,!cBU>V>V:V5f#   - 3 3KN C C C & - -k!""o > > >  " "9 - - - -'G#&7+K#L#L ,/ $&@- -  G G ( Xtjt99J////14Z1F1F $X . . I  = =A q"; < < < < =! I IA q";% H H H H H ? ( ()A B BBrrbglobal_split_sizescg}g}g}g}d}d}|t|kr|||krD|||ksJ||||g}g}d}|dz }nFt|\}} |||| || jz }|t|k||fS)z > Y  ! !) , , ,    % % % $- /  S!344 4 4 , &&rTindicesrct|}t|dkr,ggt|dzzggt|dzzfS|t|}t||}t ||}t |||}dg|zt |gzfdtdtD}t||S)a!Split blocks at the provided indices. Args: blocks_with_metadata: Block futures to split, including the associated metadata. indices: The (global) indices at which to split the blocks. owned_by_consumer: Whether the provided blocks are owned by the consumer. block_rows: The number of rows for each block, in case it has already been computed. Returns: The block split futures and their metadata. If an index split is empty, the corresponding block split will be empty . rr,Nc8g|]}||dz z S)r,r#)r%ihelpers rr(z%_split_at_indices..!s*LLL6!9va!e},LLLr) listr-rr*r5rlr)rangerv) rrwrTr valid_indicesr.rb split_sizesr{s @r_split_at_indicesrs, 455   A%%G q()B43w<intrr*r5rNrSboolrlrvrrr#rrrs//////////// @@@@@@999999   8 $ $( #Y"?S ?9? #Y????)#S )#9)# $s)_)#)#)#)#X++ + +9 +  5sD//0%7 8# => ++++\ccdSVi"9CuYu%5}%DEF9C!$s)_9C9CeIe$m345 9C9C9C9Cx'&uYu-=}-L'MN'S ' 4Yu%& 'd=.A)B BC''''F# -Q-QuYu%5}%DEF-Q #Y-Q-QS -Q  4Yu%& 'd=.A)B BC -Q-Q-Q-Q`553555555r