Merge branch 'SERVE-657-update-naming' of github.com:ray-project/ray into SERVE-657-update-naming

Signed-off-by: Omkar Kulkarni <omkar@omkar-JKJHCX74L6.local>

Author: Omkar Kulkarni

.editorconfig

@@ -9,7 +9,7 @@ indent_size = 4
 
 [*.{h,hpp,hxx,cpp,cxx,cc,c}]
 indent_size = 2
-wrap_width = 90
+max_line_length = 90
 
 [*.html]
 indent_size = 2
@@ -22,7 +22,7 @@ indent_size = 2
 
 [*.{py,pyx,pxd,pxi}]
 indent_size = 4
-wrap_width = 79
+max_line_length = 79
 
 [*.sh]
 indent_size = 2

python/ray/data/BUILD

@@ -704,6 +704,20 @@ py_test(
     ],
 )
 
+py_test(
+    name = "test_operator_fusion",
+    size = "medium",
+    srcs = ["tests/test_operator_fusion.py"],
+    tags = [
+        "exclusive",
+        "team:data",
+    ],
+    deps = [
+        ":conftest",
+        "//:ray_lib",
+    ],
+)
+
 py_test(
     name = "test_execution_optimizer",
     size = "medium",

python/ray/data/_internal/logical/optimizers.py

@@ -13,7 +13,7 @@
 from ray.data._internal.logical.rules.inherit_target_max_block_size import (
     InheritTargetMaxBlockSizeRule,
 )
-from ray.data._internal.logical.rules.operator_fusion import OperatorFusionRule
+from ray.data._internal.logical.rules.operator_fusion import FuseOperators
 from ray.data._internal.logical.rules.randomize_blocks import ReorderRandomizeBlocksRule
 from ray.data._internal.logical.rules.set_read_parallelism import SetReadParallelismRule
 from ray.data._internal.logical.rules.zero_copy_map_fusion import (
@@ -35,7 +35,7 @@
     [
         InheritTargetMaxBlockSizeRule,
         SetReadParallelismRule,
-        OperatorFusionRule,
+        FuseOperators,
         EliminateBuildOutputBlocks,
         ConfigureMapTaskMemoryUsingOutputSize,
     ]

python/ray/data/_internal/logical/rules/__init__.py

@@ -1,4 +1,4 @@
-from ray.data._internal.logical.rules.operator_fusion import OperatorFusionRule
+from ray.data._internal.logical.rules.operator_fusion import FuseOperators
 from ray.data._internal.logical.rules.randomize_blocks import ReorderRandomizeBlocksRule
 
-__all__ = ["ReorderRandomizeBlocksRule", "OperatorFusionRule"]
+__all__ = ["ReorderRandomizeBlocksRule", "FuseOperators"]

python/ray/data/_internal/logical/rules/operator_fusion.py

@@ -38,7 +38,7 @@
 INHERITABLE_REMOTE_ARGS = ["scheduling_strategy"]
 
 
-class OperatorFusionRule(Rule):
+class FuseOperators(Rule):
     """Fuses linear chains of compatible physical operators."""
 
     def apply(self, plan: PhysicalPlan) -> PhysicalPlan:
@@ -54,21 +54,21 @@ def apply(self, plan: PhysicalPlan) -> PhysicalPlan:
         # Update output dependencies after fusion.
         # TODO(hchen): Instead of updating the depdencies manually,
         # we need a better abstraction for manipulating the DAG.
-        self._remove_output_depes(fused_dag)
-        self._update_output_depes(fused_dag)
+        self._remove_output_deps(fused_dag)
+        self._update_output_deps(fused_dag)
 
         new_plan = PhysicalPlan(fused_dag, self._op_map, plan.context)
         return new_plan
 
-    def _remove_output_depes(self, op: PhysicalOperator) -> None:
+    def _remove_output_deps(self, op: PhysicalOperator) -> None:
         for input in op._input_dependencies:
             input._output_dependencies = []
-            self._remove_output_depes(input)
+            self._remove_output_deps(input)
 
-    def _update_output_depes(self, op: PhysicalOperator) -> None:
+    def _update_output_deps(self, op: PhysicalOperator) -> None:
         for input in op._input_dependencies:
             input._output_dependencies.append(op)
-            self._update_output_depes(input)
+            self._update_output_deps(input)
 
     def _fuse_map_operators_in_dag(self, dag: PhysicalOperator) -> MapOperator:
         """Starting at the given operator, traverses up the DAG of operators
@@ -305,6 +305,7 @@ def _get_fused_map_operator(
         # Merge minimum block sizes.
         down_min_rows_per_bundled_input = down_logical_op._min_rows_per_bundled_input
         up_min_rows_per_bundled_input = up_logical_op._min_rows_per_bundled_input
+
         if (
             down_min_rows_per_bundled_input is not None
             and up_min_rows_per_bundled_input is not None

python/ray/data/_internal/logical/rules/zero_copy_map_fusion.py

@@ -10,7 +10,7 @@
 )
 from ray.data._internal.logical.interfaces.optimizer import Rule
 from ray.data._internal.logical.interfaces.physical_plan import PhysicalPlan
-from ray.data._internal.logical.rules.operator_fusion import OperatorFusionRule
+from ray.data._internal.logical.rules.operator_fusion import FuseOperators
 
 
 class ZeroCopyMapFusionRule(Rule):
@@ -27,7 +27,7 @@ class ZeroCopyMapFusionRule(Rule):
 
     @classmethod
     def dependencies(cls) -> List[Type[Rule]]:
-        return [OperatorFusionRule]
+        return [FuseOperators]
 
     def apply(self, plan: PhysicalPlan) -> PhysicalPlan:
         self._traverse(plan.dag)