tfddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl ZddlmZddlmZdd lmZmZmZmZdd lmZdd lmZdd lmZmZmZm Z Gd dZ!GddeZ"Gdde"Z#Gdde"Z$GddeZ%d$dZ&dZ'dZ(GddeZ)Gdde)Z*GddeZ+Gd d!eZ,Gd"d#eZ-y)% annotationsN) defaultdict)Callable)product)Any)map)tokenize) Blockwise BlockwiseDepBlockwiseDepDictblockwise_token)flatten)Layer)apply cached_cumsumconcreteinsertceZdZdZdZdZy)CallableLazyImportzFunction Wrapper for Lazy Importing. This Class should only be used when materializing a graph on a distributed scheduler. c||_yN) function_path)selfrs U/var/www/html/software/conda/envs/higlass/lib/python3.12/site-packages/dask/layers.py__init__zCallableLazyImport.__init__"s *c>ddlm}||j|i|S)Nr) import_term)distributed.utilsrr)rargskwargsrs r__call__zCallableLazyImport.__call__%s"1.{4--.???rN)__name__ __module__ __qualname____doc__rr#rrrrs +@rrcDeZdZUdZded<ded<dZded<d d Zd d Zy )ArrayBlockwiseDepzg Blockwise dep for array-likes, which only needs chunking information to compute its data. tuple[tuple[int, ...], ...]chunkstuple[int, ...] numblocksFboolproduces_taskscN||_td|D|_d|_y)Nc32K|]}t|ywr)len).0chunks r z-ArrayBlockwiseDep.__init__..>s>es5z>sF)r,tupler.r0)rr,s rrzArrayBlockwiseDep.__init__<s# >v>>#rctd)Nz%Subclasses must implement __getitem__)NotImplementedErrorridxs r __getitem__zArrayBlockwiseDep.__getitem__As!"IJJrNr,r+r;r-)r$r%r&r'__annotations__r0rr<r(rrr*r*2s* (' ND $ Krr*ceZdZdZddZy)ArrayChunkShapeDepz(Produce chunk shapes given a chunk indexcNtdt||jDS)Nc3,K|] \}}||ywrr()r4ir5s rr6z1ArrayChunkShapeDep.__getitem__..IsD(!UU1XDs)r7zipr,r:s rr<zArrayChunkShapeDep.__getitem__HsDc#t{{.CDDDrNr>)r$r%r&r'r<r(rrrArAEs 2ErrAc8eZdZUdZded<dfd ZddZxZS) ArraySliceDepz>Produce slice(s) into the full-sized array given a chunk indexr+startscRt||td|D|_y)Nc36K|]}t|dyw)T) initial_zeroN)r)r4cs rr6z)ArraySliceDep.__init__..SsPAM!$??P)superrr7rH)rr, __class__s rrzArraySliceDep.__init__Qs"  PPP rcrtdt||jD}td|DS)Nc3<K|]\}}||||dzfyw)Nr()r4rDstarts rr6z,ArraySliceDep.__getitem__..Vs%SEU1XuQU|,Sc36K|]}tg|dywrslice)r4ss rr6z,ArraySliceDep.__getitem__..Ws2U_A_t_2rM)r7rErH)rr;locs rr<zArraySliceDep.__getitem__Us.SSdkk=RSS2c222rr=)r;r7)r$r%r&r'r?rr< __classcell__rOs@rrGrGLsH ''Q3rrGcdeZdZdZfdZdZedZdZdZ dZ dZ d Z d Z d d ZxZS) ArrayOverlapLayera\Simple HighLevelGraph array overlap layer. Lazily computed High-level graph layer for a array overlap operations. Parameters ---------- name : str Name of new output overlap array. array : Dask array axes: Mapping Axes dictionary indicating overlap in each dimension, e.g. ``{'0': 1, '1': 1}`` cvt|||_||_||_||_||_d|_yr)rNrnameaxesr,r.token _cached_keys)rr_r`r,r.rarOs rrzArrayOverlapLayer.__init__is;    "  rc"d|jdS)NzArrayOverlapLayer.keyssy d)CQw#i.(9>y~9NOA4'D.A4/OOM9>in8MN1$.NNMPNs A-A2)rbr_r,r.)rrr,rr_r.s @@@@r _dask_keyszArrayOverlapLayer._dask_keyssR    ($$ $"&))T[[$..fi &*V+F rc|j}|j}|j}|j}d|jz}d|jz}|r t d}nddlm}ttt|} tjt| |} tj|t t| tt tj"t} i} i} | D]N}t%|f|z|}|f|z|k7r || |f|z<&|f|z| |f|z<|t&| d|z|ff| |f|z<Ptj(| | }|S) z/Construct graph for a simple overlap operation.zgetitem-zoverlap-zdask.array.core.concatenate3r) concatenate3)dimsr`rre)r`r,r_rrardask.array.corerlistr r3 functoolspartial_expand_keys_around_centertoolzpiperconcatfractional_slicermerge)r deserializingr`r,r_ dask_keys getitem_name overlap_namerr expand_key2 interior_keysinterior_slicesoverlap_blocksk frac_sliceros rrmz"ArrayOverlapLayer._construct_graphsOyyyyOO% !DJJ. !DJJ. ..LML 5CV$%'' &T  wK 0#g, d  A)4'A+t: rF)r$r%r&r'rrgpropertyrpr<rzr~rrrrmrZr[s@rr]r]ZsI ! 7!! -&)rr]c fd}g}t|ddD]2\}}d}|dkDr|dz }||dz kr|dz }|j|4t|ddDcgc]-\}}t|j|dfr |||n|g/} }}||gg| z} t t | } t|D cgc]%\}} t|j|dfr| nd'} }} t | | } | Scc}}wcc} }w)aGet all neighboring keys around center Parameters ---------- k: Key The key around which to generate new keys dims: Sequence[int] The number of chunks in each dimension name: Option[str] The name to include in the output keys, or none to include no name axes: Dict[int, int] The axes active in the expansion. We don't expand on non-active axes Examples -------- >>> _expand_keys_around_center(('x', 2, 3), dims=[5, 5], name='y', axes={0: 1, 1: 1}) # noqa: E501 # doctest: +NORMALIZE_WHITESPACE [[('y', 1.1, 2.1), ('y', 1.1, 3), ('y', 1.1, 3.9)], [('y', 2, 2.1), ('y', 2, 3), ('y', 2, 3.9)], [('y', 2.9, 2.1), ('y', 2.9, 3), ('y', 2.9, 3.9)]] >>> _expand_keys_around_center(('x', 0, 4), dims=[5, 5], name='y', axes={0: 1, 1: 1}) # noqa: E501 # doctest: +NORMALIZE_WHITESPACE [[('y', 0, 3.1), ('y', 0, 4)], [('y', 0.9, 3.1), ('y', 0.9, 4)]] cg}|dz dkDr|j|dz |j||dz|dz kr|j|dz|S)Ng?rrR)append)rDrrvrs rindsz(_expand_keys_around_center..indssW  9q= IIcCi  # 9tAw{ " IIcCi  rrRNr) enumerateranygetrr reshapelist)rrr_r`rshaperDrnumr!seqdshape2rs ` rrrs-4 EAabE"3 7 1HC a1  1HC S IRRSTUTVRWHX >DaTXXa^-.Q SE9 D  x$ w~ C=Fu=M NTQ3A()aq0 NF N  %F M  Os 2C5:*C;ct|dk(r t|Stt||dz }tj||Dcgc]}t |dd|c}Scc}w)zgReshape iterator to nested shape >>> reshapelist((2, 3), range(6)) [[0, 1, 2], [3, 4, 5]] rRrN)r3rintr partitionr)rrnparts rrr s\  5zQCy C58# $9>C9PQ E!"It,QQQs A%cX|dftd|ddDz}g}tt|dd|ddD]\}\}}|j|d}t |tr |d}|d} n|}|} ||k(r|j t ddd\||kr| r|j t d| ||kDr|r|j t | d|j t ddt|}td|Dr|Stj||fS)a >>> fractional_slice(('x', 5.1), {0: 2}) (, ('x', 5), (slice(-2, None, None),)) >>> fractional_slice(('x', 3, 5.1), {0: 2, 1: 3}) (, ('x', 3, 5), (slice(None, None, None), slice(-3, None, None))) >>> fractional_slice(('x', 2.9, 5.1), {0: 2, 1: 3}) (, ('x', 3, 5), (slice(0, 2, None), slice(-3, None, None))) rc3DK|]}tt|ywr)rround)r4rDs rr6z#fractional_slice..$s A1U1X As rRNc3<K|]}|tdddk(ywrrV)r4rs rr6z#fractional_slice..:s ;c3%dD) ) ;rT) r7rrEr isinstancerrWalloperatorgetitem) taskr`roundedindexrDtrdepth left_depth right_depths rrrs'Awj5 AQR AAAG Es48WQR[9:& 6AqA eU #qJ(KJK 6 LLtT40 1 U{ LLq+. / Uz LL {D1 2 LLq! %!&" %LE ;U ;;   '511rceZdZdZ dfd ZdZdZdZdZe dZ dZ d Z d Z d Zdd Zd ZdZddZxZS)SimpleShuffleLayeraSimple HighLevelGraph Shuffle layer High-level graph layer for a simple shuffle operation in which each output partition depends on all input partitions. Parameters ---------- name : str Name of new shuffled output collection. column : str or list of str Column(s) to be used to map rows to output partitions (by hashing). npartitions : int Number of output partitions. npartitions_input : int Number of partitions in the original (un-shuffled) DataFrame. ignore_index: bool, default False Ignore index during shuffle. If ``True``, performance may improve, but index values will not be preserved. name_input : str Name of input collection. meta_input : pd.DataFrame-like object Empty metadata of input collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations c ||_||_||_||_||_||_||_|xs t||_d|jz|_ | xsi} d|_ d| vr|j| d<t |5| y)Nsplit-priorityr)r_column npartitionsnpartitions_input ignore_index name_input meta_inputr parts_out split_name _split_keys _key_priorityrNr) rr_rrrrrrrrrOs rrzSimpleShuffleLayer.__init__ds  &!2($$"8eK&8"TYY."'R  [ (&*&8&8K # [1rcNt|tsJ|d|jk(ryy)NrrR)rr7rrts rrz SimpleShuffleLayer._key_prioritys(#u%%% q6T__ $rcX|jDchc]}|j|fc}Scc}wrrr_rrs rrz"SimpleShuffleLayer.get_output_keys#.2nn=dD!==='cNdj|j|jS)Nz-SimpleShuffleLayer)formatr_rrfs rrgzSimpleShuffleLayer.__repr__s$>EE IIt''  rct|dSrrrfs rrz"SimpleShuffleLayer.is_materializedrrcxt|dr |jS|j}||_|jSrirkrns rrpzSimpleShuffleLayer._dictrqrc |j|Srrsrts rr<zSimpleShuffleLayer.__getitem__rvrc,t|jSrrxrfs rrzzSimpleShuffleLayer.__iter__r{rc,t|jSrr}rfs rr~zSimpleShuffleLayer.__len__rrct}|D])} |\}}||jk7r|j|+|S#t$rY9wxYwz4Simple utility to convert keys to partition indices.set ValueErrorr_addrrpartsru_name_parts r_keys_to_partsz!SimpleShuffleLayer._keys_to_parts^ C " u ! IIe     ; AAc tt}|xs|j|}|D]H}||j|fxxt |j Dchc]}|j |fc}zcc<J|Scc}w)zDetermine the necessary dependencies to produce `keys`. For a simple shuffle, output partitions always depend on all input partitions. This method does not require graph materialization. )rrrr_rrr)rrrdepsrrDs r_cull_dependenciesz%SimpleShuffleLayer._cull_dependenciess~3:!4!4T!:  D $))T" #.3D4J4J.K()*!$(  #  (sA4 c t|j|j|j|j|j |j |j|SNr)rr_rrrrrrrrs r_cullzSimpleShuffleLayer._cullsI! II KK     " "    OO OO  rc|j|}|j||}|t|jk7r|j |}||fS||fS)aCull a SimpleShuffleLayer HighLevelGraph layer. The underlying graph will only include the necessary tasks to produce the keys (indices) included in `parts_out`. Therefore, "culling" the layer only requires us to reset this parameter. rrrrrrrrall_keysr culled_deps culled_layers rcullzSimpleShuffleLayer.cull^''- --di-H DNN+ +::i0L, ,$ $rc @d|jz}|rtd}td}n ddlm}ddlm}i}|j D]}t|jDcgc]}|j||f}}|||jf||j|f<|D]\} } } tj|| f| f||j| | f<|| f|vs3||j| f|jd|j|j|j|jf||| f<|Scc}w)z/Construct graph for a simple shuffle operation.group-dask.dataframe.core._concat$dask.dataframe.shuffle.shuffle_groupr_concat shuffle_group)r_rdask.dataframe.corer dask.dataframe.shuffler rrrrrrrrrr) rrshuffle_group_name concat_funcshuffle_group_funcropart_outpart_in _concat_list_ _part_out_part_ins rrmz#SimpleShuffleLayer._construct_graphsZ& 1 --JKK!36"  C R H %T%;%;<(G4L !!*CH% & +7 &9h$$'2?T__i:; '1<*(3 (((())(( ;C+X67  8 7sDNNrr)r$r%r&r'rrrrgrrrpr<rzr~rrrrrmrZr[s@rrrGsgJ(2T> -!!     % .rrcDeZdZdZ dfd ZdZddZdZd dZxZ S) ShuffleLayera"Shuffle-stage HighLevelGraph layer High-level graph layer corresponding to a single stage of a multi-stage inter-partition shuffle operation. Stage: (shuffle-group) -> (shuffle-split) -> (shuffle-join) Parameters ---------- name : str Name of new (partially) shuffled collection. column : str or list of str Column(s) to be used to map rows to output partitions (by hashing). inputs : list of tuples Each tuple dictates the data movement for a specific partition. stage : int Index of the current shuffle stage. npartitions : int Number of output partitions for the full (multi-stage) shuffle. npartitions_input : int Number of partitions in the original (un-shuffled) DataFrame. k : int A partition is split into this many groups during each stage. ignore_index: bool, default False Ignore index during shuffle. If ``True``, performance may improve, but index values will not be preserved. name_input : str Name of input collection. meta_input : pd.DataFrame-like object Empty metadata of input collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations c ||_||_||_t ||||||| | | xst t ||  y)N)rr)inputsstagensplitsrNrrr3)rr_rrrrrrrrrrrrOs rrzShuffleLayer.__init__AsU           55V#5#  rczdj|j|j|j|jS)Nz=ShuffleLayer)rr_rrrrfs rrgzShuffleLayer.__repr___s0NUU IItzz4<<1A1A  rc8tt}|xs|j|}t|jDcic]\}}|| }}}|D]}|j|}t |j D]} t||j| } || } |jdk(r@| |jk\r1||j|fjd|jz| dfn||j|fj|j| f|Scc}}w)zqDetermine the necessary dependencies to produce `keys`. Does not require graph materialization. rrempty) rrrrrrrrrrr_rr) rrrrrDinp inp_part_maproutr_inprs rrzShuffleLayer._cull_dependenciesds 3:!4!4T!: -6t{{-CD61cQD D JD++d#C4<<( Jc4::q1$T*::?u0F0F'F$))T*+//DII1EtW0UV$))T*+//%0HI  J J Es Dc t|j|j|j|j|j |j |j|j|j|j| Sr) rr_rrrrrrrrrrs rrzShuffleLayer._cullws^ II KK KK JJ     " " LL    OO OO  rc d|jz}|rtd}td}n ddlm}ddlm}i}t |jDcic]\}}|| }}}|jD]g} |j| } g} t|jD]F}t| |j|} | |j} | j|j| | fH|| |jf||j| f<| D]\}} } t j"|| f| f||j| | f<|| f|vs3|| }|jdk(r3||j$kr|j&|f}n#|| df}|j(||<n|j&|f}|||j*|j|j|j$|j|j,f||| f<j|Scc}}w)z2Construct graph for a "rearrange-by-column" stage.rrrrrr r )r_rr r r r rrrrrrrrrrrrrrrrr)rrrrrrorDr!r"rr#rr$_idxrr input_keys rrmzShuffleLayer._construct_graphs& 1 --JKK!36"  C R-6t{{-CD61cQD DNN0 D++d#CL4<<( Cc4::q14::##T__dD$AB  C!!&CD! " ".  4$$'.6T__dD12 '-S8(.EzzQ 4#9#99)-%(@I*)rr_r,r/r1rfs rrgzBroadcastJoinLayer.__repr__ s.FMM IItxx   rct|dSrrrfs rrz"BroadcastJoinLayer.is_materializedrrcxt|dr |jS|j}||_|jSrirkrns rrpzBroadcastJoinLayer._dictrqrc |j|Srrsrts rr<zBroadcastJoinLayer.__getitem__rvrc,t|jSrrxrfs rrzzBroadcastJoinLayer.__iter__r{rc,t|jSrr}rfs rr~zBroadcastJoinLayer.__len__!rrct}|D])} |\}}||jk7r|j|+|S#t$rY9wxYwrrrs rrz!BroadcastJoinLayer._keys_to_parts$rrc|j|jkr.|j|j|j|jfS|j|j|j|j fSr)r0r2r/r1r.r-rfs r_broadcast_planz"BroadcastJoinLayer._broadcast_plan1sl   $"6"6 6 $$    $$   rc (|jdd\}}}tt}|xs|j|}|D]P}||j|fxxt |Dchc]}||fc}zcc<||j|fxx||fhzcc<R|Scc}w)zDetermine the necessary dependencies to produce `keys`. For a broadcast join, output partitions always depend on all partitions of the broadcasted collection, but only one partition of the "other" collection. N)r=rrrr_r) rrr bcast_name bcast_size other_namerrrDs rrz%BroadcastJoinLayer._cull_dependenciesKs.2-A-A"1-E* J 3:!4!4T!:  D $))T" #zAR'SAQ'S S # $))T" #T"(  #   (Ts B ct|j|j|j|j|j |j f|j|d|jS)N)rr) r*r_rr/r0r1r2rr3rs rrzBroadcastJoinLayer._cull^sb! II    MM  MM   ((    rc|j|}|j||}|t|jk7r|j |}||fS||fS)aCull a BroadcastJoinLayer HighLevelGraph layer. The underlying graph will only include the necessary tasks to produce the keys (indices) included in `parts_out`. Therefore, "culling" the layer only requires us to reset this parameter. rrrs rrzBroadcastJoinLayer.cullkrrcd|jz}d|jz}|r"td}td}td}nddlm}ddlm}dd lm}|j \}}} } |j|jkrd nd } i} |jD]} |jd k7r || | f| |f| || f<g}t|D]p}|jd k7rtj|| f|fn| | f||fg}| d k(r|j|| |f}t|||j f| |<|j#|r||f| |j| f<| S) z/Construct graph for a broadcast join operation.zinter-rz%dask.dataframe.multi._split_partitionz$dask.dataframe.multi._concat_wrapperz)dask.dataframe.multi._merge_chunk_wrapperr)_concat_wrapper)_merge_chunk_wrapper)_split_partitionleftrightinner)r_rdask.dataframe.multirFrGrHr=r0r2rr,rrrreverserr3r)rr inter_namersplit_partition_funcrmerge_chunk_funcr@rArBother_on bcast_siderorDrj _merge_args inter_keys rrmz#BroadcastJoinLayer._construct_graph{s )  ) $67$ --STK1;   L U U8<7K7K4 J H#33d6J6JJVPW , >Axx7"(O (ZO$L:& /xx7* !((#Q %aO  ' '')'A. $%% "I ##I./ /4$/ "=CA Y, >\ r)NNNNrr)r$r%r&r'rrrgrrrpr<rzr~rr=rrrrmrZr[s@rr*r*s|D6:> -!!  2&  % Mrr*cHeZdZdZ dfd ZedZdZdZxZ S)DataFrameIOLayera DataFrame-based Blockwise Layer with IO Parameters ---------- name : str Name to use for the constructed layer. columns : str, list or None Field name(s) to read in as columns in the output. inputs : list or BlockwiseDep List of arguments to be passed to ``io_func`` so that the materialized task to produce partition ``i`` will be: ``(, inputs[i])``. Note that each element of ``inputs`` is typically a tuple of arguments. io_func : callable A callable function that takes in a single tuple of arguments, and outputs a DataFrame partition. Column projection will be supported for functions that satisfy the ``DataFrameIOFunction`` protocol. label : str (optional) String to use as a prefix in the place-holder collection name. If nothing is specified (default), "subset-" will be used. produces_tasks : bool (optional) Whether one or more elements of `inputs` is expected to contain a nested task. This argument in only used for serialization purposes, and will be deprecated in the future. Default is False. creation_info: dict (optional) Dictionary containing the callable function ('func'), positional arguments ('args'), and key-word arguments ('kwargs') used to produce the dask collection with this underlying ``DataFrameIOLayer``. annotations: dict (optional) Layer annotations to pass through to Blockwise. c ||_||_||_||_||_||_||_||_t|ts@tt|jD cic] \} } | f|  c} } |j } n|} |j|tdfi} t |9|jd| | dfgi|ycc} } w)N)r0rrD)outputoutput_indicesroindicesr.r)r__columnsrio_funclabelr0r creation_inforr r rrrNr)rr_columnsrr]r^r0r_rrDr! io_arg_maprorOs rrzDataFrameIOLayer.__init__s     ,&*&,/))24;;)?@vq#!s@#22J  Jyy7OA$678 99 #&'#  As'C c|jS)z(Current column projection for this layer)r\rfs rr`zDataFrameIOLayer.columnss}}rc ddlm}t|}|j$t |jj |rt |j|r|jj|}n |j}t|jxsddzt|j|z||j||j|j|j}|S|S)zProduce a column projection for this IO layer. Given a list of required output columns, this method returns the projected layer. r)DataFrameIOFunctionsubset-)r^r0r)dask.dataframe.io.utilsrdrr`r issupersetrr]project_columnsrWr^r r_rr0r)rr`rdr]layers rriz DataFrameIOLayer.project_columnss @w- << 3t||#4#?#?#H$,,(;<,,66w?,,$'x3.$))W1MM jj#22 ,,ELKrcvdj|jt|j|jS)Nz3DataFrameIOLayer)rr_r3rr`rfs rrgzDataFrameIOLayer.__repr__<s-DKK IIs4;;'  r)NFNN) r$r%r&r'rrr`rirgrZr[s@rrWrWs;"T& P> rrWc<eZdZUdZded<ded<ded<ded<ded <d ed <ded <ded <ded<ded<ded<ded< d$ d%fd ZdddZd&dZdZdZ dZ dZ dZ e dZdZdZd Zd!Zd"Zd#ZxZS)'DataFrameTreeReductionagDataFrame Tree-Reduction Layer Parameters ---------- name : str Name to use for the constructed layer. name_input : str Name of the input layer that is being reduced. npartitions_input : str Number of partitions in the input layer. concat_func : callable Function used by each tree node to reduce a list of inputs into a single output value. This function must accept only a list as its first positional argument. tree_node_func : callable Function used on the output of ``concat_func`` in each tree node. This function must accept the output of ``concat_func`` as its first positional argument. finalize_func : callable, optional Function used in place of ``tree_node_func`` on the final tree node(s) to produce the final output for each split. By default, ``tree_node_func`` will be used. split_every : int, optional This argument specifies the maximum number of input nodes to be handled by any one task in the tree. Defaults to 32. split_out : int, optional This argument specifies the number of output nodes in the reduction tree. If ``split_out`` is set to an integer >=1, the input tasks must contain data that can be indexed by a ``getitem`` operation with a key in the range ``[0, split_out)``. output_partitions : list, optional List of required output partitions. This parameter is used internally by Dask for high-level culling. tree_node_name : str, optional Name to use for intermediate tree-node tasks. strr_rrrrrtree_node_funcCallable | None finalize_func split_every split_outz list[int]output_partitionstree_node_namewidthsheightc t || ||_||_||_||_||_||_||_||_ | "tt|jxsdn| |_ | xsd|jz|_ |j} | g|_| dkDrLtj | |jz } |jj#t%| | dkDrLt'|j|_y)NrrRz tree_node-)rNrr_rrrrorqrrrsrrrtrurvmathceilrrr3rw)rr_rrrrorqrrrsrtrurrrOs rrzDataFrameTreeReduction.__init__us [1 $!2&,*&"!( t~~*+ ,"  -H tyy0H&&g aiIIed&6&667E KK  s5z *ai$++& rrsplitc*|jr||fzS|Sr)rs)rr| name_partss r _make_keyz DataFrameTreeReduction._make_keys)-zUH$FJFrc|r|jr |j}n |j}tj||j|fSr)rqrorrr)r input_keys final_task outer_funcs r _define_taskz#DataFrameTreeReduction._define_tasks< $,,++J,,J J(8(8*EErc i}|js|S|j}|jrd|dz }|jD]P}t|jD]6}t j |j|f|f||j|||<8R|jdk\rm|jD][}td|jD]>}t|j|D]}|j|dz }|j|z}t||jz|} |dk(r,t|| Dcgc]}|j|||} }n9t|| Dcgc]$}|j|j||dz |&} }||jdz k(r3|dk(s Jd|d|j| d ||j|f<|j| d ||j|j|||<"A^|S|jD]9}|j|d|g} |j| d ||j|f<;|Scc}wcc}w) z%Construct graph for a tree reduction.z-splitr{rRrzgroup = z%, not 0 for final tree reduction taskT)rF)rtrrsrrrrrrwrvrrminrurr_) rroname_input_userXprgroupp_maxlstartlstoprs rrmz'DataFrameTreeReduction._construct_graphs%%J >> h &N++ t556A ((!,GC~qBC  ;;! ++' P"1dkk2%PE!&t{{5'9!:#P $ EAI 6!%!1!1E!9 #FT-=-=$=u E A:*/vu)=*$%!%~q J*J**/vu)= *%&!%$($7$7EAIQ!/!"*J*!DKK!O3!& W!)%0UVW *262C2C *t3D3CA/!% 1 1* 1 O  $$($7$7Q!/!"?#P%P' P\ ++ U"nn^QanHI &*&7&7 t&7&TTYYN# U G* *s 0I )I cddj|j|j|jS)Nz>DataFrameTreeReduction)rr_rrsrfs rrgzDataFrameTreeReduction.__repr__s(OVV IIt  rcX|jDchc]}|j|fc}Scc}wr)rtr_)rrXs r _output_keysz#DataFrameTreeReduction._output_keyss$(,(>(>?1A???rcxt|dr |jS|j}||_|jS)N_cached_output_keys)rlrr)r output_keyss rrz&DataFrameTreeReduction.get_output_keyss; 4. /++ +++-K'2D $'''rct|dSrrrfs rrz&DataFrameTreeReduction.is_materializedrrcxt|dr |jS|j}||_|jSrirkrns rrpzDataFrameTreeReduction._dictrqrc |j|Srrsrts rr<z"DataFrameTreeReduction.__getitem__rvrc,t|jSrrxrfs rrzzDataFrameTreeReduction.__iter__r{rct|jddxsd|jxsdz}|jr%||jt |j zzS|Sr)sumrvrsrr3rt)r tree_sizes rr~zDataFrameTreeReduction.__len__sWQR).Q4>>3FQG >>t55Drs" #$UU%== @ @.K K&E*E 3% 3wwt4n R%2^RRjo%odzzzt yt nEUEr