Repository: PrefectHQ/prefect

Issue:
Allow multiple identical clocks with different default params
## Current behavior
*Please describe how the feature works today*
Schedules don't allow for multiple identical clocks with different default parameters.


## Proposed behavior
*Please describe your proposed change to the current behavior*
Allow multiple clocks by correctly scheduling multiple runs when clocks have the same interval/cron but different default parameters. 



## Example
*Please give an example of how the enhancement would be useful*
This would be useful in a scenario where I want to schedule multiple runs at the same time but with different parameters. 

Example:
```
import prefect
from prefect import Flow, task, Parameter
from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock


@task
def print_param(x):
    return print(f"Where is my  body? {x}")


clock1 = CronClock(
    "00 12 * * *", parameter_defaults=dict(text="My body lies over the ~ocean~.")
)
clock2 = CronClock(
    "00 12 * * *", parameter_defaults=dict(text="My body lies over the ~sea~.")
)
schedule = Schedule(clocks=[clock1, clock2])
with Flow("CronClocks", schedule=schedule) as flow:
    param = Parameter("text")

    print_param(param)
```


Discussion:

Patch:
diff --git a/changes/issue2510.yaml b/changes/issue2510.yaml
new file mode 100644
index 0000000000..d219cb9c0f
--- /dev/null
+++ b/changes/issue2510.yaml
@@ -0,0 +1,2 @@
+feature:
+  - "Allow for scheduling the same flow at the same time with multiple parameter values - [#2510](https://github.com/PrefectHQ/prefect/issues/2510)"
diff --git a/src/prefect/schedules/clocks.py b/src/prefect/schedules/clocks.py
index 64a48afcc6..64366ff83d 100644
--- a/src/prefect/schedules/clocks.py
+++ b/src/prefect/schedules/clocks.py
@@ -18,7 +18,12 @@ class ClockEvent:
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, (ClockEvent, datetime)):
             return False
-        return self.start_time == other
+        if isinstance(other, datetime):
+            return self.start_time == other
+        return (
+            self.start_time == other.start_time
+            and self.parameter_defaults == other.parameter_defaults
+        )
 
     def __gt__(self, other: Union[datetime, "ClockEvent"]) -> bool:
         if not isinstance(other, (ClockEvent, datetime)):


Test patch:
diff --git a/tests/schedules/test_clocks.py b/tests/schedules/test_clocks.py
index a2fe881550..e01376b68b 100644
--- a/tests/schedules/test_clocks.py
+++ b/tests/schedules/test_clocks.py
@@ -41,19 +41,11 @@ class TestClockEvent:
         assert e.start_time == now
         assert e.parameter_defaults == dict(x=42, z=[1, 2, 3])
 
-    @pytest.mark.parametrize(
-        "dt,params",
-        list(
-            itertools.product(
-                [pendulum.now("UTC").add(hours=1), pendulum.now("UTC").add(hours=-2)],
-                [None, dict(x=1)],
-            )
-        ),
-    )
-    def test_clock_event_comparisons_are_datetime_comparisons(self, dt, params):
+    def test_clock_event_comparisons_are_datetime_comparisons(self):
         now = pendulum.now("UTC")
+        dt = now.add(hours=1)
 
-        e = clocks.ClockEvent(now, parameter_defaults=params)
+        e = clocks.ClockEvent(now)
         e2 = clocks.ClockEvent(dt)
 
         ## compare to raw datetimes
@@ -68,6 +60,26 @@ class TestClockEvent:
         assert (e < e2) == (e.start_time < e2.start_time)
         assert (e > e2) == (e.start_time > e2.start_time)
 
+    def test_clock_event_comparisons_take_parameters_into_account(self):
+        now = pendulum.now("UTC")
+        dt = now.add(hours=1)
+
+        e = clocks.ClockEvent(now, parameter_defaults=dict(a=1))
+        e2 = clocks.ClockEvent(dt)
+        e3 = clocks.ClockEvent(dt, parameter_defaults=dict(a=2))
+
+        ## compare to raw datetimes
+        assert e2 == dt
+        assert e == now
+        assert e3 == dt
+        assert e2 > now
+        assert e3 > now
+
+        ## compare to each other
+        assert e2 != e3
+        assert e2 > e
+        assert e3 > e
+
 
 class TestIntervalClock:
     def test_create_interval_clock(self):
diff --git a/tests/schedules/test_schedules.py b/tests/schedules/test_schedules.py
index 8bed00dd03..ebaf1a4e08 100644
--- a/tests/schedules/test_schedules.py
+++ b/tests/schedules/test_schedules.py
@@ -66,6 +66,31 @@ def test_create_schedule_multiple_overlapping_clocks():
     ]
 
 
+def test_create_schedule_multiple_overlapping_clocks_distinguishes_parameters():
+    dt = pendulum.datetime(2019, 1, 1)
+    special_day = pendulum.datetime(2020, 3, 1)
+    s = schedules.Schedule(
+        clocks=[
+            clocks.DatesClock([special_day]),
+            clocks.DatesClock([special_day]),
+        ]
+    )
+    output = s.next(2, after=dt, return_events=True)
+    assert len(output) == 1
+    assert output[0].start_time == special_day
+
+    s = schedules.Schedule(
+        clocks=[
+            clocks.DatesClock([special_day], parameter_defaults=dict(a=1)),
+            clocks.DatesClock([special_day], parameter_defaults=dict(a=2)),
+        ]
+    )
+    output = s.next(2, after=dt, return_events=True)
+    assert len(output) == 2
+    assert [e.start_time for e in output] == [special_day] * 2
+    assert set(e.parameter_defaults["a"] for e in output) == {1, 2}
+
+
 def test_create_schedule_multiple_overlapping_clocks_emit_events_if_asked():
     dt = pendulum.datetime(2019, 1, 1)
     s = schedules.Schedule(
@@ -100,16 +125,14 @@ def test_create_schedule_multiple_overlapping_clocks_emit_events_with_correct_pa
     output = s.next(6, after=dt, return_events=True)
 
     assert all([isinstance(e, clocks.ClockEvent) for e in output])
-    assert all(
-        [e.parameter_defaults == dict(x=(i + 1) % 2) for i, e in enumerate(output)]
-    )
-    assert output == [
+    assert [e.parameter_defaults["x"] for e in output] == [1, 0, 1, 1, 0, 1]
+    assert [e.start_time for e in output] == [
         dt.add(hours=12),
         dt.add(days=1),
+        dt.add(days=1),
         dt.add(days=1, hours=12),
         dt.add(days=2),
-        dt.add(days=2, hours=12),
-        dt.add(days=3),
+        dt.add(days=2),
     ]