Repository: PrefectHQ/prefect

Issue:
Cycle flow error with apply_map
Hi:

## Description
Using `apply_map`with multiple variables returns cyclical graph.  The `apply_map` function adds a "**random**" dependency from a downstream task to an upstream task.

For more context see [slack thread](https://prefect-community.slack.com/archives/CL09KU1K7/p1607732581070700)

## Reproduction
```
from prefect import Flow, task, case, apply_map, Parameter, unmapped
from prefect.tasks.control_flow import merge

@task
def generate_input1():
    return{
        "num":1,
        "text": "hello"
    }
@task
def generate_input2():
    return {
        "num":2,
        "text": "bye"
    }


@task
def inc(input1, squared):
    return (input1["num"] + 1)**squared

@task
def negate(input1, input2):
    return input1.extend(input2)

@task
def is_even(input1):
    return input1["num"] % 2 == 0

def inc_or_negate(input1, input2, squared):
    
    cond = is_even(input1)

    # If x is even, increment it
    with case(cond, True):
        res1 = inc(input1, unmapped(squared))

    # If x is odd, negate it
    with case(cond, False):
        res2 = negate(input1, input2)
    
    return merge(res1, res2)

with Flow("apply-map example") as flow:
    
    squared = Parameter("squared", default=2)

    input1 = generate_input1.map()
    input2 = generate_input2.map()
    
    result = apply_map(inc_or_negate, input1, input2, squared)


flow.visualize()

state = flow.run()

flow.visualize(flow_state=state)
```

![image](https://user-images.githubusercontent.com/16763201/103023264-4046a980-4513-11eb-856b-3299e8a4093d.png)



## Environment
version `0.14.00`

Optionally run `prefect diagnostics` from the command line and paste the information here. -->


Discussion:

Patch:
diff --git a/changes/pr3920.yaml b/changes/pr3920.yaml
new file mode 100644
index 0000000000..566d89eccf
--- /dev/null
+++ b/changes/pr3920.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Fix edge case in `apply_map` that resulted in cycles in the `Flow` graph - [#3920](https://github.com/PrefectHQ/prefect/pull/3920)"
diff --git a/src/prefect/utilities/tasks.py b/src/prefect/utilities/tasks.py
index c74609c729..2915aaf251 100644
--- a/src/prefect/utilities/tasks.py
+++ b/src/prefect/utilities/tasks.py
@@ -106,6 +106,7 @@ def apply_map(func: Callable, *args: Any, flow: "Flow" = None, **kwargs: Any) ->
         arg_info[a2] = (is_mapped, is_constant)
         if not is_constant:
             flow.add_task(a2)  # type: ignore
+            flow2.add_task(a2)  # type: ignore
         if is_mapped and is_constant:
             id_to_const[id(a2.value)] = a2  # type: ignore
         return a2


Test patch:
diff --git a/tests/utilities/test_tasks.py b/tests/utilities/test_tasks.py
index 3866654e48..62ebe61107 100644
--- a/tests/utilities/test_tasks.py
+++ b/tests/utilities/test_tasks.py
@@ -321,6 +321,32 @@ class TestApplyMap:
         assert state.result[b].is_skipped()
         assert state.result[c].is_skipped()
 
+    def test_apply_map_inputs_added_to_subflow_before_calling_func(self):
+        """We need to ensure all args to `appy_map` are added to the temporary
+        subflow *before* calling the mapped func. Otherwise things like
+        `case`/`resource_manager` statements that check the subgraph can get
+        confused and create new edges that shouldn't exist, leading to cycles."""
+
+        def func(cond, a, b):
+            with case(cond, True):
+                res1 = a + 1
+            with case(cond, False):
+                res2 = b + 1
+            return merge(res1, res2)
+
+        @tasks.task
+        def identity(x):
+            return x
+
+        with Flow("test") as flow:
+            cond = identity([True, False, True])
+            a = identity([1, 2, 3])
+            b = identity([4, 5, 6])
+            c = apply_map(func, cond, a, b)
+
+        state = flow.run()
+        assert state.result[c].result == [2, 6, 4]
+
 
 class TestAsTask:
     @pytest.mark.parametrize(