0Fie4ddlmZddlmZmZddlmZmZddlZddl m Z m Z ddl m Z ddlmZddlmZdd lmZmZmZdd lmZmZerddlZdd lmZmZmZdd lm Z d Z!d;dZ"dd&Z$e$Z%d Z!d?d/Z&d@d8Z'Gd9d:eZ(dS)A annotations)IterableSequence) TYPE_CHECKINGAnyN)is_dask_collectiontokenize)HighLevelGraph)Layer)check_minimal_arrow_version) ShuffleId barrier_keyget_worker_plugin)shuffle_barriershuffle_transfer) IndexLabelMergeHowSuffixes_Frame__hash_partitiondf pd.DataFrameindexrc.ddl}ddlm}|jj|ot | }t||s'|rt|}| |}n$t|dr| }|S)Nrrto_frame) pandasdask.dataframe.corerapitypes is_list_liker isinstancelist_select_columns_or_indexhasattrr)rrpdr list_likes :lib/python3.11/site-packages/distributed/shuffle/_merge.py_prepare_index_for_partitioningr*s****** ))%00R9KE9R9R5RI eV $ $ !  KKE++E22  # #!   L npartitionsintcBt||}ddlm}|jdg}|jjdd|_|||p|j|d}|j dit|i}|jjj |jj_ |S)Nr)partitioning_indexF)r,metatransform_divisions) r*dask.dataframe.shuffler/_meta_constructor_sliced_meta_nonemptyrmap_partitionsr,assign_HASH_COLUMN_NAMEname)rrr,r/r1 partitionsdf2s r)_calculate_partitionsr>.s +B 6 6E999999 8 ' ' , ,D"(!,DJ%%12> ! &J ") 6 6(*5 6 6C8>.CIO Jr+inner_x_yFlhsrleft_onIndexLabel | Nonerhsright_onhowr int | Nonesuffixesr indicatorboolc 4ddlm}m} |t|j|j}t ||rd} d} nd} |} t ||rd} d} nd} |} t || | | | ||}t|jr|j n|j }t|jr|j n|j }|j |fi|}t|||}t|||}dt||fi|z}tjd}t#did|d |jd |j d | d |jd |jd|j d| d|jd|d|d|d|d|d| d| d|}t'j||||g}| |||dg|dzzS)Nr)Index new_dd_objectTF)rHrDrG left_index right_indexrJrKz hash-join-zdistributed.p2p.diskr;name_input_leftmeta_input_leftrDn_partitions_leftname_input_rightmeta_input_rightrGn_partitions_right meta_outputrHr,rJrKrPrQdisk) dependenciesr0r3)rrNrOmaxr,r#dictlencolumnsr7r5merger>r daskconfiggetHashJoinP2PLayer_namer from_collections)rCrDrFrGrHr,rJrKrNrO_left_onrP _right_onrQ merge_kwargs _lhs_meta _rhs_metar1 merge_namerY join_layergraphs r) hash_join_p2prnAs98888888#/3?;; '5!!  (E""     L'*#+&6&6E""CII&)#+&6&6E""CII 9?9 5 5 5 5D Wk : :C X{ ; ;Cc B B\ B BBJ!788D! Z    //   ??D C K):  K!"T#J&  +Jc3Z   E = D4&K!O2L M MMr+inputidrinput_partitionr1 parts_outset[int]rYc :t||||t|||S)N)rorprqr,columnr1rrrY)rr:)rorprqr,r1rrrYs r)merge_transferrvs3  '      r+shuffle_id_leftshuffle_id_rightoutput_partition barrier_left barrier_right result_metarPrQc ddlm} t} | |||t d}| |||t d}| ||||||| | |  S)Nr) merge_chunkignore)r^errors)rHr|rDrGrJrPrQ)dask.dataframe.multir~rget_output_partitiondropr:)rwrxryrzr{rHrDrGr|rJrPrQr~extleftrights r) merge_unpackrs100000   C  # #'7   d$Xd66   $ $-)9   d$Xd66  ;      r+c>eZdZUded<ded<ded<ded<d ed <d ed <d ed<ded<d ed<ded<ded<d ed<ded<d ed<ded<ded<d ed< d:d;fd$ Z dd*Zd+Zd,Zd-Z d.Z d/Z d0Z e d1Zd?d3Zd@d7ZdAd9ZxZS)Brcstrr;r-r,rrHrrJrLrKrrXz Sequence[int]rrrRrSrTrErDrPrUrVrWrGrQr?r@FNrYSequence | Noner dict | NonereturnNonect||_||_||_||_||_||_| |_||_| |_ ||_ ||_ | |_ |ptt| |_||_||_| |_| |_||_t+|dS)Nr)r r;rRrSrDrUrVrGrHr,rJrKrXr$rangerrrTrWrPrQrYsuper__init__)selfr;rRrSrDrTrWrUrVrGrXrPrQr,rYrHrJrKrrr __class__s r)rzHashJoinP2PLayer.__init__s, $%%% .. 0 0  &  "&">d5+=+=&>&>!2"4$&  [11111r+keys Iterable[str]Iterable[str] | Noneci}|p|}fdtjD}|fdtjDz}t|}|D]}||j|f<|S)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. c"h|] }j|f Sr3)rR.0irs r) z6HashJoinP2PLayer._cull_dependencies..s!KKKa%q)KKKr+c"h|] }j|f Sr3)rUrs r)rz6HashJoinP2PLayer._cull_dependencies..s!MMM$'+MMMr+)_keys_to_partsrr, frozensetr;)rrrrdepsparts` r)_cull_dependenciesz#HashJoinP2PLayer._cull_dependenciess:!4!4T!:!: KKKK59I3J3JKKK MMMMU4;K5L5LMMMM + +D&*D$)T" # # r+set[str]ct}|D]:} |\}}n#t$rYwxYw||jkr%||;|S)z4Simple utility to convert keys to partition indices.)set ValueErrorr;add)rrpartskeyrd_parts r)rzHashJoinP2PLayer._keys_to_partsst  C " uu     !! IIe     s  ''c*fdjDS)Nc"h|] }j|f Sr3)r;)rrrs r)rz3HashJoinP2PLayer.get_output_keys..(s ===dD!===r+rrrs`r)get_output_keysz HashJoinP2PLayer.get_output_keys's====dn====r+c(d|jd|jdS)NzHashJoin)r;r,rs r)__repr__zHashJoinP2PLayer.__repr__*s NNN4;KNNNNr+c"t|dS)N _cached_dict)r&rs r)is_materializedz HashJoinP2PLayer.is_materialized-st^,,,r+c|j|SN)_dict)rrs r) __getitem__zHashJoinP2PLayer.__getitem__0sz#r+c*t|jSr)iterrrs r)__iter__zHashJoinP2PLayer.__iter__3sDJr+c*t|jSr)r]rrs r)__len__zHashJoinP2PLayer.__len__6s4:r+ctt|dr|jS|}||_|jS)z$Materialize full dict representationr)r&r_construct_graph)rdsks r)rzHashJoinP2PLayer._dict9sA 4 ( ( $$ $''))C #D   r+ Sequence[str]c<tdid|jd|jd|jd|jd|jd|jd|jd|jd |j d |j d |j d |j d |d|j d|jd|jd|jd|jd|jS)Nr;rRrSrDrUrVrGrHr,rJrKrXrrrPrQrYrrTrWr3)rcr;rRrSrDrUrVrGrHr,rJrKrXrPrQrYrrTrW)rrrs r)_cullzHashJoinP2PLayer._cullCs1     00 !00 LL  "22  "22  ]]  (( ]] nn ((  i  (( ! "((# $#44% & $66'  r+all_keysrtuple[HashJoinP2PLayer, dict]c||}|||}|t|jkr||}||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. r)rrrrrr)rrrrr culled_deps culled_layers r)cullzHashJoinP2PLayer.cullZsk''-- --di-HH DN++ + +::i00L, ,$ $r+dict[tuple | str, tuple]ctd|j|j|j|j}td|j|j|j|j}i}d|z}d|z}t}t}t|j D]H}| ||ft|j|f|||j|j |j|j f|||f<It|jD]H}| ||ft|j|f|||j|j|j|j f|||f<It!t#|} t!t#|} t$||f|| <t$||f|| <|j} |jD]>} t(||| | | |j|j|j|j|j|j|jf || | f<?|S)Nz hash-joinzhash-join-transfer-)r rRrDr,rrrUrGr$rrTappendrvrSrYrWrVrrrr;rrHrXrJrPrQ) r token_left token_rightr name_left name_righttransfer_keys_lefttransfer_keys_rightr_barrier_key_left_barrier_key_rightr;part_outs r)rz!HashJoinP2PLayer._construct_graphjs!   L   N      ! M   N   )+)J6 *[8 !VV"fft-..  A  % %y!n 5 5 5%q) $  #CA  t.//  A  & & A 7 7 7&* %  $CQ ( *(=(=>>(;)?)?@@"1:?Q!R #2KAT"U y  H!"     %Cx ! ! r+)r?r@FNN)(r;rrRrrSrrDrErTr-rWr-rUrrVrrGrErXrrPrLrQrLr,r-rYrLrHrrJrrKrLrrrrrrrr)rrrrr)rrrr)rrr)rrrrrr)rr)__name__ __module__ __qualname____annotations__rrrrrrrrrpropertyrrrr __classcell__)rs@r)rcrcs  IIIMMMOOO!!!!""""$ )%)#'))2)2)2)2)2)2)2XFJ&    >>>OOO---   !!X!    .%%%% CCCCCCCCr+rc)rrrr)rrrrr,r-)r?Nr@F)rCrrDrErFrrGrErHrr,rIrJrrKrL)rorrprrqr-r,r-r1rrrrsrYrL)rwrrxrryr-rzr-r{r-rHrrDrrGrr|rrJrrPrLrQrL)) __future__rcollections.abcrrtypingrrr` dask.baser r dask.highlevelgraphr dask.layersr distributed.shuffle._arrowr distributed.shuffle._corerrrdistributed.shuffle._shufflerrrr'pandas._typingrrrrrr:r*r>rn hash_joinrvrrcr3r+r)rs""""""........%%%%%%%% 22222222......BBBBBBOOOOOOOOOOJJJJJJJJ+==========******'(0"%DNDNDNDNDNN  &*""""Jfffffufffffr+