&`idRddlZddlZddlZddlmZmZmZmZmZm Z m Z ddl Z ddl Z ddl mZmZddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!m"Z"dd l#m$Z$ddl%m&Z&ddl'm(Z(er ddl)m*Z*ddl+m,Z,dgZ-ej.e/Z0GddZ1dS)N) TYPE_CHECKINGIteratorListOptionalTupleTypeUnion)get_memory_info_replyget_state_from_address) RefBundle)SourceOperator)LogicalOperator) LogicalPlan)Operator)Read)get_plan_conversion_fns) DatasetStats)BlockMetadataWithSchema_take_first_non_empty_schema) DataContext)omit_traceback_stdout)log_onceStreamingExecutor)Datasetscheduling_strategyc eZdZdZdedefdZdefdZd/dZ defd Z defd Z e d0de dedededef dZdeddefdZd1dZd2dZd2dZdeefdZ d3dedeedffd Zd!eedffd"Zdeeefd#Zdeefd$Zede e!e"eedffd%Z#e d3d&ede"fd'Z$e%defd(Z&d4d*Z'defd+Z(defd,Z)defd-Z*defd.Z+d)S)5 ExecutionPlanaA lazy execution plan for a Dataset. This lazy execution plan builds up a chain of ``List[RefBundle]`` --> ``List[RefBundle]`` operators. Prior to execution, we apply a set of logical plan optimizations, such as operator fusion, in order to reduce Ray task overhead and data copies. Internally, the execution plan holds a snapshot of a computed list of blocks and their associated metadata under ``self._snapshot_bundle``, where this snapshot is the cached output of executing the operator chain.stats data_contextc||_d|_d|_d|_d|_d|_d|_d|_d|_d|_ ||_ dS)zCreate a plan with no transformation operators. Args: stats: Stats for the base blocks. data_context: :class:`~ray.data.context.DataContext` object to use for execution. NF) _in_stats_snapshot_operator_snapshot_stats_snapshot_bundle_snapshot_metadata_schema_schema _dataset_uuid _run_index _dataset_name_has_started_execution_context)selfrr s k/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/plan.py__init__zExecutionPlan.__init__1se>B# $OS& !!&+#$ returnc8|jpdd|jd|jS)ziUnique ID of the dataset, including the dataset name, UUID, and current execution index. dataset_)r+r)r*r.s r/get_dataset_idzExecutionPlan.get_dataset_id[s/ !.Y W W1C W Wdo W W r1rcxddlm}|xjdz c_||j|}|S)z!Create an executor for this plan.rr)/ray.data._internal.execution.streaming_executorrr*r-r7)r.rexecutors r/create_executorzExecutionPlan.create_executorcsIUUUUUU 1$$T]D4G4G4I4IJJr1c(d|jd|jdS)NzExecutionPlan(dataset_uuid=z, snapshot_operator=))r)r$r6s r/__repr__zExecutionPlan.__repr__ks1  .  !%!8    r1c.dgtz}gd}|j}g}t||D]P\}}||}||jd\}}d|d} | |} || Qd|S)z@Return a string representation of the logical and physical plan.c|SN)xs r/z'ExecutionPlan.explain..vsr1)z Logical PlanzLogical Plan (Optimized)z Physical PlanzPhysical Plan (Optimized)T) show_op_reprz -------- z -------- )r _logical_planzipgenerate_plan_stringdagappendjoin) r. convert_fnstitlesplansectionstitle convert_fnplan_strr5bannersections r/explainzExecutionPlan.explainss#{m&=&?&??    !!$V[!9!9 % % E::d##D33DH43PPKHa55555F+++G OOG $ $ $ $wwx   r1rGrTFopcurr_strdepthincluding_sourcerFc:|st|tr||fS|}|rt|n|j}|dkr ||dz }nd|dz dzz}||d|dz }|jD]6}t |||dz||\}} t|| }7||fS)zXTraverse (DFS) the Plan DAG and return a string representation of the operators.r  r9+- ) isinstancer reprnameinput_dependenciesrrJmax) rXrYrZr[rFcurr_max_depthop_strtrailing_spaceinputinput_max_depths r/rJz"ExecutionPlan.generate_plan_strings  #Jr>$B$B #U? "*6brw A:: 6 %HH UQY!O4N >88f888 8H* B BE(5(J(Jx,r`)"ray.data.datasetrmhas_computed_outputrJrHrKr&schemanum_rowsr'rorar lenrdrrr-link_logical_planr meta_counttypestrrInamestypeshasattrrrrLrMinitial_num_blocksr+formatre)r.rkrmrTplan_max_depthr}counthas_n_ary_operatorrKrP schema_strnt num_blocksname_strnum_blocks_str dataset_strSCHEMA_LINE_CHAR_LIMITMIN_FIELD_LENGTH INDENT_STRrhschema_str_on_new_linecol_strshortened_suffixchars_left_for_col_names r/get_plan_as_stringz ExecutionPlan.get_plan_as_strings 988888''))) 5'+'@'@"&(A(( $Hn$0.5-6688/;7>6?H&+"(,$S.9943122Q66-1*03C %S.994& .!F EE%c>::??C??:($b>>> D**;sDM+J+JKKK![[]]F OO--EE[[%[88F)2244E >)JJ  % % 0VJJJFL&,77 / /11j))# A!!Q++!++....:..Jz)C/J =E - - -0022J)))!-   t1 2 2 2  :DK5z55556==        "$ #n4 {  4 4 4 )7%W %W%W:%W%W ")**-CCC  fl;;//DAqq*--'J!/IaIIIaIIG7||&<<<+;3q66+;+;(362S9I5J5JJ,33/ ''?(?'?@TBRTT %%g...."ZZ 33 J&)Jn)Jj)J)JJSP %1L^KZKKd6HKKKK I^IZIIJIIII  ''''!''$'&0'';@''$ '&0 ''9C '' $ '''  Q    #HH :!);<NNNN NHr1 logical_planrc6||_|j|j_dS)zLink the logical plan into this execution plan. This is used for triggering execution for optimizer code path in this legacy execution plan. N)rHr-)r.rs r/rzExecutionPlan.link_logical_planNs *&*m###r1ct|j|j}|j$|j|_|j|_|j|_|j|_|S)zCreate a shallow copy of this execution plan. This copy can be executed without mutating the original, but clearing the copy will also clear the original. Returns: A shallow copy of this execution plan. r )rr#r-r&r$r%r+r. plan_copys r/copyzExecutionPlan.copyWs_" N     ,)-)>I &+/+BI ((,(|jjS)zGet the estimated number of blocks from the logical plan after applying execution plan optimizations, but prior to fully executing the dataset.)rHrKestimated_num_outputsr6s r/rz ExecutionPlan.initial_num_blockss!%;;===r1rqzpyarrow.lib.Schemacv|j|jSd}|r |jj}nm|jj}|M|rK|\}}}|5td|D}dddn #1swxYwY| ||jS)aKGet the schema after applying all execution plan optimizations, but prior to fully executing the dataset (unless `fetch_if_missing` is set to True). Args: fetch_if_missing: Whether to execute the plan to fetch the schema. Returns: The schema of the output dataset. Nc3$K|] }|jV dSrBr}.0bundles r/ z'ExecutionPlan.schema..s5::*0 ::::::r1) r(r|r&r}rHrK infer_schemaexecute_to_iteratorr cache_schema)r.rqr}iter_ref_bundlesr5r;s r/r}zExecutionPlan.schemas < #<   # # % % *1FF'+88::F~"2~150H0H0J0J- !X9::4D:::F &!!!|s.BBBr}c||_dSrB)r()r.r}s r/rzExecutionPlan.cache_schemas  r1cH|jjjS)z1Get the input files of the dataset, if available.)rHrKinfer_metadata input_filesr6s r/rzExecutionPlan.input_filess!%4466BBr1c|jj}|r$td|jjD}n5|j|j}nd}|S)zGet the number of rows after applying all plan optimizations, if possible. This method will never trigger any computation. Returns: The number of records of the result Dataset, or None. c3$K|] }|jV dSrB)r~)rms r/rz+ExecutionPlan.meta_count..s$NN!1:NNNNNNr1N)rHrKr|sumr&rorr~)r.rKr~s r/rzExecutionPlan.meta_counts| $  # # % % NNt/D/MNNNNNHH    ! ! * 6))++4HHHr1cd|_|r,|}t|g|jdfSddlm}|}|||}t|} tj t|g|}n#t$rYnwxYw| |_||j|fS)aExecute this plan, returning an iterator. This will use streaming execution to generate outputs. NOTE: Executor will be shutdown upon either of the 2 following conditions: - Iterator is fully exhausted (ie until StopIteration is raised) - Executor instances is garbage-collected Returns: Tuple of iterator over output RefBundles, DatasetStats, and the executor. TNr)!execute_to_legacy_bundle_iterator) r,r|executeiterr%*ray.data._internal.execution.legacy_compatrr< itertoolschainnext StopIteration get_stats)r.rrr; bundle_itergens r/rz!ExecutionPlan.execute_to_iterators '+#  # # % % >\\^^F>>4#7= =      ''))77$GG ; #/499+s;;KK    D '1133D0(::s>#B"" B/.B/preserve_ordercTd|_|j}tjds)t drt d|sddl m }m }t|j jtr|j jx|||j j}t#d|D}t%d |D}t'd |D|| }n|5} || ||j| } t't-| | j|  }dddn #1swxYwY| d } |jrt |  t?tAtj!j"} | j#j$dkrtK| j#j&_'| j#j(dkrtK| j#j)_*n4#tV$r'} t ,d| Yd} ~ nd} ~ wwxYwd_-fd||_.|j j|_/|_0|j|j0_1|j.S)zExecutes this plan (eagerly). Args: preserve_order: Whether to preserve order in execution. Returns: The blocks of the output dataset. TCPU cpu_warninga<Warning: The Ray cluster currently does not have any available CPUs. The Dataset job will hang unless more CPUs are freed up. A common reason is that cluster resources are used by Actors or Tune trials; see the following link for more details: https://docs.ray.io/en/latest/data/data-internals.html#ray-data-and-tuner)_get_initial_stats_from_planexecute_to_legacy_block_listNc3$K|] }|jV dSrB) owns_blocksrs r/rz(ExecutionPlan.execute..s%!R!R&"4!R!R!R!R!R!Rr1c3$K|] }|jV dSrBrrs r/rz(ExecutionPlan.execute..s566&,FM666666r1c0g|]}|jD] \}}||f SrC)blocks)rrblockros r/ z)ExecutionPlan.execute..sJ"/5},E8)r1)rr}) dataset_uuidrF)include_parentzLSkipping recording memory spilled and restored statistics due to exception: cxj|jddz c_|jD] }|dS)Nobj_store_mem_spilledr)dataset_bytes_spilled extra_metricsgetparents) cur_statsrp collect_statsrs r/rz,ExecutionPlan.execute..collect_statsKsb++y/F/J/J+Q00++(/**F!M&))))**r1)2r,r-rayavailable_resourcesrrloggerwarningr|rrrrarHrKr output_dataallrr r<r)tupleiter_blocks_with_metadata_owned_by_consumer get_schemar to_summary to_stringenable_auto_log_statsinfor r get_runtime_context gcs_address store_statsspill_time_total_sintspilled_bytes_totalglobal_bytes_spilledrestore_time_total_srestored_bytes_totalglobal_bytes_restored Exceptiondebugrr&r$r%r)r.rcontextrroutput_bundlesrr}rr;rstats_summary_stringreplyerrs @@r/rzExecutionPlan.executes '+#-&((,,U33  && _''))X C         4-1>BB. 6&*6688D 54T:: "&!3!7!C!C!E!E!!R!R>!R!R!RRR 5660>666#&4 !,!))++ x99 %)%7'5 F 'f>>@@AA$*$=%0022F               !**,,','7'7'9'9'C'C#((D(($06KK 4555 -*3+B+D+D+PQQ$7!;;14)=22E.$9A==25)>33E/    &"#&&  +,E ' * * * * * * M% %+D !&*&8&.rs,OOB:b$''OOOOOOr1)rrHsourcesr6s r/has_lazy_inputzExecutionPlan.has_lazy_inputps.OO$2D2L2L2N2NOOOOOOr1c>|jduo|j|jjkS)ztWhether this plan has a computed snapshot for the final operator, i.e. for the output of this plan. N)r&r$rHrKr6s r/r|z!ExecutionPlan.has_computed_outputts+  ! - B'4+=+AA r1cddlm}ddlm}|jjD]}t|||frdSdS)z-Whether this plan requires to preserve order.r)Sort)ZipTF)8ray.data._internal.logical.operators.all_to_all_operatorr3ray.data._internal.logical.operators.n_ary_operatorrrHrKpost_order_iterra)r.rrrXs r/require_preserve_orderz$ExecutionPlan.require_preserve_order}snQQQQQQKKKKKK$(88::  B"sDk** tt ur1)r2r)rGrTF)rr)r2r)F)r2N),rr __module__ __qualname____doc__rrr0rr7r<r?rW staticmethodrrboolrJrrrrrrrr rr}rrrrrrrr rrpropertyrrrrr|r rCr1r/rr%so Q Q(%(%"(%(%(%(%T      #    !!!!!:!%" (( ((( (  (((\(>\d9o\#\\\\|4444*(>HSM>>>>(- $ t)) *B5/C)C#DCXd3i0CCCCHSM"#; x "L(;N2OO P#;#;#;#;J %t%t%t% t%t%t%t%l+t+++X+$$$$ $|$$$$PPPPP T    r1r)2rrloggingtypingrrrrrrr pyarrowrray._private.internal_apir r 'ray.data._internal.execution.interfacesr %ray.data._internal.logical.interfacesr 6ray.data._internal.logical.interfaces.logical_operatorr2ray.data._internal.logical.interfaces.logical_planr.ray.data._internal.logical.interfaces.operatorr2ray.data._internal.logical.operators.read_operatorr%ray.data._internal.logical.optimizersrray.data._internal.statsrray.data.blockrrray.data.contextrray.data.exceptionsrray.util.debugrr:rr{rINHERITABLE_REMOTE_ARGS getLoggerrrrrrCr1r/r$s NNNNNNNNNNNNNNNNNN SSSSSSSS======@@@@@@RRRRRRJJJJJJCCCCCCCCCCCCIIIIII111111PPPPPPPP((((((555555######))(((((11  8 $ $` ` ` ` ` ` ` ` ` ` r1