o $id@s.ddlZddlZddlZddlmZmZmZmZmZm 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(erddl)m*Z*ddl+m,Z,dgZ-e.e/Z0GdddZ1dS)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 @seZdZdZdedefddZdefddZdFd d Z defd d Z defddZ e    dGde dedededef ddZdeddefddZdHd!d"ZdId#d$ZdId%d&Zdeefd'd(Z dJd)edeed*ffd+d,Zd-eed*ffd.d/Zdeeefd0d1Zdeefd2d3Zede e!e"eed ffd4d5Z#e dJd6ede"fd7d8Z$e%defd9d:Z&dKdd?Z(defd@dAZ)defdBdCZ*defdDdEZ+d;S)L 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_contextcCsF||_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)selfrrr,T/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/_internal/plan.py__init__1s   zExecutionPlan.__init__returncCs|jpdd|jd|jS)ziUnique ID of the dataset, including the dataset name, UUID, and current execution index. dataset_)r(r&r'r+r,r,r-get_dataset_id[szExecutionPlan.get_dataset_idrcCs.ddlm}|jd7_||j|}|S)z!Create an executor for this plan.rr)/ray.data._internal.execution.streaming_executorrr'r*r3)r+rexecutorr,r,r-create_executorcs zExecutionPlan.create_executorcCsd|jd|jdS)NzExecutionPlan(dataset_uuid=z, snapshot_operator=))r&r!r2r,r,r-__repr__ks zExecutionPlan.__repr__c Cs~ddgt}gd}|j}g}t||D]#\}}||}|j|jdd\}}d|d} | |} || qd|S) z@Return a string representation of the logical and physical plan.cSs|SNr,)xr,r,r-vsz'ExecutionPlan.explain..)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_strr1bannersectionr,r,r-explainss    zExecutionPlan.explainr>rTFopcurr_strdepthincluding_sourcer=c Cs|s t|tr ||fS|}|rt|n|j}|dkr"||d7}nd|dd}||d|d7}|jD]}t|||d||\}} t|| }q7||fS)zXTraverse (DFS) the Plan DAG and return a string representation of the operators.r  r4+- ) isinstancer reprnameinput_dependenciesrrAmax) rOrPrQrRr=curr_max_depthop_strtrailing_spaceinputinput_max_depthr,r,r-rAs   z"ExecutionPlan.generate_plan_string dataset_clsrcCsddlm}d}d}|s|j|jjdd\}}|jdur(|jj}|j}ne|j dur7|j j}|j j j}nVd}|jj}t |t sVt |jdkrLd}n |jd}t |t rB|r]d}d}n0t |t sfJ|ttidd |j} | t|| j| }| }n |jdd }|j}|durd } n5t |trt|} n+g} t|j|jD]\} } t| d r| j} | | d | qd| } d| d} |durd}d} ||kr|} | dusJ|jdurd |jnd}| rd| dnd}d |j|||| }d}d}d}||}t ||kr||d| }t ||kr~g} t|j|jD]A\} } t| d r7| j} ||d| d | }t ||kredt| }t!|t ||}|d||}| |q*d| } d| d||d} |jdurd||d|jdnd}| rd||d| dnd}|jd ||d||d!|d||d| d|d"}|dkr||7}|S|||dd#|7}|S)$zCreate a cosmetic string representation of this execution plan. Returns: The string representation of this execution plan. r)MaterializedDatasetr>F)rRNr4Tmetadataparent)fetch_if_missingzUnknown schema__name__z: z, {}?z name={}, z num_blocks=z{}({}{}num_rows={}, schema={})P z zschema=z...: z, z{ rSzname=,(z num_rows=r8rV)"ray.data.datasetrbhas_computed_outputrAr?rBr#schemanum_rowsr$rdrWr lenrZrrr*link_logical_planr meta_counttypestrr@namestypeshasattrrgrCrDinitial_num_blocksr(formatr[)r+rarbrKplan_max_depthrrcounthas_n_ary_operatorrBrG schema_strnt num_blocksname_strnum_blocks_str dataset_strSCHEMA_LINE_CHAR_LIMITMIN_FIELD_LENGTH INDENT_STRr^schema_str_on_new_linecol_strshortened_suffixchars_left_for_col_namer,r,r-get_plan_as_strings                          z ExecutionPlan.get_plan_as_string logical_planrcCs||_|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)r?r*)r+rr,r,r-ruNszExecutionPlan.link_logical_plancCs>t|j|jd}|jdur|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. rN)rr r*r#r!r"r(r+ plan_copyr,r,r-copyWs  zExecutionPlan.copycCsVtt|j|jd}|jr%t|j|_t|j|_t|j|_|j|_|S)zCreate a deep copy of this execution plan. This copy can be executed AND cleared without mutating the original. Returns: A deep copy of this execution plan. r)rrr r*r#r!r"r(rr,r,r- deep_copyls zExecutionPlan.deep_copycCs |jjS)zGet the estimated number of blocks from the logical plan after applying execution plan optimizations, but prior to fully executing the dataset.)r?rBestimated_num_outputsr2r,r,r-r|s z ExecutionPlan.initial_num_blocksrfzpyarrow.lib.SchemacCs|jdur|jSd}|r|jj}n.|jj}|durA|rA|\}}}|tdd|D}Wdn1s z'ExecutionPlan.schema..) r%rqr#rrr?rB infer_schemaexecute_to_iteratorr cache_schema)r+rfrriter_ref_bundlesr1r6r,r,r-rrs      zExecutionPlan.schemarrcCs ||_dSr:)r%)r+rrr,r,r-rs zExecutionPlan.cache_schemacCs|jjjS)z1Get the input files of the dataset, if available.)r?rBinfer_metadata input_filesr2r,r,r-rszExecutionPlan.input_filescCsN|jj}|rtdd|jjD}|S|jdur#|j}|Sd}|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. csrr:)rs)rmr,r,r-rz+ExecutionPlan.meta_count..N)r?rBrqsumr#rdrrs)r+rBrsr,r,r-rvs zExecutionPlan.meta_countcCsd|_|r|}t|g|jdfSddlm}|}|||}t|}z t t |g|}Wn t y;Ynw| |_||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)rqexecuteiterr"*ray.data._internal.execution.legacy_compatrr7 itertoolschainnext StopIteration get_stats)r+rrr6 bundle_itergenr,r,r-rs     z!ExecutionPlan.execute_to_iteratorpreserve_orderc sd|_|j}tdstdrtd|sddl m }m }t |j jtrZ|j jdurZ|||j j}tdd |D}td d |D}td d |D||d }n?|} || ||j|d} tt| | j| d }Wdn1swY| jdd} |jrt| z%tt t!j"} | j#j$dkrt%| j#j&_'| j#j(dkrt%| j#j)_*Wnt+y} zt,d| WYd} ~ nd} ~ wwd_-fdd||_.|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_listNcsrr:) owns_blocksrr,r,r-rrz(ExecutionPlan.execute..csrr:rrr,r,r-rrcSs$g|]}|jD]\}}||fqqSr,)blocks)rrblockrdr,r,r- sz)ExecutionPlan.execute..)rrr) dataset_uuidrF)include_parentzLSkipping recording memory spilled and restored statistics due to exception: cs0j|jdd7_|jD]}|qdS)Nobj_store_mem_spilledr)dataset_bytes_spilled extra_metricsgetparents) cur_statsre collect_statsrr,r-rKs   z,ExecutionPlan.execute..collect_stats)2r)r*rayavailable_resourcesrrloggerwarningrqrrrrWr?rBr output_dataallrr r7r&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_bundlesrrrrr6rstats_summary_stringreplyer,rr-rs            zExecutionPlan.executecCs|jS)zBReturn ``True`` if this plan has been partially or fully executed.)r)r2r,r,r-has_started_execution\sz#ExecutionPlan.has_started_executionNcCsd|_d|_d|_dS)z;Clear the snapshot kept in the plan to the beginning state.N)r#r!r"r2r,r,r-clear_snapshotas zExecutionPlan.clear_snapshotcCs|js tiddS|jS)zqReturn stats for this plan. If the plan isn't executed, an empty stats object will be returned. Nrc)r"rr2r,r,r-rgs zExecutionPlan.statscCstdd|jDS)z/Return whether this plan has lazy input blocks.css|]}t|tVqdSr:)rWr)rrOr,r,r-rrsz/ExecutionPlan.has_lazy_input..)rr?sourcesr2r,r,r-has_lazy_inputpszExecutionPlan.has_lazy_inputcCs|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!r?rBr2r,r,r-rqts  z!ExecutionPlan.has_computed_outputcCsBddlm}ddlm}|jjD] }t|||frdSqdS)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_operatorrr?rBpost_order_iterrW)r+rrrOr,r,r-require_preserve_order}s  z$ExecutionPlan.require_preserve_order)r/r)r>rTF)rr)r/r)F)r/N),rg __module__ __qualname____doc__rrr.rxr3r7r9rN staticmethodrrboolrArrrurrrr|rrwrrrrrrvrrrr rrpropertyrrrrrqrr,r,r,r-r%sx  *      ! % v   r)2rrloggingtypingrrrrrrrpyarrowrray._private.internal_apir r 'ray.data._internal.execution.interfacesr %ray.data._internal.logical.interfacesr 6ray.data._internal.logical.interfaces.logical_operatorr 2ray.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.debugrr5rrprINHERITABLE_REMOTE_ARGS getLoggerrgrrr,r,r,r-s2$