o $iNS@sddlZddlZddlmZmZmZmZmZmZm Z m Z m Z m Z m Z ddlZddlmZddlmZddlmZddlmZddlmZmZmZmZddlmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'errdd l(m)Z)dd l*m+Z+e d Z,ed e'Z-GdddeZ.GdddeZ/dS)N) TYPE_CHECKINGAnyDictIteratorListMappingOptionalSequenceTupleTypeVarUnion) env_integer)TENSOR_COLUMN_NAME) BlockBuilder) SizeEstimator) NULL_SENTINELfind_partition_indexis_nan keys_equal)Block BlockAccessorBlockColumnAccessorBlockExecStatsBlockMetadataWithSchema BlockTypeKeyTypeU)DEFAULT_TARGET_MAX_BLOCK_SIZE)SortKey) AggregateFnT#RAY_DATA_MAX_UNCOMPACTED_SIZE_BYTESc@seZdZddZdeeeejfddfddZ de ddfd d Z e d e eee fdefd d Ze deedefddZe de fddZe defddZdefddZdefddZdefddZdefddZdddZdS) TableBlockBuildercCsBtt|_g|_d|_d|_t|_d|_ d|_ d|_ ||_ dSNr) collections defaultdictlist_columns_tables_tables_size_cursor_tables_size_bytesr_uncompacted_size _num_rows_num_uncompacted_rows_num_compactions _block_type)self block_typer2[/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/_internal/table_block.py__init__7s  zTableBlockBuilder.__init__itemreturnNcCst|dr |}n t|tjrt|i}t|tjjs%t d |t ||D]}||j vr7dg|j |j |<q'|j D]}||}|j ||q;|jd7_|j d7_ ||j|dS)N as_pydictzLReturned elements of an TableBlock must be of type `dict`, got {} (type {}).)hasattrr7 isinstancenpndarrayrr$abcr ValueErrorformattyper'r-getappendr,_compact_if_neededr+add)r0r5 column_namevaluer2r2r3rDLs(       zTableBlockBuilder.addblockcCsXt||jstdt|d|jd|t|}|j||j| 7_dS)NzGot a block of type z , expected z[.If you are mapping a function, ensure it returns an object with the expected type. Block: ) r:r/ TypeErrorr@r for_blockr(rBr,num_rows)r0rGaccessorr2r2r3 add_blockes   zTableBlockBuilder.add_blockcolumnscCtNNotImplementedError)rMr2r2r3_table_from_pydictqz$TableBlockBuilder._table_from_pydicttablescCrNrOrP)rTr2r2r3_combine_tablesurSz!TableBlockBuilder._combine_tablescCrNrOrPr2r2r2r3 _empty_tableyrSzTableBlockBuilder._empty_tablecCrNrOrPr2r2r2r3_concat_would_copy}rSz$TableBlockBuilder._concat_would_copycCs |jrdS|ot|jdkS)NTr8)r'rWlenr(r0r2r2r3will_build_yield_copysz'TableBlockBuilder.will_build_yield_copycCsD|jr ||jg}ng}||jt|dkr|S||Sr#)r'rRextendr(rXrVrU)r0rTr2r2r3builds   zTableBlockBuilder.buildcC|jSrO)r,rYr2r2r3rJzTableBlockBuilder.num_rowscCsX|jdkrdS|j|jdD]}|jt|7_qt|j|_|j|jSr#) r,r(r)r*rrI size_bytesrXr+r0tabler2r2r3get_estimated_memory_usages  z,TableBlockBuilder.get_estimated_memory_usagecCs\|jsJ|jtkrdS||j}||t|_|j|jd7_d|_ dS)Nr8r) r'r+r_MAX_UNCOMPACTED_SIZE_BYTESrRrLrclearr.r-)r0rGr2r2r3rCs     z$TableBlockBuilder._compact_if_needed)r6N)__name__ __module__ __qualname__r4r dictrr;r<rDrrL staticmethodrstrrrrRrUrVboolrWrZr\intrJrbrCr2r2r2r3r"6s"  r"c@sTeZdZdefddZeddZededede j fd d Z de fd d Z deefd dZdedede fddZde fddZdeddfddZdPddZedefddZdedd defd!d"Zdedd defd#d$ZdQd&ed'edeefd(d)Zd&ed'edeefd*d+Zd&ed'edeefd,d-Zd&ed'edeefd.d/Zd&ed'edeefd0d1Z 2dRd&ed'ed3eedeefd4d5Z!d6efd7d8Z"dd d9e#d:de fd;d<Z$e% =dSd>ee dd d9e#d:d?ede#e d@ff dAdBZ&dCee#edd fdDdEZ'e% 2dRd>ee dFee(dee fdGdHZ)e%dIe dJe(fdKdLZ*dMe de fdNdOZ+d2S)TTableBlockAccessorracCs ||_dSrO_tabler`r2r2r3r4s zTableBlockAccessor.__init__cCs|d|dS)N_r8r2)namecountr2r2r3_munge_conflictsz"TableBlockAccessor._munge_conflictrowrow_idxr6cCrNrOrP)rtrur2r2r3_build_tensor_rowrSz$TableBlockAccessor._build_tensor_rowcCs |}|SrO) to_pandas)r0defaultr2r2r3 to_defaultszTableBlockAccessor.to_defaultcCrNrOrPrYr2r2r3 column_nameszTableBlockAccessor.column_namesrqrFcCrNrOrP)r0rqrFr2r2r3 fill_columnr{zTableBlockAccessor.fill_columncCr]rOrnrYr2r2r3to_blockr^zTableBlockAccessor.to_blockaccrcCrNrOrP)r0r~r2r2r3_zipr{zTableBlockAccessor._zipothercCst|}t|t|s4t|tr(t|tr(t|j|g\}}t||Std t|t|| | krHtd | | | |S)Nz#Cannot zip {} with block of type {}z3Cannot zip self (length {}) with block of length {}) rrIr:r@rmnormalize_block_typesrozipr>r?rJr)r0rr~ self_norm other_normr2r2r3rs*     zTableBlockAccessor.zipcCrNrOrPr2r2r2r3rVrSzTableBlockAccessor._empty_table n_samplessort_keyrcCrNrOrP)r0rrr2r2r3_sampler{zTableBlockAccessor._samplecCsL|dust|rtd||dkr|St||}|||S)Nz+Table sort key must be a column name, was: r)callablerQrJrVminr)r0rrkr2r2r3samples  zTableBlockAccessor.sampleFon ignore_nullscCst|j|}|j|dSN)r)r for_columnrorrr0rrrKr2r2r3rrs zTableBlockAccessor.countcC&||t|j|}|j|dSr)_validate_columnrrrosumrr2r2r3r  zTableBlockAccessor.sumcCrr)rrrrorrr2r2r3rrzTableBlockAccessor.mincCrr)rrrromaxrr2r2r3rrzTableBlockAccessor.maxcCrr)rrrromeanrr2r2r3r rzTableBlockAccessor.meanNrcCs&||t|j|}|j|dSr)rrrrosum_of_squared_diffs_from_mean)r0rrrrKr2r2r3rs  z1TableBlockAccessor.sum_of_squared_diffs_from_meancolcCs@|dur td|d||vrtd|d|dS)Nz-Provided `on` value has to be non-null (got 'z')zReferencing column 'z' not present in the schema: )r>rzschema)r0rr2r2r3rs z#TableBlockAccessor._validate_columnaggsrcs<|dtttttfffdd }}|D]|\}}|t|dkr-|dfdd|D}tt|D]}t |} | dkrU|| |||||<q.iter_groupsr8rcsg|]}|qSr2)init).0agg) init_valsr2r3 [sz1TableBlockAccessor._aggregate..) get_columnsrr r rrbuilderrXrangerrIrJaccumulate_blockrr$r%rlrqrsrDr\)r0rrrr group_keys group_view accumulatorsirKrtrgkrrr accumulatorrqr2)rrr0r3 _aggregate&s8$        zTableBlockAccessor._aggregateTblocksfinalizercst|}t}|fddfdd}tjdd|Dd|idt|d } zdur;t }fd d } d } dgt |} dgt |} | D]S} | rt t }tt |D]'}||j}||d krt|||}||d 7<|| |<| || |<qld} q]tt |D]}||| || | || |<qq]i}rt|D]\}}|||<qt|| | D]\}}}|r||||<q|||<q||Wn tyYnwq2|}|tj||dfS)aXCombine previously aggregated blocks. This assumes blocks are already sorted by key in ascending order, so we can do merge sort to get all the rows with the same key. Args: blocks: A list of partially combined and sorted blocks. sort_key: The column name of key or None for global aggregation. aggs: The aggregations to do. finalize: Whether to finalize the aggregation. This is used as an optimization for cases where we repeatedly combine partially aggregated groups. Returns: A block of [k, v_1, ..., v_n] columns and its metadata where k is the groupby key and v_i is the corresponding aggregation result for the ith given aggregation. If key is None then the k column is omitted. csrt|SdS)N)rr)r)rr2r3_key_fns z>TableBlockAccessor._combine_aggregated_blocks.._key_fncs|}tdd|DS)NcSs$g|]}|dus t|rtn|qSrO)rr)rvr2r2r3rs$zVTableBlockAccessor._combine_aggregated_blocks..safe_key_fn..r)rvalues)rr2r3 safe_key_fns zBTableBlockAccessor._combine_aggregated_blocks..safe_key_fncSsg|] }t|jddqS)Fr)rrIrrrGr2r2r3rszATableBlockAccessor._combine_aggregated_blocks..keyNrTc3sRtr'VztWn tydYdSwtsdSdSrO)rrrr2)rrrrr2r3gens  z:TableBlockAccessor._combine_aggregated_blocks..genr8F)stats)rmrrrrheapqmergerrIrrXr$r%rlrrqrsrrrDrr\r from_block)clsrrrrrrrnext_key_columnsrfirstrresolved_agg_namesrrrrrqrtcol_namenext_keyragg_namerretr2)rrrrrr3_combine_aggregated_blockswsv            =z-TableBlockAccessor._combine_aggregated_blocks boundariescsVg}fdd|D}d}|D]}|j|||}q|j|d|S)Ncsg|] }tj|qSr2)rro)rboundaryr0rr2r3rs z>TableBlockAccessor._find_partitions_sorted..r)rBro)r0rr partitionsboundslast_idxidxr2rr3_find_partitions_sorteds z*TableBlockAccessor._find_partitions_sortedtarget_block_typecstt}|D] }t|}t|tstdt|d|| d7<qt |dkr=dus;gt | kr=|SdurPt |ddddd \}fd d |Dtfd d Drrtddd DdS)aNormalize input blocks to the specified `normalize_type`. If the blocks are already all of the same type, returns original blocks. Args: blocks: A list of TableBlocks to be normalized. target_block_type: The type to normalize the blocks to. If None, Ray Data chooses a type to minimize the amount of data conversions. Returns: A list of blocks of the same type. zWBlock type normalization is only supported for TableBlock, but received block of type: .r8NcSs|dS)Nr8r2)xr2r2r32sz:TableBlockAccessor.normalize_block_types..T)rreversercsg|]}|qSr2)try_convert_block_typer)rrr2r3r6s z.c3s$|] }t|td VqdS)rN)r:r@r)resultsr2r3 :s"z;TableBlockAccessor.normalize_block_types..zYExpected all blocks to be of the same type after normalization, but got different types: cSsg|]}t|qSr2)r@)rbr2r2r3r=szP. Try using blocks of the same type to avoid the issue with block normalization.)r$r%rlrrIr:rmr>r@r1rXr&rsorteditemsany)rrr seen_typesrGblock_accessorrpr2)rrrr3r s>      z(TableBlockAccessor.normalize_block_typesrGr1cCs>|tjkr t|S|tjkrt|St|SrO)rARROWrrIto_arrowPANDASrwry)rrGr1r2r2r3rCs  z)TableBlockAccessor.try_convert_block_type other_blockcCrN)aCombine this table with another table horizontally (column-wise). This will append the columns. Args: other_block: The table to hstack side-by-side with. Returns: A new table with columns from both tables combined. rP)r0rr2r2r3hstackLs zTableBlockAccessor.hstack)rrr6r)FrO)T),rerfrgrr4rirsrrlr;r<rvrryrrjrzr|r}rrrrVrrrkrrrrrrrrrrr r classmethodrrrrrrr2r2r2r3rms~      Q  z   7rm)0r$rtypingrrrrrrrr r r r numpyr;ray._private.ray_constantsr ray.air.constantsr ray.data._internal.block_builderr!ray.data._internal.size_estimatorrray.data._internal.utilrrrrray.data.blockrrrrrrrrray.data.contextr2ray.data._internal.planner.exchange.sort_task_specrray.data.aggregaterr rcr"rmr2r2r2r3s(4    (   u