chore(rail): prod migration regen rail perf 20260101 to 20260106 (#712)

Author: huangh

src/lamp_py/migrations/README

@@ -0,0 +1,19 @@
+Generic single-database configuration.
+
+# Regeneration instructions
+
+1. run migration_template_generator.py
+2. fill it in with the migration instructions. 
+3. deploy this to performance manager and ingestion
+    - ingestion updates Metadata table
+    - performance manager updates Data Tables
+
+    It is allowable to put both metadata and prod db migrations in a single prod migration script. 
+
+    This is valid because ingestion deploys first (e.g. resetting metadata "finished processing" flags)
+    Then performance manager deploys (e.g. clearing old data)
+    Then performance manager checks if there is work (there is now, bc metadata is rest) and starts to re-process and fill in days (which are empty b/c they were cleared)
+
+4. verify that the migration has worked (either by looking at updated_date and the reset-metadata flags on the database, or running another adhoc query to check those...)
+
+5. in a new PR, bump version of the tableau upload file to regenerate the whole dataset (otherwise we only grab daily incremental changes)

src/lamp_py/migrations/README.md

@@ -80,10 +80,14 @@ def downgrade() -> None:
 if __name__ == "__main__":
     import uuid
 
-    short_desc = "reprocess_422_423"
+    short_desc = input("Enter a short description for the migration: ").strip()
     uuid_new = uuid.uuid4().hex[-12:]
 
-    versions_dir = "{LAMP}/src/lamp_py/migrations/versions"
+    import pathlib
+
+    # Get the directory containing the current script file
+    BASE_DIR = pathlib.Path(__file__).resolve().parent
+    versions_dir = BASE_DIR / "versions"
 
     # List directories in the versions directory
     if os.path.exists(versions_dir):
@@ -96,7 +100,9 @@ def downgrade() -> None:
 
     print(options)
     for o in options:
-        latest_migration = sorted([d for d in os.listdir(os.path.join(versions_dir, o)) if not d.startswith("sql")])[-1]
+        latest_migration = sorted(
+            [d for d in os.listdir(os.path.join(versions_dir, o)) if not d.startswith("sql") and d.endswith(".py")]
+        )[-1]
         parts = os.path.basename(latest_migration).split("_")
         breakpoint()
         increment_migration_count = str(int(parts[0]) + 1).zfill(3)

src/lamp_py/migrations/migration_template_generator.py

@@ -0,0 +1,86 @@
+"""regen-rail-perf-20260101-to-20260104
+
+    Revision ID: f138635d1338
+    Revises: 5e3066f113ff
+    Create Date: 2026-02-28 11:31:59.737856
+
+    Details: regen-rail-perf-20260101-to-20260104 - 
+    This will clean up missing data from RDS performance issues/outage from 1/1 to 1/4
+    Rerunning 1/1 - 1/5 just in case. 
+    * upgrade -> Delete all records from 1/1 to 1/5 in vehicle events and vehicle_trips
+              -> Set all flags to "unprocessed" in metadata log from 1/1 to 1/5
+    * downgrade -> Nothing
+"""
+
+import logging
+import os
+import tempfile
+import polars as pl
+import pyarrow as pa
+import pyarrow.parquet as pq
+from typing import List
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.exc import ProgrammingError
+
+from lamp_py.aws.s3 import download_file, upload_file
+from lamp_py.postgres.postgres_utils import DatabaseIndex, DatabaseManager
+
+# revision identifiers, used by Alembic.
+revision = "f138635d1338"
+down_revision = "5e3066f113ff"
+branch_labels = None  # tbd
+depends_on = None  # tbd
+
+
+def upgrade() -> None:
+
+    # SELECT FROM vehicle_events WHERE service_date >= 20250404 AND service_date <= 20250423;"
+
+    clear_events = "DELETE FROM vehicle_events WHERE service_date >= 20260101 AND service_date <= 20260105;"
+    op.execute(clear_events)
+
+    clear_trips = "DELETE FROM vehicle_trips WHERE service_date >= 20260101 AND service_date <= 20260105;"
+    op.execute(clear_trips)
+
+    # Query to Check
+    # SELECT created_on, rail_pm_processed, rail_pm_process_fail
+    # FROM public.metadata_log
+    # WHERE created_on > '2026-01-01' and created_on < '2026-01-05 23:59:59'
+    # AND (path LIKE '%/RT_TRIP_UPDATES/%' or path LIKE '%/RT_VEHICLE_POSITIONS/%')
+    # ORDER BY created_on;
+
+    try:
+        update_md_query = """
+        UPDATE
+            metadata_log
+        SET
+            rail_pm_process_fail = false
+            , rail_pm_processed = false
+        WHERE
+            created_on > '2026-01-01 00:00:00'
+            and created_on < '2026-01-05 23:59:59'
+            and (
+                path LIKE '%/RT_TRIP_UPDATES/%'
+                or path LIKE '%/RT_VEHICLE_POSITIONS/%'
+            )
+        ;
+        """
+        md_manager = DatabaseManager(DatabaseIndex.METADATA)
+        md_manager.execute(sa.text(update_md_query))
+
+    except ProgrammingError as error:
+        # Error 42P01 is an 'Undefined Table' error. This occurs when there is
+        # no metadata_log table in the rail performance manager database
+        #
+        # Raise all other sql errors
+        original_error = error.orig
+        if original_error is not None and hasattr(original_error, "pgcode") and original_error.pgcode == "42P01":
+            logging.info("No Metadata Table in Rail Performance Manager")
+        else:
+            raise
+
+
+def downgrade() -> None:
+    pass

…3ce_backfill_2026_01_01_to_2026_01_04.py …3ce_backfill_2026_01_01_to_2026_01_04.pysrc/lamp_py/migrations/versions/metadata_staging/5853837233ce_backfill_2026_01_01_to_2026_01_04.py renamed to src/lamp_py/migrations/versions/metadata_staging/004_5853837233ce_backfill_2026_01_01_to_2026_01_04.py src/lamp_py/migrations/versions/metadata_staging/5853837233ce_backfill_2026_01_01_to_2026_01_04.py renamed to src/lamp_py/migrations/versions/metadata_staging/004_5853837233ce_backfill_2026_01_01_to_2026_01_04.py

@@ -0,0 +1,84 @@
+"""backfill_2026_01_06
+
+    Revision ID: 7cb3dbb1dac0
+    Revises: f138635d1338
+    Create Date: 2026-03-01 09:05:04.036779
+
+    Details: Query of metadata table revealed that 1/6 also failed processing, so generating that again
+
+    * upgrade -> same as previous migration but for 1/6
+    * downgrade -> None
+    """
+
+import logging
+import os
+import tempfile
+import polars as pl
+import pyarrow as pa
+import pyarrow.parquet as pq
+from typing import List
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.exc import ProgrammingError
+
+from lamp_py.aws.s3 import download_file, upload_file
+from lamp_py.postgres.postgres_utils import DatabaseIndex, DatabaseManager
+
+# revision identifiers, used by Alembic.
+revision = "7cb3dbb1dac0"
+down_revision = "f138635d1338"
+branch_labels = None  # tbd
+depends_on = None  # tbd
+
+
+def upgrade() -> None:
+
+    # SELECT FROM vehicle_events WHERE service_date >= 20250404 AND service_date <= 20250423;"
+
+    clear_events = "DELETE FROM vehicle_events WHERE service_date = 20260106;"
+    op.execute(clear_events)
+
+    clear_trips = "DELETE FROM vehicle_trips WHERE service_date = 20260106;"
+    op.execute(clear_trips)
+
+    # Query to Check
+    # SELECT created_on, rail_pm_processed, rail_pm_process_fail
+    # FROM public.metadata_log
+    # WHERE created_on > '2026-01-06' and created_on < '2026-01-06 23:59:59'
+    # AND (path LIKE '%/RT_TRIP_UPDATES/%' or path LIKE '%/RT_VEHICLE_POSITIONS/%')
+    # ORDER BY created_on;
+
+    try:
+        update_md_query = """
+        UPDATE
+            metadata_log
+        SET
+            rail_pm_process_fail = false
+            , rail_pm_processed = false
+        WHERE
+            created_on > '2026-01-06 00:00:00'
+            and created_on < '2026-01-06 23:59:59'
+            and (
+                path LIKE '%/RT_TRIP_UPDATES/%'
+                or path LIKE '%/RT_VEHICLE_POSITIONS/%'
+            )
+        ;
+        """
+        md_manager = DatabaseManager(DatabaseIndex.METADATA)
+        md_manager.execute(sa.text(update_md_query))
+
+    except ProgrammingError as error:
+        # Error 42P01 is an 'Undefined Table' error. This occurs when there is
+        # no metadata_log table in the rail performance manager database
+        #
+        # Raise all other sql errors
+        original_error = error.orig
+        if original_error is not None and hasattr(original_error, "pgcode") and original_error.pgcode == "42P01":
+            logging.info("No Metadata Table in Rail Performance Manager")
+        else:
+            raise
+
+
+def downgrade() -> None:
+    pass