prompt: 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()
  
  ```
  
golden_diff: 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)
  
verification_info: {"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"}]}