create time filtering for bulk actions table

python_modules/dagster-graphql/dagster_graphql/implementation/fetch_runs.py

@@ -1,3 +1,4 @@
+import datetime
 from collections import defaultdict
 from typing import (
     TYPE_CHECKING,
@@ -18,13 +19,13 @@
 )
 from dagster._core.definitions.selector import JobSubsetSelector
 from dagster._core.errors import DagsterInvariantViolationError, DagsterRunNotFoundError
-from dagster._core.execution.backfill import PartitionBackfill
+from dagster._core.execution.backfill import BulkActionsFilter
 from dagster._core.instance import DagsterInstance
 from dagster._core.storage.dagster_run import DagsterRunStatus, RunRecord, RunsFilter
 from dagster._core.storage.event_log.base import AssetRecord
 from dagster._core.storage.tags import BACKFILL_ID_TAG, TagType, get_tag_type
 from dagster._record import record
-from dagster._time import datetime_from_timestamp, get_current_timestamp
+from dagster._time import datetime_from_timestamp
 
 from .external import ensure_valid_config, get_external_job_or_raise
 
@@ -436,14 +437,10 @@ def _fetch_runs_not_in_backfill(
     instance: DagsterInstance,
     cursor: Optional[str],
     limit: int,
-    created_before: Optional[float],
+    created_before: Optional[datetime.datetime],
 ) -> Sequence[RunRecord]:
     """Fetches limit RunRecords that are not part of a backfill and were created before a given timestamp."""
-    runs_filter = (
-        RunsFilter(created_before=datetime_from_timestamp(created_before))
-        if created_before
-        else None
-    )
+    runs_filter = RunsFilter(created_before=created_before) if created_before else None
 
     runs = []
     while len(runs) < limit:
@@ -458,38 +455,6 @@ def _fetch_runs_not_in_backfill(
     return runs[:limit]
 
 
-def _fetch_backfills_created_before_timestamp(
-    instance: DagsterInstance,
-    cursor: Optional[str],
-    limit: int,
-    created_before: Optional[float] = None,
-) -> Sequence[PartitionBackfill]:
-    """Fetches limit PartitionBackfills that were created before a given timestamp.
-
-    Note: This is a reasonable filter to add to the get_backfills instance method. However, we should have a
-    more generalized way of adding filters than adding new parameters to get_backfills. So for now, doing this
-    in a separate function.
-    """
-    created_before = created_before if created_before else get_current_timestamp()
-    backfills = []
-    while len(backfills) < limit:
-        # fetch backfills in a loop discarding backfills that were created after created_before until
-        # we have limit backfills to return or have reached the end of the backfills table
-        new_backfills = instance.get_backfills(cursor=cursor, limit=limit)
-        if len(new_backfills) == 0:
-            return backfills
-        cursor = new_backfills[-1].backfill_id
-        backfills.extend(
-            [
-                backfill
-                for backfill in new_backfills
-                if backfill.backfill_timestamp <= created_before
-            ]
-        )
-
-    return backfills[:limit]
-
-
 def get_runs_feed_entries(
     graphene_info: "ResolveInfo",
     limit: int,
@@ -518,13 +483,15 @@ def get_runs_feed_entries(
     fetch_limit = limit + 1
     # filter out any backfills/runs that are newer than the cursor timestamp. See RunsFeedCursor docstring
     # for case when theis is necessary
+    created_before_cursor = (
+        datetime_from_timestamp(runs_feed_cursor.timestamp) if runs_feed_cursor.timestamp else None
+    )
     backfills = [
         GraphenePartitionBackfill(backfill)
-        for backfill in _fetch_backfills_created_before_timestamp(
-            instance,
+        for backfill in instance.get_backfills(
             cursor=runs_feed_cursor.backfill_cursor,
-            limit=fetch_limit,
-            created_before=runs_feed_cursor.timestamp,
+            limit=limit,
+            filters=BulkActionsFilter(created_before=created_before_cursor),
         )
     ]
     runs = [
@@ -533,7 +500,7 @@ def get_runs_feed_entries(
             instance,
             cursor=runs_feed_cursor.run_cursor,
             limit=fetch_limit,
-            created_before=runs_feed_cursor.timestamp,
+            created_before=created_before_cursor,
         )
     ]
 

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_runs_feed.py

@@ -51,7 +51,7 @@
 # CURRENT_TIMESTAMP only has second precision for sqlite, so if we create runs and backfills without any delay
 # the resulting list is a chunk of runs and then a chunk of backfills when ordered by time. Adding a small
 # delay between creating a run and a backfill makes the resulting list more interwoven
-CREATE_DELAY = 0.5
+CREATE_DELAY = 1
 
 
 def _create_run(graphql_context) -> DagsterRun:
@@ -99,6 +99,20 @@ def gql_context_with_runs_and_backfills(self, class_scoped_graphql_context):
         return class_scoped_graphql_context
 
     def test_get_runs_feed(self, gql_context_with_runs_and_backfills):
+        result = execute_dagster_graphql(
+            gql_context_with_runs_and_backfills.create_request_context(),
+            GET_RUNS_FEED_QUERY,
+            variables={
+                "limit": 25,
+                "cursor": None,
+            },
+        )
+        prev_run_time = None
+        for res in result.data["runsFeedOrError"]["results"]:
+            if prev_run_time:
+                assert res["creationTime"] <= prev_run_time
+            prev_run_time = res["creationTime"]
+
         result = execute_dagster_graphql(
             gql_context_with_runs_and_backfills.create_request_context(),
             GET_RUNS_FEED_QUERY,

python_modules/dagster/dagster/_core/execution/backfill.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from enum import Enum
 from typing import Mapping, NamedTuple, Optional, Sequence, Union
 
@@ -14,6 +15,7 @@
 from dagster._core.remote_representation.origin import RemotePartitionSetOrigin
 from dagster._core.storage.tags import USER_TAG
 from dagster._core.workspace.workspace import IWorkspace
+from dagster._record import record
 from dagster._serdes import whitelist_for_serdes
 from dagster._utils.error import SerializableErrorInfo
 
@@ -38,6 +40,28 @@ def from_graphql_input(graphql_str):
         return BulkActionStatus(graphql_str)
 
 
+@record
+class BulkActionsFilter:
+    """Filters to use when querying for bulk actions (i.e. backfills) from the BulkActionsTable.
+
+    Each field of the BulkActionsFilter represents a logical AND with each other. For
+    example, if you specify status and created_before, then you will receive only bulk actions
+    with the specified states AND the created before created_before. If left blank, then
+    all values will be permitted for that field.
+
+    Args:
+        status (Optional[BulkActionStatus]): A status to filter by.
+        created_before (Optional[DateTime]): Filter by bulk actions that were created before this datetime. Note that the
+            create_time for each bulk action is stored in UTC.
+        created_after (Optional[DateTime]): Filter by bulk actions that were created after this datetime. Note that the
+            create_time for each bulk action is stored in UTC.
+    """
+
+    status: Optional[BulkActionStatus] = None
+    created_before: Optional[datetime] = None
+    created_after: Optional[datetime] = None
+
+
 @whitelist_for_serdes
 class PartitionBackfill(
     NamedTuple(

python_modules/dagster/dagster/_core/instance/__init__.py

@@ -123,7 +123,11 @@
         JobFailureData,
     )
     from dagster._core.events.log import EventLogEntry
-    from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+    from dagster._core.execution.backfill import (
+        BulkActionsFilter,
+        BulkActionStatus,
+        PartitionBackfill,
+    )
     from dagster._core.execution.plan.plan import ExecutionPlan
     from dagster._core.execution.plan.resume_retry import ReexecutionStrategy
     from dagster._core.execution.stats import RunStepKeyStatsSnapshot
@@ -3076,8 +3080,11 @@ def get_backfills(
         status: Optional["BulkActionStatus"] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
+        filters: Optional["BulkActionsFilter"] = None,
     ) -> Sequence["PartitionBackfill"]:
-        return self._run_storage.get_backfills(status=status, cursor=cursor, limit=limit)
+        return self._run_storage.get_backfills(
+            status=status, cursor=cursor, limit=limit, filters=filters
+        )
 
     def get_backfill(self, backfill_id: str) -> Optional["PartitionBackfill"]:
         return self._run_storage.get_backfill(backfill_id)

python_modules/dagster/dagster/_core/storage/legacy_storage.py

@@ -33,7 +33,11 @@
     from dagster._core.event_api import AssetRecordsFilter, RunStatusChangeRecordsFilter
     from dagster._core.events import DagsterEvent, DagsterEventType
     from dagster._core.events.log import EventLogEntry
-    from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+    from dagster._core.execution.backfill import (
+        BulkActionsFilter,
+        BulkActionStatus,
+        PartitionBackfill,
+    )
     from dagster._core.execution.stats import RunStepKeyStatsSnapshot
     from dagster._core.instance import DagsterInstance
     from dagster._core.remote_representation.origin import RemoteJobOrigin
@@ -312,8 +316,9 @@ def get_backfills(
         status: Optional["BulkActionStatus"] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
+        filters: Optional["BulkActionsFilter"] = None,
     ) -> Sequence["PartitionBackfill"]:
-        return self._storage.run_storage.get_backfills(status, cursor, limit)
+        return self._storage.run_storage.get_backfills(status, cursor, limit, filters=filters)
 
     def get_backfill(self, backfill_id: str) -> Optional["PartitionBackfill"]:
         return self._storage.run_storage.get_backfill(backfill_id)

python_modules/dagster/dagster/_core/storage/runs/base.py

@@ -4,7 +4,7 @@
 from typing_extensions import TypedDict
 
 from dagster._core.events import DagsterEvent
-from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+from dagster._core.execution.backfill import BulkActionsFilter, BulkActionStatus, PartitionBackfill
 from dagster._core.execution.telemetry import RunTelemetryData
 from dagster._core.instance import MayHaveInstanceWeakref, T_DagsterInstance
 from dagster._core.snap import ExecutionPlanSnapshot, JobSnapshot
@@ -373,6 +373,7 @@ def get_backfills(
         status: Optional[BulkActionStatus] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
+        filters: Optional[BulkActionsFilter] = None,
     ) -> Sequence[PartitionBackfill]:
         """Get a list of partition backfills."""
 

python_modules/dagster/dagster/_core/storage/runs/sql_run_storage.py

@@ -38,7 +38,7 @@
     DagsterEventType,
     RunFailureReason,
 )
-from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+from dagster._core.execution.backfill import BulkActionsFilter, BulkActionStatus, PartitionBackfill
 from dagster._core.remote_representation.origin import RemoteJobOrigin
 from dagster._core.snap import (
     ExecutionPlanSnapshot,
@@ -837,16 +837,27 @@ def get_backfills(
         status: Optional[BulkActionStatus] = None,
         cursor: Optional[str] = None,
         limit: Optional[int] = None,
+        filters: Optional[BulkActionsFilter] = None,
     ) -> Sequence[PartitionBackfill]:
         check.opt_inst_param(status, "status", BulkActionStatus)
-        query = db_select([BulkActionsTable.c.body])
-        if status:
+        query = db_select([BulkActionsTable.c.body, BulkActionsTable.c.timestamp])
+        if status or (filters and filters.status):
+            if status and filters and filters.status and status != filters.status:
+                raise DagsterInvariantViolationError(
+                    "Conflicting status filters provided to get_backfills. Choose one of status or BulkActionsFilter.status."
+                )
+            status = status or (filters.status if filters else None)
+            assert status
             query = query.where(BulkActionsTable.c.status == status.value)
         if cursor:
             cursor_query = db_select([BulkActionsTable.c.id]).where(
                 BulkActionsTable.c.key == cursor
             )
             query = query.where(BulkActionsTable.c.id < cursor_query)
+        if filters and filters.created_after:
+            query = query.where(BulkActionsTable.c.timestamp > filters.created_after)
+        if filters and filters.created_before:
+            query = query.where(BulkActionsTable.c.timestamp < filters.created_before)
         if limit:
             query = query.limit(limit)
         query = query.order_by(BulkActionsTable.c.id.desc())

python_modules/dagster/dagster_tests/storage_tests/utils/run_storage.py

@@ -14,7 +14,7 @@
     DagsterSnapshotDoesNotExist,
 )
 from dagster._core.events import DagsterEvent, DagsterEventType, JobFailureData, RunFailureReason
-from dagster._core.execution.backfill import BulkActionStatus, PartitionBackfill
+from dagster._core.execution.backfill import BulkActionsFilter, BulkActionStatus, PartitionBackfill
 from dagster._core.instance import DagsterInstance, InstanceType
 from dagster._core.launcher.sync_in_memory_run_launcher import SyncInMemoryRunLauncher
 from dagster._core.remote_representation import (
@@ -44,7 +44,7 @@
 from dagster._daemon.daemon import SensorDaemon
 from dagster._daemon.types import DaemonHeartbeat
 from dagster._serdes import serialize_pp
-from dagster._time import create_datetime
+from dagster._time import create_datetime, datetime_from_timestamp
 
 win_py36 = _seven.IS_WINDOWS and sys.version_info[0] == 3 and sys.version_info[1] == 6
 
@@ -1337,6 +1337,81 @@ def test_backfill(self, storage: RunStorage):
         assert len(storage.get_backfills()) == 1
         assert len(storage.get_backfills(status=BulkActionStatus.REQUESTED)) == 0
 
+    def test_backfill_status_filtering(self, storage: RunStorage):
+        origin = self.fake_partition_set_origin("fake_partition_set")
+        backfills = storage.get_backfills()
+        assert len(backfills) == 0
+
+        one = PartitionBackfill(
+            "one",
+            partition_set_origin=origin,
+            status=BulkActionStatus.REQUESTED,
+            partition_names=["a", "b", "c"],
+            from_failure=False,
+            tags={},
+            backfill_timestamp=time.time(),
+        )
+        storage.add_backfill(one)
+        assert (
+            len(storage.get_backfills(filters=BulkActionsFilter(status=BulkActionStatus.REQUESTED)))
+            == 1
+        )
+        assert (
+            len(storage.get_backfills(filters=BulkActionsFilter(status=BulkActionStatus.COMPLETED)))
+            == 0
+        )
+        backfills = storage.get_backfills(
+            filters=BulkActionsFilter(status=BulkActionStatus.REQUESTED)
+        )
+        assert backfills[0] == one
+
+        storage.update_backfill(one.with_status(status=BulkActionStatus.COMPLETED))
+        assert (
+            len(storage.get_backfills(filters=BulkActionsFilter(status=BulkActionStatus.REQUESTED)))
+            == 0
+        )
+        assert (
+            len(storage.get_backfills(filters=BulkActionsFilter(status=BulkActionStatus.COMPLETED)))
+            == 1
+        )
+
+    def test_backfill_created_time_filtering(self, storage: RunStorage):
+        origin = self.fake_partition_set_origin("fake_partition_set")
+        backfills = storage.get_backfills()
+        assert len(backfills) == 0
+
+        all_backfills = []
+        for i in range(5):
+            backfill = PartitionBackfill(
+                f"backfill_{i}",
+                partition_set_origin=origin,
+                status=BulkActionStatus.REQUESTED,
+                partition_names=["a", "b", "c"],
+                from_failure=False,
+                tags={},
+                backfill_timestamp=time.time(),
+            )
+            storage.add_backfill(backfill)
+            all_backfills.append(backfill)
+
+        created_before = storage.get_backfills(
+            filters=BulkActionsFilter(
+                created_before=datetime_from_timestamp(all_backfills[2].backfill_timestamp)
+            )
+        )
+        assert len(created_before) == 2
+        for backfill in created_before:
+            assert backfill.backfill_timestamp < all_backfills[2].backfill_timestamp
+
+        created_after = storage.get_backfills(
+            filters=BulkActionsFilter(
+                created_after=datetime_from_timestamp(all_backfills[2].backfill_timestamp)
+            )
+        )
+        assert len(created_after) == 2
+        for backfill in created_after:
+            assert backfill.backfill_timestamp > all_backfills[2].backfill_timestamp
+
     def test_secondary_index(self, storage):
         self._skip_in_memory(storage)