We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Submit query condition to remote node instead of fetch to local then query
<!--
Thank you for your contribution!

Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
-->

**Is your feature request related to a problem? Please describe.**
Curently if ray fetcher gets objects with condition, it will fetch objects to local, then filter local objects with conditions. This will incur much objects transfer cost and high memory footsprint in local node. 

**Describe the solution you'd like**
We should submit query to objects' node then fetch queried result instead. 



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mars/services/task/execution/ray/fetcher.py`
Content:
```
1 # Copyright 1999-2021 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import asyncio
16 from collections import namedtuple
17 from typing import Dict, List
18 from ..api import Fetcher, register_fetcher_cls
19 
20 
21 _FetchInfo = namedtuple("FetchInfo", ["key", "object_ref", "conditions"])
22 
23 
24 @register_fetcher_cls
25 class RayFetcher(Fetcher):
26     name = "ray"
27     required_meta_keys = ("object_refs",)
28 
29     def __init__(self, **kwargs):
30         self._fetch_info_list = []
31         self._no_conditions = True
32 
33     async def append(self, chunk_key: str, chunk_meta: Dict, conditions: List = None):
34         if conditions is not None:
35             self._no_conditions = False
36         self._fetch_info_list.append(
37             _FetchInfo(chunk_key, chunk_meta["object_refs"][0], conditions)
38         )
39 
40     async def get(self):
41         objects = await asyncio.gather(
42             *(info.object_ref for info in self._fetch_info_list)
43         )
44         if self._no_conditions:
45             return objects
46         results = []
47         for o, fetch_info in zip(objects, self._fetch_info_list):
48             if fetch_info.conditions is None:
49                 results.append(o)
50             else:
51                 try:
52                     results.append(o.iloc[fetch_info.conditions])
53                 except AttributeError:
54                     results.append(o[fetch_info.conditions])
55         return results
56
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mars/services/task/execution/ray/fetcher.py b/mars/services/task/execution/ray/fetcher.py
--- a/mars/services/task/execution/ray/fetcher.py
+++ b/mars/services/task/execution/ray/fetcher.py
@@ -15,9 +15,10 @@
 import asyncio
 from collections import namedtuple
 from typing import Dict, List
+from .....utils import lazy_import
 from ..api import Fetcher, register_fetcher_cls
 
-
+ray = lazy_import("ray")
 _FetchInfo = namedtuple("FetchInfo", ["key", "object_ref", "conditions"])
 
 
@@ -38,18 +39,27 @@
         )
 
     async def get(self):
-        objects = await asyncio.gather(
-            *(info.object_ref for info in self._fetch_info_list)
-        )
         if self._no_conditions:
-            return objects
-        results = []
-        for o, fetch_info in zip(objects, self._fetch_info_list):
+            return await asyncio.gather(
+                *(info.object_ref for info in self._fetch_info_list)
+            )
+        refs = [None] * len(self._fetch_info_list)
+        for index, fetch_info in enumerate(self._fetch_info_list):
             if fetch_info.conditions is None:
-                results.append(o)
+                refs[index] = fetch_info.object_ref
             else:
-                try:
-                    results.append(o.iloc[fetch_info.conditions])
-                except AttributeError:
-                    results.append(o[fetch_info.conditions])
-        return results
+                refs[index] = query_object_with_condition.remote(
+                    fetch_info.object_ref, fetch_info.conditions
+                )
+        return await asyncio.gather(*refs)
+
+
+def query_object_with_condition(o, conditions):
+    try:
+        return o.iloc[conditions]
+    except AttributeError:
+        return o[conditions]
+
+
+if ray:
+    query_object_with_condition = ray.remote(query_object_with_condition)

{"golden_diff": "diff --git a/mars/services/task/execution/ray/fetcher.py b/mars/services/task/execution/ray/fetcher.py\n--- a/mars/services/task/execution/ray/fetcher.py\n+++ b/mars/services/task/execution/ray/fetcher.py\n@@ -15,9 +15,10 @@\n import asyncio\n from collections import namedtuple\n from typing import Dict, List\n+from .....utils import lazy_import\n from ..api import Fetcher, register_fetcher_cls\n \n-\n+ray = lazy_import(\"ray\")\n _FetchInfo = namedtuple(\"FetchInfo\", [\"key\", \"object_ref\", \"conditions\"])\n \n \n@@ -38,18 +39,27 @@\n         )\n \n     async def get(self):\n-        objects = await asyncio.gather(\n-            *(info.object_ref for info in self._fetch_info_list)\n-        )\n         if self._no_conditions:\n-            return objects\n-        results = []\n-        for o, fetch_info in zip(objects, self._fetch_info_list):\n+            return await asyncio.gather(\n+                *(info.object_ref for info in self._fetch_info_list)\n+            )\n+        refs = [None] * len(self._fetch_info_list)\n+        for index, fetch_info in enumerate(self._fetch_info_list):\n             if fetch_info.conditions is None:\n-                results.append(o)\n+                refs[index] = fetch_info.object_ref\n             else:\n-                try:\n-                    results.append(o.iloc[fetch_info.conditions])\n-                except AttributeError:\n-                    results.append(o[fetch_info.conditions])\n-        return results\n+                refs[index] = query_object_with_condition.remote(\n+                    fetch_info.object_ref, fetch_info.conditions\n+                )\n+        return await asyncio.gather(*refs)\n+\n+\n+def query_object_with_condition(o, conditions):\n+    try:\n+        return o.iloc[conditions]\n+    except AttributeError:\n+        return o[conditions]\n+\n+\n+if ray:\n+    query_object_with_condition = ray.remote(query_object_with_condition)\n", "issue": "Submit query condition to remote node instead of fetch to local then query\n<!--\r\nThank you for your contribution!\r\n\r\nPlease review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.\r\n-->\r\n\r\n**Is your feature request related to a problem? Please describe.**\r\nCurently if ray fetcher gets objects with condition, it will fetch objects to local, then filter local objects with conditions. This will incur much objects transfer cost and high memory footsprint in local node. \r\n\r\n**Describe the solution you'd like**\r\nWe should submit query to objects' node then fetch queried result instead. \r\n\r\n\n", "before_files": [{"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport asyncio\nfrom collections import namedtuple\nfrom typing import Dict, List\nfrom ..api import Fetcher, register_fetcher_cls\n\n\n_FetchInfo = namedtuple(\"FetchInfo\", [\"key\", \"object_ref\", \"conditions\"])\n\n\n@register_fetcher_cls\nclass RayFetcher(Fetcher):\n    name = \"ray\"\n    required_meta_keys = (\"object_refs\",)\n\n    def __init__(self, **kwargs):\n        self._fetch_info_list = []\n        self._no_conditions = True\n\n    async def append(self, chunk_key: str, chunk_meta: Dict, conditions: List = None):\n        if conditions is not None:\n            self._no_conditions = False\n        self._fetch_info_list.append(\n            _FetchInfo(chunk_key, chunk_meta[\"object_refs\"][0], conditions)\n        )\n\n    async def get(self):\n        objects = await asyncio.gather(\n            *(info.object_ref for info in self._fetch_info_list)\n        )\n        if self._no_conditions:\n            return objects\n        results = []\n        for o, fetch_info in zip(objects, self._fetch_info_list):\n            if fetch_info.conditions is None:\n                results.append(o)\n            else:\n                try:\n                    results.append(o.iloc[fetch_info.conditions])\n                except AttributeError:\n                    results.append(o[fetch_info.conditions])\n        return results\n", "path": "mars/services/task/execution/ray/fetcher.py"}], "after_files": [{"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport asyncio\nfrom collections import namedtuple\nfrom typing import Dict, List\nfrom .....utils import lazy_import\nfrom ..api import Fetcher, register_fetcher_cls\n\nray = lazy_import(\"ray\")\n_FetchInfo = namedtuple(\"FetchInfo\", [\"key\", \"object_ref\", \"conditions\"])\n\n\n@register_fetcher_cls\nclass RayFetcher(Fetcher):\n    name = \"ray\"\n    required_meta_keys = (\"object_refs\",)\n\n    def __init__(self, **kwargs):\n        self._fetch_info_list = []\n        self._no_conditions = True\n\n    async def append(self, chunk_key: str, chunk_meta: Dict, conditions: List = None):\n        if conditions is not None:\n            self._no_conditions = False\n        self._fetch_info_list.append(\n            _FetchInfo(chunk_key, chunk_meta[\"object_refs\"][0], conditions)\n        )\n\n    async def get(self):\n        if self._no_conditions:\n            return await asyncio.gather(\n                *(info.object_ref for info in self._fetch_info_list)\n            )\n        refs = [None] * len(self._fetch_info_list)\n        for index, fetch_info in enumerate(self._fetch_info_list):\n            if fetch_info.conditions is None:\n                refs[index] = fetch_info.object_ref\n            else:\n                refs[index] = query_object_with_condition.remote(\n                    fetch_info.object_ref, fetch_info.conditions\n                )\n        return await asyncio.gather(*refs)\n\n\ndef query_object_with_condition(o, conditions):\n    try:\n        return o.iloc[conditions]\n    except AttributeError:\n        return o[conditions]\n\n\nif ray:\n    query_object_with_condition = ray.remote(query_object_with_condition)\n", "path": "mars/services/task/execution/ray/fetcher.py"}]}