o biv_@sddlZddlZddlZddlmZmZmZmZmZm Z m Z ddl Z ddl Z ddl mZmZddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!dd l"m#Z#ddl$m%Z%er~ddl&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)Read) DatasetStats)unify_ref_bundles_schema)BlockMetadataWithSchema) DataContext)omit_traceback_stdout)log_onceStreamingExecutor)Datasetscheduling_strategyc@s|eZdZdZdedefddZdefddZd:d d Z defd d Z de ddefddZ d;ddZ dd0d1Z#defd2d3Z$defd4d5Z%defd6d7Z&defd8d9Z'd/S)? 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*K/home/ubuntu/.local/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. dataset_)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(r1)r)rexecutorr*r*r+create_executorbs zExecutionPlan.create_executorcCsd|jd|jdS)NzExecutionPlan(dataset_uuid=z, snapshot_operator=))r$rr0r*r*r+__repr__js zExecutionPlan.__repr__ dataset_clsrcsddlm}d}d}|s  d(dtdtdtffdd |jj\}}|jd ur5|jj }|j }ne|j d urD|j j }|j j j }nVd }|jj}t |tsct|jd krYd }n |jd}t |trO|rjd }d }n0t |tssJ|ttid d |j} | t|| j| }| }n |j d d}|j }|d urd} n5t |trt|} n+g} t|j|jD]\} } t| dr| j} | | d| qd| } d| d} |d urd}d } ||kr|} | d usJ|j d urd!|j nd}| rd| dnd}d!|j|||| }d}d}d}||}t||kr||d| }t||krg} t|j|jD]A\} } t| drE| j} ||d| d| }t||krsdt| }t"|t||}|d ||}| |q8d| } d | d!||d} |j d urd!||d"|j d#nd}| rd!||d| d#nd}|jd$||d!||d%|d||d| d!|d&}|dkr||7}|S|||d d'|7}|S))zCreate a cosmetic string representation of this execution plan. Returns: The string representation of this execution plan. r)MaterializedDatasetopcurr_strdepthcst|tr ||fS|}|j}|dkr||d7}nd|dd}||d|d7}|jD]}|||d\}}t||}q/||fS)zgTraverse (DFS) the LogicalPlan DAG and return a string representation of the operators.r  r2+- ) isinstancer nameinput_dependenciesmax)r;r<r=curr_max_depthop_nametrailing_spaceinputinput_max_depthgenerate_logical_plan_stringr*r+rLs    zFExecutionPlan.get_plan_as_string..generate_logical_plan_stringNFr2Tmetadataparent)fetch_if_missingzUnknown schema__name__z: z, {}?z name={}, z num_blocks=z{}({}{}num_rows={}, schema={})P z zschema=z...: z, z{ r>zname=,(z num_rows=r6rA)r:r)#ray.data.datasetr9has_computed_outputr strint _logical_plandagr!schemanum_rowsr"rNrBr lenrDrrr(link_logical_planr meta_counttypezipnamestypeshasattrrQappendjoininitial_num_blocksr&formatrE)r)r8r9plan_strplan_max_depthr`counthas_n_ary_operatorr_plan 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_namer*rKr+get_plan_as_stringrs                          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+rc*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) plan_copyr*r*r+copy3s  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)rrrr(r!rr r&rr*r*r+ deep_copyHs 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^r_estimated_num_outputsr0r*r*r+rl\s z ExecutionPlan.initial_num_blocksFrPzpyarrow.lib.SchemacCs|jdur|jSd}|r|jj}n4|jj}|durG|rG|\}}}||D] }|jdur7|j}nq+Wdn1sBwY|||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. N) r#r[r!r`r^r_ infer_schemaexecute_to_iterator cache_schema)r)rPr`iter_ref_bundlesr/r4bundler*r*r+r`bs$      zExecutionPlan.schemar`cCs ||_dSN)r#)r)r`r*r*r+rs zExecutionPlan.cache_schemacCs|jjjS)z1Get the input files of the dataset, if available.)r^r_infer_metadata input_filesr0r*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. cs|]}|jVqdSr)ra).0mr*r*r+ z+ExecutionPlan.meta_count..N)r^r_r[sumr!rNrra)r)r_rar*r*r+rds 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_compatrr5 itertoolschainnext StopIteration get_stats)r)rrr4 bundle_itergenr*r*r+rs     z!ExecutionPlan.execute_to_iteratorpreserve_orderc sd|_|j}tdstdrtd|sddl m }m }t |j jtr[|j jdur[|||j j}|j j}tdd |D}t|}td d |D||d }n?|} || ||j|d } tt| | j| d }Wdn1swY| jdd} |jrt| z%t t!t"j#} | j$j%dkrt&| j$j'_(| j$j)dkrt&| j$j*_+Wnt,y} zt-d| WYd} ~ nd} ~ wwd_.fdd||_/|j j|_0|_1|j|j1_2|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_blocks)rrr*r*r+rrz(ExecutionPlan.execute..cSs$g|]}|jD]\}}||fqqSr*)blocks)rrblockrNr*r*r+ sz)ExecutionPlan.execute..)rr`) 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_statsrO collect_statsrr*r+r's   z,ExecutionPlan.execute..collect_stats)3r'r(rayavailable_resourcesrrloggerwarningr[rrrrBr^r_r output_datarallrr r5r$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_sr]spilled_bytes_totalglobal_bytes_spilledrestore_time_total_srestored_bytes_totalglobal_bytes_restored Exceptiondebugrr!rr r)r)rcontextrroutput_bundlesr`rrr4rstats_summary_stringreplyer*rr+rs             zExecutionPlan.executecCs|jS)zBReturn ``True`` if this plan has been partially or fully executed.)r'r0r*r*r+has_started_execution8sz#ExecutionPlan.has_started_executionNcCsd|_d|_d|_dS)z;Clear the snapshot kept in the plan to the beginning state.N)r!rr r0r*r*r+clear_snapshot=s zExecutionPlan.clear_snapshotcCs|js tiddS|jS)zqReturn stats for this plan. If the plan isn't executed, an empty stats object will be returned. NrM)r rr0r*r*r+rCs zExecutionPlan.statscCstdd|jDS)z/Return whether this plan has lazy input blocks.css|]}t|tVqdSr)rBr)rr;r*r*r+rNsz/ExecutionPlan.has_lazy_input..)rr^sourcesr0r*r*r+has_lazy_inputLszExecutionPlan.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!rr^r_r0r*r*r+r[Ps  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^r_post_order_iterrB)r)rrr;r*r*r+require_preserve_orderYs  z$ExecutionPlan.require_preserve_order)r-r)rr)r-r)F)r-N)(rQ __module__ __qualname____doc__rrr,r\r1r5r7rrrcrrrr]rlboolrrer`rrrrdrrrr rrpropertyrrrrr[rr*r*r*r+r$sV  *  9   " % u   r)/rrloggingtypingrrrrrrrpyarrowrray._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_planr2ray.data._internal.logical.operators.read_operatorrray.data._internal.statsrray.data._internal.utilrray.data.blockrray.data.contextrray.data.exceptionsrray.util.debugrr3rrZrINHERITABLE_REMOTE_ARGS getLoggerrQrrr*r*r*r+s0$