prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  Implement `map` for `LocalExecutor`
  For some reason we avoided doing this, but it's actually entirely possible to do!  Would be great for local debugging.
  
  </issue>
  <code>
  [start of src/prefect/engine/executors/local.py]
  1 # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula
  2 
  3 from prefect.engine.executors.base import Executor
  4 
  5 
  6 class LocalExecutor(Executor):
  7     """
  8     An executor that runs all functions synchronously and immediately in
  9     the local thread.  To be used mainly for debugging purposes.
  10     """
  11 
  12     def submit(self, fn, *args, **kwargs):
  13         """
  14         Submit a function to the executor for execution. Returns the result of the computation.
  15 
  16         Args:
  17             - fn (Callable): function which is being submitted for execution
  18             - *args (Any): arguments to be passed to `fn`
  19             - **kwargs (Any): keyword arguments to be passed to `fn`
  20 
  21         Returns:
  22             - Any: the result of `fn(*args, **kwargs)`
  23         """
  24         return fn(*args, **kwargs)
  25 
  26     def wait(self, futures, timeout=None):
  27         """
  28         Returns:
  29             - Any: whatever `futures` were provided
  30         """
  31         return futures
  32
  [end of src/prefect/engine/executors/local.py]
  [start of src/prefect/engine/executors/__init__.py]
  1 # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula
  2 
  3 """
  4 Prefect Executors implement the logic for how Tasks are run. The standard interface
  5 for an Executor consists of the following methods:
  6 
  7 - `submit(fn, *args, **kwargs)`: submit `fn(*args, **kwargs)` for execution;
  8     note that this function is (in general) non-blocking, meaning that `executor.submit(...)`
  9     will _immediately_ return a future-like object regardless of whether `fn(*args, **kwargs)`
  10     has completed running
  11 - `submit_with_context(fn, *args, context, **kwargs)`: submit `fn(*args,
  12     **kwargs)` for execution with the provided `prefect.context`
  13 - `wait(object)`: resolves any objects returned by `executor.submit` to
  14     their values; this function _will_ block until execution of `object` is complete
  15 - `map(fn, *args, upstream_states, **kwargs)`: submit function to be mapped
  16     over based on the edge information contained in `upstream_states`.  Any "mapped" Edge
  17     will be converted into multiple function submissions, one for each value of the upstream mapped tasks.
  18 
  19 Currently, the available executor options are:
  20 
  21 - `LocalExecutor`: the no frills, straightforward executor - great for simple
  22     debugging; tasks are executed immediately upon being called by `executor.submit()`.
  23     Note that the `map` feature is currently _not_ supported with this executor.
  24 - `SynchronousExecutor`: an executor that runs on `dask` primitives with the
  25     synchronous dask scheduler; currently the default executor
  26 - `DaskExecutor`: the most feature-rich of the executors, this executor runs
  27     on `dask.distributed` and has support for multiprocessing, multithreading, and distributed execution.
  28 
  29 Which executor you choose depends on whether you intend to use things like parallelism
  30 of task execution.
  31 """
  32 import sys
  33 
  34 from warnings import warn as _warn
  35 from importlib import import_module as _import_module
  36 
  37 import prefect as _prefect
  38 from prefect.engine.executors.base import Executor
  39 from prefect.engine.executors.local import LocalExecutor
  40 from prefect.engine.executors.sync import SynchronousExecutor
  41 
  42 if sys.version_info >= (3, 5):
  43     from prefect.engine.executors.dask import DaskExecutor
  44 
  45 try:
  46     cfg_exec = _prefect.config.engine.executor
  47     *module, cls_name = cfg_exec.split(".")
  48     module = _import_module(".".join(module))
  49     DEFAULT_EXECUTOR = getattr(module, cls_name)()
  50 except:
  51     _warn(
  52         "Could not import {}, using prefect.engine.executors.LocalExecutor instead.".format(
  53             _prefect.config.engine.executor
  54         )
  55     )
  56     DEFAULT_EXECUTOR = LocalExecutor()
  57
  [end of src/prefect/engine/executors/__init__.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/src/prefect/engine/executors/__init__.py b/src/prefect/engine/executors/__init__.py
  --- a/src/prefect/engine/executors/__init__.py
  +++ b/src/prefect/engine/executors/__init__.py
  @@ -20,7 +20,6 @@
   
   - `LocalExecutor`: the no frills, straightforward executor - great for simple
       debugging; tasks are executed immediately upon being called by `executor.submit()`.
  -    Note that the `map` feature is currently _not_ supported with this executor.
   - `SynchronousExecutor`: an executor that runs on `dask` primitives with the
       synchronous dask scheduler; currently the default executor
   - `DaskExecutor`: the most feature-rich of the executors, this executor runs
  diff --git a/src/prefect/engine/executors/local.py b/src/prefect/engine/executors/local.py
  --- a/src/prefect/engine/executors/local.py
  +++ b/src/prefect/engine/executors/local.py
  @@ -1,6 +1,9 @@
   # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula
   
  +from typing import Any, Callable, Iterable
  +
   from prefect.engine.executors.base import Executor
  +from prefect.utilities.executors import dict_to_list
   
   
   class LocalExecutor(Executor):
  @@ -9,6 +12,17 @@
       the local thread.  To be used mainly for debugging purposes.
       """
   
  +    def map(
  +        self, fn: Callable, *args: Any, upstream_states=None, **kwargs: Any
  +    ) -> Iterable[Any]:
  +
  +        states = dict_to_list(upstream_states)
  +        results = []
  +        for elem in states:
  +            results.append(self.submit(fn, *args, upstream_states=elem, **kwargs))
  +
  +        return results
  +
       def submit(self, fn, *args, **kwargs):
           """
           Submit a function to the executor for execution. Returns the result of the computation.
  
verification_info: {"golden_diff": "diff --git a/src/prefect/engine/executors/__init__.py b/src/prefect/engine/executors/__init__.py\n--- a/src/prefect/engine/executors/__init__.py\n+++ b/src/prefect/engine/executors/__init__.py\n@@ -20,7 +20,6 @@\n \n - `LocalExecutor`: the no frills, straightforward executor - great for simple\n     debugging; tasks are executed immediately upon being called by `executor.submit()`.\n-    Note that the `map` feature is currently _not_ supported with this executor.\n - `SynchronousExecutor`: an executor that runs on `dask` primitives with the\n     synchronous dask scheduler; currently the default executor\n - `DaskExecutor`: the most feature-rich of the executors, this executor runs\ndiff --git a/src/prefect/engine/executors/local.py b/src/prefect/engine/executors/local.py\n--- a/src/prefect/engine/executors/local.py\n+++ b/src/prefect/engine/executors/local.py\n@@ -1,6 +1,9 @@\n # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\n \n+from typing import Any, Callable, Iterable\n+\n from prefect.engine.executors.base import Executor\n+from prefect.utilities.executors import dict_to_list\n \n \n class LocalExecutor(Executor):\n@@ -9,6 +12,17 @@\n     the local thread.  To be used mainly for debugging purposes.\n     \"\"\"\n \n+    def map(\n+        self, fn: Callable, *args: Any, upstream_states=None, **kwargs: Any\n+    ) -> Iterable[Any]:\n+\n+        states = dict_to_list(upstream_states)\n+        results = []\n+        for elem in states:\n+            results.append(self.submit(fn, *args, upstream_states=elem, **kwargs))\n+\n+        return results\n+\n     def submit(self, fn, *args, **kwargs):\n         \"\"\"\n         Submit a function to the executor for execution. Returns the result of the computation.\n", "issue": "Implement `map` for `LocalExecutor`\nFor some reason we avoided doing this, but it's actually entirely possible to do!  Would be great for local debugging.\n", "before_files": [{"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\n\nfrom prefect.engine.executors.base import Executor\n\n\nclass LocalExecutor(Executor):\n    \"\"\"\n    An executor that runs all functions synchronously and immediately in\n    the local thread.  To be used mainly for debugging purposes.\n    \"\"\"\n\n    def submit(self, fn, *args, **kwargs):\n        \"\"\"\n        Submit a function to the executor for execution. Returns the result of the computation.\n\n        Args:\n            - fn (Callable): function which is being submitted for execution\n            - *args (Any): arguments to be passed to `fn`\n            - **kwargs (Any): keyword arguments to be passed to `fn`\n\n        Returns:\n            - Any: the result of `fn(*args, **kwargs)`\n        \"\"\"\n        return fn(*args, **kwargs)\n\n    def wait(self, futures, timeout=None):\n        \"\"\"\n        Returns:\n            - Any: whatever `futures` were provided\n        \"\"\"\n        return futures\n", "path": "src/prefect/engine/executors/local.py"}, {"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\n\n\"\"\"\nPrefect Executors implement the logic for how Tasks are run. The standard interface\nfor an Executor consists of the following methods:\n\n- `submit(fn, *args, **kwargs)`: submit `fn(*args, **kwargs)` for execution;\n    note that this function is (in general) non-blocking, meaning that `executor.submit(...)`\n    will _immediately_ return a future-like object regardless of whether `fn(*args, **kwargs)`\n    has completed running\n- `submit_with_context(fn, *args, context, **kwargs)`: submit `fn(*args,\n    **kwargs)` for execution with the provided `prefect.context`\n- `wait(object)`: resolves any objects returned by `executor.submit` to\n    their values; this function _will_ block until execution of `object` is complete\n- `map(fn, *args, upstream_states, **kwargs)`: submit function to be mapped\n    over based on the edge information contained in `upstream_states`.  Any \"mapped\" Edge\n    will be converted into multiple function submissions, one for each value of the upstream mapped tasks.\n\nCurrently, the available executor options are:\n\n- `LocalExecutor`: the no frills, straightforward executor - great for simple\n    debugging; tasks are executed immediately upon being called by `executor.submit()`.\n    Note that the `map` feature is currently _not_ supported with this executor.\n- `SynchronousExecutor`: an executor that runs on `dask` primitives with the\n    synchronous dask scheduler; currently the default executor\n- `DaskExecutor`: the most feature-rich of the executors, this executor runs\n    on `dask.distributed` and has support for multiprocessing, multithreading, and distributed execution.\n\nWhich executor you choose depends on whether you intend to use things like parallelism\nof task execution.\n\"\"\"\nimport sys\n\nfrom warnings import warn as _warn\nfrom importlib import import_module as _import_module\n\nimport prefect as _prefect\nfrom prefect.engine.executors.base import Executor\nfrom prefect.engine.executors.local import LocalExecutor\nfrom prefect.engine.executors.sync import SynchronousExecutor\n\nif sys.version_info >= (3, 5):\n    from prefect.engine.executors.dask import DaskExecutor\n\ntry:\n    cfg_exec = _prefect.config.engine.executor\n    *module, cls_name = cfg_exec.split(\".\")\n    module = _import_module(\".\".join(module))\n    DEFAULT_EXECUTOR = getattr(module, cls_name)()\nexcept:\n    _warn(\n        \"Could not import {}, using prefect.engine.executors.LocalExecutor instead.\".format(\n            _prefect.config.engine.executor\n        )\n    )\n    DEFAULT_EXECUTOR = LocalExecutor()\n", "path": "src/prefect/engine/executors/__init__.py"}]}