[DOP-31810] Rename run.expected_%_time fields to expected_%_at

Author: dolfinus

data_rentgen/consumer/extractors/generic/run.py

@@ -140,9 +140,9 @@ def _enrich_nominal_times(self, run: RunDTO, event: OpenLineageRunEvent) -> RunD
         if not event.run.facets.nominalTime:
             return run
 
-        run.nominal_start_time = event.run.facets.nominalTime.nominalStartTime
-        run.nominal_end_time = event.run.facets.nominalTime.nominalEndTime
-        if run.nominal_start_time == run.nominal_end_time:
-            run.nominal_end_time = None
+        run.expected_start_at = event.run.facets.nominalTime.nominalStartTime
+        run.expected_end_at = event.run.facets.nominalTime.nominalEndTime
+        if run.expected_start_at == run.expected_end_at:
+            run.expected_end_at = None
 
         return run

data_rentgen/db/migrations/versions/2026-02-03_0e9bb788b04b_rename_run_expected___time_columns.py

@@ -0,0 +1,28 @@
+# SPDX-FileCopyrightText: 2024-present MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+"""Rename run.expected_%_time columns
+
+Revision ID: 0e9bb788b04b
+Revises: b6a12e72a04e
+Create Date: 2026-02-03 11:10:05.902453
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0e9bb788b04b"
+down_revision = "b6a12e72a04e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(sa.text("ALTER TABLE run RENAME COLUMN expected_start_time TO expected_start_at"))
+    op.execute(sa.text("ALTER TABLE run RENAME COLUMN expected_end_time TO expected_end_at"))
+
+
+def downgrade() -> None:
+    op.execute(sa.text("ALTER TABLE run RENAME COLUMN expected_start_at TO expected_start_time"))
+    op.execute(sa.text("ALTER TABLE run RENAME COLUMN expected_end_at TO expected_end_time"))

data_rentgen/db/models/run.py

@@ -152,12 +152,12 @@ class Run(Base):
         nullable=True,
         doc="End reason of the run, e.g. exception string",
     )
-    expected_start_time: Mapped[datetime] = mapped_column(
+    expected_start_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         nullable=True,
         doc="Timestamp representing the nominal start time (included) of the run. AKA the schedule time",
     )
-    expected_end_time: Mapped[datetime] = mapped_column(
+    expected_end_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         nullable=True,
         doc=(

data_rentgen/db/repositories/run.py

@@ -61,8 +61,8 @@
     started_by_user_id=bindparam("started_by_user_id"),
     start_reason=bindparam("start_reason"),
     ended_at=bindparam("ended_at"),
-    expected_start_time=bindparam("expected_start_time"),
-    expected_end_time=bindparam("expected_end_time"),
+    expected_start_at=bindparam("expected_start_at"),
+    expected_end_at=bindparam("expected_end_at"),
 )
 inserted_row = insert_statement.excluded
 insert_statement = insert_statement.on_conflict_do_update(
@@ -79,8 +79,8 @@
         "attempt": func.coalesce(inserted_row.attempt, Run.attempt),
         "persistent_log_url": func.coalesce(inserted_row.persistent_log_url, Run.persistent_log_url),
         "running_log_url": func.coalesce(inserted_row.running_log_url, Run.running_log_url),
-        "expected_start_time": func.coalesce(inserted_row.expected_start_time, Run.expected_start_time),
-        "expected_end_time": func.coalesce(inserted_row.expected_end_time, Run.expected_end_time),
+        "expected_start_at": func.coalesce(inserted_row.expected_start_at, Run.expected_start_at),
+        "expected_end_at": func.coalesce(inserted_row.expected_end_at, Run.expected_end_at),
     },
 )
 
@@ -275,8 +275,8 @@ async def create(self, run: RunDTO) -> Run:
             attempt=run.attempt,
             persistent_log_url=run.persistent_log_url,
             running_log_url=run.running_log_url,
-            expected_start_time=run.nominal_start_time,
-            expected_end_time=run.nominal_end_time,
+            expected_start_at=run.expected_start_at,
+            expected_end_at=run.expected_end_at,
         )
         self._session.add(result)
         await self._session.flush([result])
@@ -304,8 +304,8 @@ async def update(
             "attempt": new.attempt,
             "persistent_log_url": new.persistent_log_url,
             "running_log_url": new.running_log_url,
-            "expected_start_time": new.nominal_start_time,
-            "expected_end_time": new.nominal_end_time,
+            "expected_start_at": new.expected_start_at,
+            "expected_end_at": new.expected_end_at,
         }
         for col_name, value in optional_fields.items():
             if value is not None:
@@ -335,8 +335,8 @@ async def create_or_update_bulk(self, runs: list[RunDTO]) -> None:
                     "attempt": run.attempt,
                     "persistent_log_url": run.persistent_log_url,
                     "running_log_url": run.running_log_url,
-                    "expected_start_time": run.nominal_start_time,
-                    "expected_end_time": run.nominal_end_time,
+                    "expected_start_at": run.expected_start_at,
+                    "expected_end_at": run.expected_end_at,
                 }
                 for run in runs
             ],

data_rentgen/db/scripts/seed/airflow.py

@@ -59,8 +59,8 @@ def generate_airflow_run(dag_id: str, task_id: str, created_at: datetime, ended_
         persistent_log_url=dag_ui,
         status=RunStatusDTO.SUCCEEDED,
         start_reason=RunStartReasonDTO.AUTOMATIC,
-        nominal_start_time=created_at.replace(minute=(created_at.minute // 5) * 5, second=0, microsecond=0),
-        nominal_end_time=ended_at.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1),
+        expected_start_at=created_at.replace(minute=(created_at.minute // 5) * 5, second=0, microsecond=0),
+        expected_end_at=ended_at.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1),
     )
 
     task_job = JobDTO(

data_rentgen/dto/run.py

@@ -44,8 +44,8 @@ class RunDTO:
     attempt: str | None = None
     running_log_url: str | None = None
     persistent_log_url: str | None = None
-    nominal_start_time: datetime | None = None
-    nominal_end_time: datetime | None = None
+    expected_start_at: datetime | None = None
+    expected_end_at: datetime | None = None
 
     def __post_init__(self):
         self.created_at = extract_timestamp_from_uuid(self.id)
@@ -74,6 +74,6 @@ def merge(self, new: RunDTO) -> RunDTO:
         self.attempt = new.attempt or self.attempt
         self.running_log_url = new.running_log_url or self.running_log_url
         self.persistent_log_url = new.persistent_log_url or self.persistent_log_url
-        self.nominal_start_time = new.nominal_start_time or self.nominal_start_time
-        self.nominal_end_time = new.nominal_end_time or self.nominal_end_time
+        self.expected_start_at = new.expected_start_at or self.expected_start_at
+        self.expected_end_at = new.expected_end_at or self.expected_end_at
         return self

data_rentgen/server/schemas/v1/run.py

@@ -81,8 +81,8 @@ class RunResponseV1(BaseModel):
     start_reason: str | None = Field(description="Start reason of the Run", default=None)
     ended_at: datetime | None = Field(description="End time of the Run", default=None)
     end_reason: str | None = Field(description="End reason of the Run", default=None)
-    expected_start_time: datetime | None = Field(description="Run expected start time", default=None)
-    expected_end_time: datetime | None = Field(description="Run expected end time", default=None)
+    expected_start_at: datetime | None = Field(description="Run expected start time", default=None)
+    expected_end_at: datetime | None = Field(description="Run expected end time", default=None)
 
     model_config = ConfigDict(from_attributes=True)
 

docs/entities/index.rst

@@ -223,8 +223,8 @@ It contains following fields:
 - ``started_at: timestamp | None`` - timestamp when OpenLineage event with ``eventType=START`` was received.
 - ``started_by user: User | None`` - Spark session started as specific OS user/Kerberos principal.
 - ``start_reason: Enum | None`` - "why this Run was started?":
-- ``expected_start_time: timestamp | None`` - scheduled/expected start time of the run, which may differ from ``started_at`` (actual start time).
-- ``expected_end_time: timestamp | None`` - scheduled/expected end time of the run, which may differ from ``ended_at`` (actual end time).
+- ``expected_start_at: timestamp | None`` - scheduled/expected start time of the run, which may differ from ``started_at`` (actual start time).
+- ``expected_end_at: timestamp | None`` - scheduled/expected end time of the run, which may differ from ``ended_at`` (actual end time).
 
   - ``MANUAL``
   - ``AUTOMATIC`` - e.g. by schedule or triggered by another run.

docs/reference/database/structure.rst

@@ -69,8 +69,8 @@ Database structure
         start_reason: varchar(32)
         ended_at: timestamptz null
         end_reason: text null
-        expected_start_time: timestamptz null
-        expected_end_time: timestamptz null
+        expected_start_at: timestamptz null
+        expected_end_at: timestamptz null
         external_id: text null
         attempt: varchar(64) null
         persistent_log_url: timestamptz null

tests/test_consumer/test_extractors/test_extractors_run_airflow.py

@@ -1374,7 +1374,7 @@ def test_extractors_extract_run_airflow_task_tags_below_2_10(tags: list[str] | s
 
 
 @pytest.mark.parametrize(
-    ["start_time", "end_time", "expected_start_time", "expected_end_time"],
+    ["start_time", "end_time", "expected_start_at", "expected_end_at"],
     [
         (
             datetime(2024, 7, 5, 9, 4, 13, 979349, tzinfo=timezone.utc),
@@ -1399,8 +1399,8 @@ def test_extractors_extract_run_airflow_task_tags_below_2_10(tags: list[str] | s
 def test_extractors_extract_run_airflow_task_nominal_times(
     start_time: datetime | None,
     end_time: datetime | None,
-    expected_start_time: datetime | None,
-    expected_end_time: datetime | None,
+    expected_start_at: datetime | None,
+    expected_end_at: datetime | None,
 ):
     now = datetime(2024, 7, 5, 9, 4, 13, 979349, tzinfo=timezone.utc)
     run_id = UUID("01908223-0782-79b8-9495-b1c38aaee839")
@@ -1482,6 +1482,6 @@ def test_extractors_extract_run_airflow_task_nominal_times(
             "http://airflow-host:8081/dags/mydag/grid?tab=logs&dag_run_id=scheduled__2024-07-05T09%3A04%3A13%3A979349%2B00%3A00&task_id=mytask&map_index=-1"
         ),
         running_log_url=None,
-        nominal_start_time=expected_start_time,
-        nominal_end_time=expected_end_time,
+        expected_start_at=expected_start_at,
+        expected_end_at=expected_end_at,
     )