o bi^ @sddlmZmZddlmZddlmZmZddlm Z ddl m Z m Z m Z ddlmZdeeded eefd d Zd e d eeded ee ee ffddZdee deeeded eee eee ffddZdedeed ee fddZdS))ListTuple) RefBundle)_calculate_blocks_rows_split_at_indices)unify_ref_bundles_schema)Block BlockMetadataBlockPartition) ObjectRefper_split_bundlesowned_by_consumerreturncCs&t|dkr|Sdd|D}dd|D}tdd|D}|t|}t|||\}}}t||D]\} } tdd| D} | |ksFJ| | |ksNJq3t|} t||| d} t| |}t|D]\}}|||tdd||D} | |ksJqcg}|D] }| t||| dq|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).0bundlerrO/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/equalize.py sz_equalize..cSg|]}t|qSr)r)rsplitrrrrscSrr)sum)r blocks_rowsrrrrscSg|]\}}|jqSrnum_rowsr_metarrrr&) owns_blocksschemacSrrrrrrrr5r) lenr_shave_all_splitsziprr_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 _equalize s8      r8rnum_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() rr9r:shavedr/ shaved_rowsblock_with_meta block_rowsnum_rows_neededrrr_shave_one_splitAs     r@ 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$r@r(r') rAr*r:r-r.r/rr9r;r? _leftoversrrrr#_s    r#r/r.cCsbt|}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$)r/r. num_splits split_indicesprevr5r? split_resultrrrr%s r%N)typingrr'ray.data._internal.execution.interfacesrray.data._internal.splitrrray.data._internal.utilrray.data.blockrr r ray.typesr boolr8intr@r#r%rrrrsL    7