o `۷i.@sfddlZddlZddlmZmZmZmZmZmZm Z ddl m Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZm Z m!Z!ddl"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2ddl3m4Z4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;ddlZ>m?Z?m@Z@ddlAmBZBddlCmDZDddlEmFZFe dedZGeeGeeeFgefZHde)deedeFdefddZIde$deedeFdefd d!ZJd"d#ZKd$d%ZLd&d'ZMd(d)ZNde*deedeFdefd*d+ZOde/deedeFfd,d-ZPGd.d/d/ZQded0eeeeHfdeeHfd1d2ZRdS)3N)CallableDictListOptionalTupleTypeTypeVar) ObjectRef)add_execution_callback)PhysicalOperator)AggregateNumRows)InputDataBuffer) JoinOperator) LimitOperator)OutputSplitter) UnionOperator) ZipOperator)LogicalOperator LogicalPlan PhysicalPlan)AbstractAllToAll AbstractFromAbstractUDFMapCountDownloadFilter InputDataJoinLimitProjectReadStreamingRepartitionStreamingSplitUnionWriteZip)#plan_read_op_with_checkpoint_filter$plan_write_op_with_checkpoint_writer)plan_all_to_all_op)plan_download_op) plan_read_op)plan_filter_opplan_project_opplan_streaming_repartition_opplan_udf_map_op) plan_write_opLoadCheckpointCallback) DataContextLogicalOperatorType)bound logical_opphysical_children data_contextreturncCst|dksJt||jdS)z>Get the corresponding DAG of physical operators for InputData.r) input_datalenr r9r5r6r7r=X/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/data/_internal/planner/planner.pyplan_input_data_opAs r?opcCst|dksJt||jS)Nrr:)r@r6r7r=r=r> plan_from_opOs rAcC t|dksJt|g|RSN)r;r_r6r7r=r=r> plan_zip_opXrGcCrBrC)r;rrEr=r=r> plan_union_op]rHrIcCs"t|dksJt|j|d|S)Nr)r;rlimitr<r=r=r> plan_limit_opbsrLcCs&t|dksJt|dg|tjdS)NrJr) column_name)r;r r COLUMN_NAMEr<r=r=r> plan_count_opgsrOc CsFt|dksJt||d|d|j|j|j|j|j|j|j|j d S)NrDrrJ) r7 left_input_opright_input_op join_typeleft_key_columnsright_key_columnsleft_columns_suffixright_columns_suffixnum_partitionspartition_size_hint#aggregator_ray_remote_args_override) r;rrRrSrTrUrV num_outputsrXaggregator_ray_remote_argsr<r=r=r> plan_join_opnsr\cCs,t|dksJt|d|j|j||jdS)NrJr)nequalr7locality_hints)r;r num_splitsr^r_r<r=r=r>plan_streaming_split_opsrac @seZdZdZieeeeee e e e e eeeeeeeeeeeeeeeeeee e!e"e#Z$efZ%ddZ&de'de(fddZ)de*de+fd d Z,de*d e-de.e/e0e*e/fffd d Z1de2fddZ3de4ge5fde0e6e*e+ffddZ7de'de8fddZ9dS)PlannerzThe planner to convert optimized logical to physical operators. Note that planner is only doing operators conversion. Physical optimization work is done by physical optimizer. cCsd|_i|_dS)NF)_supports_checkpointing_plan_fns_for_checkpointing)selfr=r=r>__init__s zPlanner.__init__ logical_planr8cCs|jj}|dur%||r%d|_||}t||j|j}|||_n|dur5||r0Jt d| |j |j\}}t |||j}|S)z@Convert logical to physical operators recursively in post-order.NTzqYou've enabled checkpointing, but the logical plan doesn't support checkpointing. Checkpointing will be disabled.)contextcheckpoint_config_check_supports_checkpointingrc_create_checkpoint_callbackr load_checkpoint_get_plan_fns_for_checkpointingrdwarningswarn_plan_recursivelydagr)rergricheckpoint_callbackrl physical_dagop_map physical_planr=r=r>plans*   z Planner.planr5cCsN|jr|jsJt||j}|dur|St||j}|dur |Std|)Nz0Found unknown logical operator during planning: )rcrd find_plan_fn_DEFAULT_PLAN_FNS ValueError)rer5plan_fnr=r=r> get_plan_fns   zPlanner.get_plan_fnr7c Csi}g}|jD]}|||\}}||||q||}||||} | g} | rE| } | jr4n| |||| <| | j| s,||| <| |fS)a]Plan a logical operator and its input dependencies recursively. Args: logical_op: The logical operator to plan. data_context: The data context. Returns: A tuple of the physical operator corresponding to the logical operator, and a mapping from physical to logical operators. ) input_dependenciesrpappendupdater{pop_logical_operatorsset_logical_operatorsextend) rer5r7rtr6childphysical_child child_op_maprz physical_opqueuecurr_physical_opr=r=r>rps&         zPlanner._plan_recursivelycCst|S)zFactory method to create the LoadCheckpointCallback. Subclasses can override this to use a different callback implementation. r0)rerir=r=r>rk sz#Planner._create_checkpoint_callbackrlcCsttjt|dtti}|S)N)rl)r functoolspartialr&r$r')rerlplan_fnsr=r=r>rmsz'Planner._get_plan_fns_for_checkpointingcs6t|jttfs dSdtdtffdd |jS)zCheck if the logical plan supports checkpointing. Subclasses can override _CHECKPOINT_FILTER_OPS to support more operators. Fr@r8cs(t|jrdStfdd|jDS)NTc3s|]}|VqdS)Nr=).0 input_dep)$_all_paths_contain_checkpoint_filterr=r> ,s  zfPlanner._check_supports_checkpointing.._all_paths_contain_checkpoint_filter..) isinstance_CHECKPOINT_FILTER_OPSallr|)r@rrer=r>r)s  zSPlanner._check_supports_checkpointing.._all_paths_contain_checkpoint_filter)rrqr$r"rbool)rergr=rr>rj!s z%Planner._check_supports_checkpointingN):__name__ __module__ __qualname____doc__r r*rr?r$r/rrArr+rr.rr(r#rIr%rGrrLrrOrr,r!r-rr\r"rarr)rxrrfrrrvrPlanLogicalOpFnr{r2rr rrpr1rkrr rrmrrjr=r=r=r>rbsj      .   rbrcCs(|D] \}}t||r|SqdS)aFind the plan function for a logical operator. This function goes through the plan functions in order and returns the first one that is an instance of the logical operator type. Args: logical_op: The logical operator to find the plan function for. plan_fns: The dictionary of plan functions. Returns: The plan function for the logical operator, or None if no plan function is found. N)itemsr)r5rop_typerzr=r=r>rw4s  rw)Srrntypingrrrrrrrrayr /ray.data._internal.execution.execution_callbackr 'ray.data._internal.execution.interfacesr 9ray.data._internal.execution.operators.aggregate_num_rowsr 8ray.data._internal.execution.operators.input_data_bufferr +ray.data._internal.execution.operators.joinr5ray.data._internal.execution.operators.limit_operatorr6ray.data._internal.execution.operators.output_splitterr5ray.data._internal.execution.operators.union_operatorr3ray.data._internal.execution.operators.zip_operatorr%ray.data._internal.logical.interfacesrrr$ray.data._internal.logical.operatorsrrrrrrrrrrr r!r"r#r$r%%ray.data._internal.planner.checkpointr&r'-ray.data._internal.planner.plan_all_to_all_opr(+ray.data._internal.planner.plan_download_opr)'ray.data._internal.planner.plan_read_opr**ray.data._internal.planner.plan_udf_map_opr+r,r-r.(ray.data._internal.planner.plan_write_opr/,ray.data.checkpoint.load_checkpoint_callbackr1ray.data.contextr2r3rr?rArGrIrLrOr\rarbrwr=r=r=r>s$          H            #