DagsterInstance.report_dagster_event timestamp override

Author: dpeng817

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

@@ -2751,6 +2751,7 @@ def report_dagster_event(
         run_id: str,
         log_level: Union[str, int] = logging.INFO,
         batch_metadata: Optional["DagsterEventBatchMetadata"] = None,
+        timestamp: Optional[float] = None,
     ) -> None:
         """Takes a DagsterEvent and stores it in persistent storage for the corresponding DagsterRun."""
         from dagster._core.events.log import EventLogEntry
@@ -2761,7 +2762,7 @@ def report_dagster_event(
             job_name=dagster_event.job_name,
             run_id=run_id,
             error_info=None,
-            timestamp=get_current_timestamp(),
+            timestamp=timestamp or get_current_timestamp(),
             step_key=dagster_event.step_key,
             dagster_event=dagster_event,
         )

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

@@ -121,7 +121,7 @@
     ASSET_PARTITION_RANGE_START_TAG,
     MULTIDIMENSIONAL_PARTITION_PREFIX,
 )
-from dagster._core.test_utils import create_run_for_test, instance_for_test
+from dagster._core.test_utils import create_run_for_test, freeze_time, instance_for_test
 from dagster._core.types.loadable_target_origin import LoadableTargetOrigin
 from dagster._core.utils import make_new_run_id
 from dagster._loggers import colored_console_logger
@@ -3098,6 +3098,50 @@ def test_write_asset_materialization_failures(self, storage, instance, test_run_
 
             assert failed_partitions_by_step_key == failed_partitions
 
+    def test_timestamp_overrides(self, storage, instance: DagsterInstance) -> None:
+        # instance.report_dagster_event(
+        #     DagsterEvent(
+        #         event_type_value="PIPELINE_START",
+        #         job_name=run.job_name
+        #     ),
+        #     timestamp=datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
+        # )
+        frozen_time = get_current_datetime()
+        with freeze_time(frozen_time):
+            instance.report_dagster_event(
+                run_id="",
+                dagster_event=DagsterEvent(
+                    event_type_value=DagsterEventType.ASSET_MATERIALIZATION.value,
+                    job_name="",
+                    event_specific_data=StepMaterializationData(
+                        AssetMaterialization(asset_key="foo")
+                    ),
+                ),
+            )
+
+            record = instance.get_asset_records([AssetKey("foo")])[0]
+            assert (
+                record.asset_entry.last_materialization_record.timestamp == frozen_time.timestamp()
+            )
+
+            report_date = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
+
+            instance.report_dagster_event(
+                run_id="",
+                dagster_event=DagsterEvent(
+                    event_type_value=DagsterEventType.ASSET_MATERIALIZATION.value,
+                    job_name="",
+                    event_specific_data=StepMaterializationData(
+                        AssetMaterialization(asset_key="foo")
+                    ),
+                ),
+                timestamp=report_date.timestamp(),
+            )
+            record = instance.get_asset_records([AssetKey("foo")])[0]
+            assert (
+                record.asset_entry.last_materialization_record.timestamp == report_date.timestamp()
+            )
+
     def test_get_latest_storage_ids_by_partition(self, storage, instance):
         a = AssetKey(["a"])
         b = AssetKey(["b"])