o 1 ic@s"ddlZddlZddlZddlmZmZmZmZmZm Z m Z ddl Z ddl Z ddl mZmZddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZm Z dd l!m"Z"dd l#m$Z$ddl%m&Z&erddl'm(Z(ddl)m*Z*dgZ+e,e-Z.GdddZ/dS)N) TYPE_CHECKINGIteratorListOptionalTupleTypeUnion)get_memory_info_replyget_state_from_address) RefBundle)SourceOperator)LogicalOperator) LogicalPlan)Operator)Read) DatasetStats)BlockMetadataWithSchema_take_first_non_empty_schema) DataContext)omit_traceback_stdout)log_onceStreamingExecutor)DatasetZscheduling_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*c/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/data/_internal/plan.py__init__0s   zExecutionPlan.__init__returncCs|jpdd|jd|jS)ziUnique ID of the dataset, including the dataset name, UUID, and current execution index. Zdataset_)r&r$r%r)r*r*r+get_dataset_idZszExecutionPlan.get_dataset_idrcCs.ddlm}|jd7_||j|}|S)z!Create an executor for this plan.rr)/ray.data._internal.execution.streaming_executorrr%r(r0)r)rexecutorr*r*r+create_executorbs zExecutionPlan.create_executorcCsd|jd|jdS)NzExecutionPlan(dataset_uuid=z, snapshot_operator=))r$rr/r*r*r+__repr__js zExecutionPlan.__repr__cCsXddlm}|j}||j\}}d|}||j}|j|jdd\}}d|}||S)z@Return a string representation of the logical and physical plan.r)get_execution_planz-------- Logical Plan -------- T) show_op_reprz -------- Physical Plan -------- )Z%ray.data._internal.logical.optimizersr7 _logical_plangenerate_plan_stringdag)r)r7 logical_planZlogical_plan_strr.Z physical_planZphysical_plan_strr*r*r+explainrs   zExecutionPlan.explainrTFopcurr_strdepthincluding_sourcer8c 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  r1+- ) isinstancer reprnameinput_dependenciesrr:max) r?r@rArBr8Zcurr_max_depthZop_strtrailing_spaceinputZinput_max_depthr*r*r+r:s   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)rBNr1Tmetadataparent)fetch_if_missingzUnknown schema__name__z: z, {}?z name={}, z num_blocks=z{}({}{}num_rows={}, schema={})P z zschema=z...: z, z{ rCzname=,(z num_rows=r5rF)"ray.data.datasetrOhas_computed_outputr:r9r;r!schemanum_rowsr"rQrGr lenrJrrr(link_logical_planr meta_counttypestrzipnamestypeshasattrrTappendjoininitial_num_blocksr&formatrK)r)rNrOZplan_strZplan_max_depthr_countZhas_n_ary_operatorr;ZplanZ schema_strntZ num_blocksZname_strZnum_blocks_strZ dataset_strZSCHEMA_LINE_CHAR_LIMITZMIN_FIELD_LENGTHZ INDENT_STRrLZschema_str_on_new_lineZcol_strZshortened_suffixZchars_left_for_col_namer*r*r+get_plan_as_strings                          z ExecutionPlan.get_plan_as_stringr<rcCs||_|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)r9r()r)r<r*r*r+rb@szExecutionPlan.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)rrr(r!rr r&r)Z plan_copyr*r*r+copyIs  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. rr)rrtrr(r!rr r&rsr*r*r+ deep_copy^s 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.)r9r;Zestimated_num_outputsr/r*r*r+rlrs z ExecutionPlan.initial_num_blocksrSzpyarrow.lib.SchemacCs|jdur|jSd}|r|jj}n.|jj}|durA|rA|\}}}|tdd|D}Wdn1s z'ExecutionPlan.schema..) r#r^r!r_r9r;Z infer_schemaexecute_to_iteratorr cache_schema)r)rSr_Ziter_ref_bundlesr.r3r*r*r+r_xs      zExecutionPlan.schemar_cCs ||_dSrw)r#)r)r_r*r*r+rs zExecutionPlan.cache_schemacCs|jjjS)z1Get the input files of the dataset, if available.)r9r;infer_metadata input_filesr/r*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. csrvrw)r`)rzmr*r*r+r|z+ExecutionPlan.meta_count..N)r9r;r^sumr!rQrr`)r)r;r`r*r*r+rcs 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'r^executeiterr *ray.data._internal.execution.legacy_compatrr4 itertoolschainnext StopIteration get_stats)r)r{rr3Z bundle_itergenr*r*r+r~s     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. TZCPUZ 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_listNcsrvrw) owns_blocksryr*r*r+r|rz(ExecutionPlan.execute..csrvrwrxryr*r*r+r|r}cSs$g|]}|jD]\}}||fqqSr*)blocks)rzr{blockrQr*r*r+ sz)ExecutionPlan.execute..)rr_) dataset_uuidrF)Zinclude_parentzLSkipping recording memory spilled and restored statistics due to exception: cs0j|jdd7_|jD]}|qdS)NZobj_store_mem_spilledr)dataset_bytes_spilledZ extra_metricsgetparents)Z cur_statsrR collect_statsrr*r+r=s   z,ExecutionPlan.execute..collect_stats)2r'r(rayZavailable_resourcesrrloggerwarningr^rrrrGr9r;r Z output_dataallrr r4r$tupleZiter_blocks_with_metadataZ_owned_by_consumerZ get_schemarZ to_summaryZ to_stringZenable_auto_log_statsinfor r Zget_runtime_contextZ gcs_addressZ store_statsZspill_time_total_sintZspilled_bytes_totalZglobal_bytes_spilledZrestore_time_total_sZrestored_bytes_totalZglobal_bytes_restored Exceptiondebugrr!rr r)r)rcontextrrZoutput_bundlesrr_r{r3rZstats_summary_stringZreplyer*rr+rs            zExecutionPlan.executecCs|jS)zBReturn ``True`` if this plan has been partially or fully executed.)r'r/r*r*r+has_started_executionNsz#ExecutionPlan.has_started_executionNcCsd|_d|_d|_dS)z;Clear the snapshot kept in the plan to the beginning state.N)r!rr r/r*r*r+clear_snapshotSs zExecutionPlan.clear_snapshotcCs|js tiddS|jS)zqReturn stats for this plan. If the plan isn't executed, an empty stats object will be returned. NrP)r rr/r*r*r+rYs zExecutionPlan.statscCstdd|jDS)z/Return whether this plan has lazy input blocks.css|]}t|tVqdSrw)rGr)rzr?r*r*r+r|dsz/ExecutionPlan.has_lazy_input..)rr9sourcesr/r*r*r+has_lazy_inputbszExecutionPlan.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!rr9r;r/r*r*r+r^fs  z!ExecutionPlan.has_computed_outputcCsBddlm}ddlm}|jjD] }t|||frdSqdS)z-Whether this plan requires to preserve order.r)Sort)ZipTF)Z8ray.data._internal.logical.operators.all_to_all_operatorrZ3ray.data._internal.logical.operators.n_ary_operatorrr9r;Zpost_order_iterrG)r)rrr?r*r*r+require_preserve_orderos  z$ExecutionPlan.require_preserve_order)r-r)r>rTF)r<r)r-r)F)r-N),rT __module__ __qualname____doc__rrr,rer0r4r6r= staticmethodrrboolr:rrqrbrtrurrlrrdr_rrrrcrrrr r~rpropertyrrrrr^rr*r*r*r+r$sx  *      ! % v   r)0rtrloggingtypingrrrrrrrZpyarrowrZray._private.internal_apir r Z'ray.data._internal.execution.interfacesr Z%ray.data._internal.logical.interfacesr Z6ray.data._internal.logical.interfaces.logical_operatorr Z2ray.data._internal.logical.interfaces.logical_planrZ.ray.data._internal.logical.interfaces.operatorrZ2ray.data._internal.logical.operators.read_operatorrZray.data._internal.statsrZray.data.blockrrZray.data.contextrZray.data.exceptionsrZray.util.debugrr2rr]rZINHERITABLE_REMOTE_ARGS getLoggerrTrrr*r*r*r+s0$