o $it @sddlmZmZddlmZddlmZmZddlm Z m Z m Z m Z ddl mZdeededeefd d Zd e d eed edee ee ffddZdee deeed edeee eee ffddZdedeedee fddZdS))ListTuple) RefBundle)_calculate_blocks_rows_split_at_indices)Block BlockMetadataBlockPartition_take_first_non_empty_schema) ObjectRefper_split_bundlesowned_by_consumerreturncCs0t|dkr|Sdd|D}dd|D}tdd|D}|t|}t|||\}}}t||D]\} } tdd| D} | |ksFJ| | |ksNJq3tdd|D} t||| d } t| |}t|D]\}}|||td d||D} | |ksJqhg}|D] }| t||| d q|S) zEqualize split ref bundles into equal number of rows. Args: per_split_bundles: ref bundles to equalize. Returns: the equalized ref bundles. rcSsg|]}|jqS)blocks.0bundlerrX/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/_internal/equalize.py sz_equalize..cSg|]}t|qSr)r)rsplitrrrrscSrr)sum)r blocks_rowsrrrrscSg|]\}}|jqSrnum_rowsr_metarrrr*css|]}|jVqdS)N)schemarrrr 0sz_equalize..) owns_blocksr!cSrrrrrrrr9r ) lenr_shave_all_splitszipr r_split_leftovers enumerateextendappend)r r per_split_blocks_with_metadataper_split_num_rows total_rowstarget_split_size shaved_splitsper_split_needed_rows leftovers shaved_splitsplit_needed_rownum_shaved_rowsr!leftover_bundleleftover_splitsileftover_splitequalized_ref_bundlesrrrr _equalizes8      r:rnum_rows_per_block target_sizec Cs\g}g}d}t||D]\}}|||kr||||7}q ||q ||}|||fS)aShave a block list to the target size. Args: split: the block list to shave. num_rows_per_block: num rows for each block in the list. target_size: the upper bound target size of the shaved list. Returns: A tuple of: - shaved block list. - num of rows needed for the block list to meet the target size. - leftover blocks. r)r&r*) rr;r<shavedr1 shaved_rowsblock_with_meta block_rowsnum_rows_neededrrr_shave_one_splitEs     rB input_splitsr,c CsZg}g}g}t||D]\}}t|||\}} } |||| || q |||fS)aShave all block list to the target size. Args: input_splits: all block list to shave. input_splits: num rows (per block) for each block list. target_size: the upper bound target size of the shaved lists. Returns: A tuple of: - all shaved block list. - num of rows needed for the block list to meet the target size. - leftover blocks. )r&rBr*r)) rCr,r<r/r0r1rr;r=rA _leftoversrrrr%cs    r%r1r0cCsbt|}g}d}t|D]\}}|||||}q t|j||j}ddt|Dd|S)z0Split leftover blocks by the num of rows needed.rcSsg|] \}}tt||qSr)listr&)r block_refsrrrrrsz$_split_leftovers..N)r$r(r*rrr#r&)r1r0 num_splits split_indicesprevr7rA split_resultrrrr's r'N)typingrr'ray.data._internal.execution.interfacesrray.data._internal.splitrrray.data.blockrrr r ray.typesr boolr:intrBr%r'rrrrsJ   7