o bi# @s,UddlmZmZmZmZmZmZddlmZddl m Z ddl m Z m Z mZddlmZddlmZddlmZede d ZeeeeegefZiZeee efed <ed ee d efd dZeddZddZeGdddZede deee efdedeeeee fffddZdS))CallableDictListTupleTypeTypeVar)PhysicalOperator) JoinOperator)LogicalOperator LogicalPlan PhysicalPlan)Join) DataContext) DeveloperAPILogicalOperatorType)bound_PLAN_LOGICAL_OP_FNSlogical_op_typeplan_fncCs |t|<dS)z5Register a plan function for a logical operator type.N)r)rrrV/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/planner/planner.pyregister_plan_logical_op_fns rcCstS)N)rcopyrrrrget_plan_logical_op_fns!srcsddlmddlmddlmddlmddlm ddl m }ddl m dd lm}dd lm}dd lm}m}m}m}dd lm}m}dd lm} ddlm} ddlm} ddl m!} ddl"m#} ddl$m%}m&}m'}m(}ddl)m*}t+| | d|dt,t-dt.dt-ffdd }t+||t+| |d|dt,t-dt.dt-ffdd }t+||t+||t+||t+|| fdd}t+||fdd }t+||fd!d"}t+| |fd#d$}t+|t+||t+||dt/dt,t-dt.dt-fd%d&}t+t/|dS)'Nr)AggregateNumRowsInputDataBuffer LimitOperator UnionOperator ZipOperator)AbstractAllToAll)Count) AbstractFrom) InputData)AbstractUDFMapFilterProjectStreamingRepartition)UnionZip)Limit)Read)Write)plan_all_to_all_op) plan_read_op)plan_filter_opplan_project_opplan_streaming_repartition_opplan_udf_map_op) plan_write_op logical_opphysical_children data_contextreturncst|dksJ||jdS)z>Get the corresponding DAG of physical operators for InputData.r) input_datalenr;r7r8r9rrrplan_input_data_opJs zA_register_default_plan_logical_op_fns..plan_input_data_opopcst|dksJ||jS)Nrr<)r@r8r9rrr plan_from_opZs z;_register_default_plan_logical_op_fns..plan_from_opcs$t|dksJ|d|d|S)Nrr=_r8r9r!rr plan_zip_opjsz:_register_default_plan_logical_op_fns..plan_zip_opcs t|dksJ|g|RS)NrBrDrErrr plan_union_oppsz<_register_default_plan_logical_op_fns..plan_union_opcs"t|dksJ|j|d|S)NrCr)r=_limitr>rrr plan_limit_opvsz<_register_default_plan_logical_op_fns..plan_limit_opcs&t|dksJ|dg|jdS)NrCr) column_name)r= COLUMN_NAMEr>)rr$rr plan_count_op|sz<_register_default_plan_logical_op_fns..plan_count_opc SsTt|dksJ|jdusJt||d|d|j|j|j|j|j|j|j|j d S)NrBrrC) r9 left_input_opright_input_op join_typeleft_key_columnsright_key_columnsleft_columns_suffixright_columns_suffixnum_partitionspartition_size_hint#aggregator_ray_remote_args_override) r= _num_outputsr _join_type_left_key_columns_right_key_columns_left_columns_suffix_right_columns_suffix_partition_size_hint_aggregator_ray_remote_argsr>rrr plan_join_opsz;_register_default_plan_logical_op_fns..plan_join_op)09ray.data._internal.execution.operators.aggregate_num_rowsr8ray.data._internal.execution.operators.input_data_bufferr5ray.data._internal.execution.operators.limit_operatorr5ray.data._internal.execution.operators.union_operatorr 3ray.data._internal.execution.operators.zip_operatorr"8ray.data._internal.logical.operators.all_to_all_operatorr#3ray.data._internal.logical.operators.count_operatorr$3ray.data._internal.logical.operators.from_operatorsr%8ray.data._internal.logical.operators.input_data_operatorr&1ray.data._internal.logical.operators.map_operatorr'r(r)r*3ray.data._internal.logical.operators.n_ary_operatorr+r,8ray.data._internal.logical.operators.one_to_one_operatorr-2ray.data._internal.logical.operators.read_operatorr.3ray.data._internal.logical.operators.write_operatorr/-ray.data._internal.planner.plan_all_to_all_opr0'ray.data._internal.planner.plan_read_opr1*ray.data._internal.planner.plan_udf_map_opr2r3r4r5(ray.data._internal.planner.plan_write_opr6rrrrr )r#r%r&r'r(r)r*r+r,r-r.r/r0r1r2r3r4r5r6r?rArGrHrJrMr`r)rr$rrr r"r%_register_default_plan_logical_op_fns&s~                               rsc@s"eZdZdZdedefddZdS)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. logical_planr:cCs,t}t|j||j\}}t|||j}|S)z@Convert logical to physical operators recursively in post-order.)rplan_recursivelydagcontextr )selfruplan_fns physical_dagop_map physical_planrrrplans  z Planner.planN)__name__ __module__ __qualname____doc__r r r~rrrrrtsrtr7rzr9r:c Csi}g}|jD]}t|||\}}||||qd}|D]\} } t|| r3| |||}nq"|dur?td||g} | rY| } | jrLn | || |j| sD|||<||fS)aPlan a logical operator and its input dependencies recursively. Args: logical_op: The logical operator to plan. plan_fns: A dictionary of planning functions for different logical operator types. 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. Nz0Found unknown logical operator during planning: ) input_dependenciesrvappendupdateitems isinstance ValueErrorpop_logical_operatorsset_logical_operatorsextend) r7rzr9r|r8childphysical_child child_op_map physical_opop_typerqueuecurr_physical_oprrrrvs4        rvN)typingrrrrrr'ray.data._internal.execution.interfacesr+ray.data._internal.execution.operators.joinr %ray.data._internal.logical.interfacesr r r 2ray.data._internal.logical.operators.join_operatorr ray.data.contextrray.util.annotationsrrPlanLogicalOpFnr__annotations__rrrsrtrvrrrrsB"        {