o $iC$@srddlmZmZmZmZmZmZmZddlm Z ddl m Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZdd lmZmZmZdd lmZdd lmZdd l m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'm(Z(m)Z)m*Z*ddl+m,Z,m-Z-ddl.m/Z/m0Z0ddl1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>m?Z?m@Z@mAZAddlBmCZCddlDmEZEededZFeeFee eEge fZGde#dee d eEd!e fd"d#ZHd$e!dee d eEd!e fd%d&ZId'd(ZJd)d*ZKd+d,ZLd-d.ZMde%dee d eEd!e fd/d0ZNde4dee d eEfd1d2ZOGd3d4d4ZPded5eeeeGfd!eeGfd6d7ZQd8S)9)CallableDictListOptionalTupleTypeTypeVar)PhysicalOperator)AggregateNumRows)InputDataBuffer) JoinOperator) LimitOperator)OutputSplitter) UnionOperator) ZipOperator)LogicalOperator LogicalPlan PhysicalPlan)AbstractAllToAll)Count) AbstractFrom) InputData)Join)AbstractUDFMapFilterProjectStreamingRepartition)UnionZip)DownloadLimit)Read)StreamingSplit)Write)plan_all_to_all_op)plan_download_op) plan_read_op)plan_filter_opplan_project_opplan_streaming_repartition_opplan_udf_map_op) plan_write_op) DataContextLogicalOperatorType)bound logical_opphysical_children data_contextreturncCst|dksJt||jdS)z>Get the corresponding DAG of physical operators for InputData.r) input_datalenr r3r/r0r1r7_/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/_internal/planner/planner.pyplan_input_data_op8s r9opcCst|dksJt||jS)Nrr4)r:r0r1r7r7r8 plan_from_opFs r;cC t|dksJt|g|RSN)r5r_r0r1r7r7r8 plan_zip_opOrAcCr<r=)r5rr?r7r7r8 plan_union_opTrBrCcCs"t|dksJt|j|d|S)Nr)r5r _limitr6r7r7r8 plan_limit_opYsrFcCs&t|dksJt|dg|tjdS)NrDr) column_name)r5r r COLUMN_NAMEr6r7r7r8 plan_count_op^srIc CsFt|dksJt||d|d|j|j|j|j|j|j|j|j d S)Nr>rrD) r1 left_input_opright_input_op join_typeleft_key_columnsright_key_columnsleft_columns_suffixright_columns_suffixnum_partitionspartition_size_hint#aggregator_ray_remote_args_override) r5r _join_type_left_key_columns_right_key_columns_left_columns_suffix_right_columns_suffix _num_outputs_partition_size_hint_aggregator_ray_remote_argsr6r7r7r8 plan_join_opesr\cCs,t|dksJt|d|j|j||jdS)NrDr)nequalr1locality_hints)r5r _num_splits_equal_locality_hintsr6r7r7r8plan_streaming_split_opzsrcc @seZdZdZieeeeee e e e e eeeeeeeeeeeeeeeeeee e!e"e#Z$de%de&fddZ'de(de)fddZ*de(d e+de,e-e.e(e-fffd d Z/d S) 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_planr2cCs&||j|j\}}t|||j}|S)z@Convert logical to physical operators recursively in post-order.)_plan_recursivelydagcontextr)selfre physical_dagop_map physical_planr7r7r8plans z Planner.planr/cCs&t||j}|dur |Std|)Nz0Found unknown logical operator during planning: ) find_plan_fn_DEFAULT_PLAN_FNS ValueError)rir/plan_fnr7r7r8 get_plan_fns zPlanner.get_plan_fnr1c 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_dependenciesrfappendupdaterrpop_logical_operatorsset_logical_operatorsextend) rir/r1rkr0childphysical_child child_op_maprq physical_opqueuecurr_physical_opr7r7r8rfs&         zPlanner._plan_recursivelyN)0__name__ __module__ __qualname____doc__r!r&rr9r#r+rr;rr'rr*rr$rrCrrAr rFrrIrr(rr)rr\r"rcrr%rorrrmrPlanLogicalOpFnrrr,rr rrfr7r7r7r8rdsX      rdplan_fnscCs(|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)items isinstance)r/rop_typerqr7r7r8rns  rnN)Rtypingrrrrrrr'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.joinr 5ray.data._internal.execution.operators.limit_operatorr 6ray.data._internal.execution.operators.output_splitterr5ray.data._internal.execution.operators.union_operatorr3ray.data._internal.execution.operators.zip_operatorr%ray.data._internal.logical.interfacesrrr8ray.data._internal.logical.operators.all_to_all_operatorr3ray.data._internal.logical.operators.count_operatorr3ray.data._internal.logical.operators.from_operatorsr8ray.data._internal.logical.operators.input_data_operatorr2ray.data._internal.logical.operators.join_operatorr1ray.data._internal.logical.operators.map_operatorrrrr3ray.data._internal.logical.operators.n_ary_operatorrr8ray.data._internal.logical.operators.one_to_one_operatorrr 2ray.data._internal.logical.operators.read_operatorr!=ray.data._internal.logical.operators.streaming_split_operatorr"3ray.data._internal.logical.operators.write_operatorr#-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.contextr,r-rr9r;rArCrFrIr\rcrdrnr7r7r7r8s$                           Z