We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN 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.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/engine/executors/local.py`
Content:
```
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
```

Path: `src/prefect/engine/executors/__init__.py`
Content:
```
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 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/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.

{"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"}], "after_files": [{"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\n\nfrom typing import Any, Callable, Iterable\n\nfrom prefect.engine.executors.base import Executor\nfrom prefect.utilities.executors import dict_to_list\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 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\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- `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"}]}