Feat: save db space by storing horizon as ints

documentation/changelog.rst

@@ -24,6 +24,7 @@ New features
 
 Infrastructure / Support
 ----------------------
+* Store timed belief horizons as integer seconds in the database to reduce belief storage size; upgrading requires running the new migration with timely-beliefs support [see `PR #XXXX <https://www.github.com/FlexMeasures/flexmeasures/pull/XXXX>`_]
 * Upgraded dependencies [see `PR #1485 <https://www.github.com/FlexMeasures/flexmeasures/pull/1485>`_, `PR #2215 <https://www.github.com/FlexMeasures/flexmeasures/pull/2215>`_ and `PR #2243 <https://www.github.com/FlexMeasures/flexmeasures/pull/2243>`_]
 * Prepare the ``device_scheduler`` to deal with commitments per device group [see `PR #1934 <https://www.github.com/FlexMeasures/flexmeasures/pull/1934>`_]
 * Support storing encrypted connection secrets on organisations and assets, including utility functions, encryption key configuration, CLI commands to set and delete secrets, and UI tables that show stored secret names and optional expiration times without exposing their values [see `PR #2236 <https://www.github.com/FlexMeasures/flexmeasures/pull/2236>`_]

flexmeasures/api/v3_0/tests/test_sensors_api.py

@@ -744,6 +744,7 @@ def test_fetch_sensor_stats(
         for source, record in response_content.items():
             assert record["First event start"]
             assert record["Last event end"]
+            assert record["Last recorded"]
             assert record["Min value"]
             assert record["Min value"]
             assert record["Max value"]

flexmeasures/data/migrations/alembic.ini

@@ -11,7 +11,7 @@
 
 # Logging configuration
 [loggers]
-keys = root,sqlalchemy,alembic
+keys = root,sqlalchemy,alembic,flask_migrate
 
 [handlers]
 keys = console
@@ -34,6 +34,11 @@ level = INFO
 handlers =
 qualname = alembic
 
+[logger_flask_migrate]
+level = INFO
+handlers =
+qualname = flask_migrate
+
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)

flexmeasures/data/migrations/versions/6d5c9e9d62cf_change_timed_belief_horizon_to_integer_seconds.py

@@ -0,0 +1,106 @@
+"""change timed belief horizon to integer seconds
+
+Also delete duplicate beliefs that would have become duplicate primary keys after the change
+(we keep the most recent beliefs). Those are beliefs with sub-second differences in belief horizons.
+That last part is not reversible.
+
+Revision ID: 6d5c9e9d62cf
+Revises: 55d8936a55f9
+Create Date: 2026-07-10 02:05:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision = "6d5c9e9d62cf"
+down_revision = "55d8936a55f9"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    out_of_range_beliefs = (
+        op.get_bind()
+        .execute(
+            sa.text(
+                """
+            SELECT COUNT(*)
+            FROM timed_belief
+            WHERE
+                EXTRACT(EPOCH FROM belief_horizon) < -2147483648
+                OR EXTRACT(EPOCH FROM belief_horizon) > 2147483647
+            """
+            )
+        )
+        .scalar_one()
+    )
+    if out_of_range_beliefs:
+        print(
+            f"Found {out_of_range_beliefs} timed_belief rows with belief_horizon "
+            "outside the supported integer-second range of about 68 years."
+        )
+        raise RuntimeError(
+            "Cannot convert timed_belief.belief_horizon to integer seconds: "
+            f"{out_of_range_beliefs} rows exceed the signed INTEGER range."
+        )
+
+    deleted_duplicate_beliefs = (
+        op.get_bind()
+        .execute(
+            sa.text(
+                """
+            WITH ranked_beliefs AS (
+                SELECT
+                    ctid,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY
+                            source_id,
+                            event_start,
+                            FLOOR(EXTRACT(EPOCH FROM belief_horizon))::INTEGER,
+                            cumulative_probability,
+                            sensor_id
+                        ORDER BY belief_horizon ASC
+                    ) AS rank_in_future_primary_key
+                FROM timed_belief
+            ),
+            deleted_rows AS (
+                DELETE FROM timed_belief
+                WHERE ctid IN (
+                    SELECT ctid
+                    FROM ranked_beliefs
+                    WHERE rank_in_future_primary_key > 1
+                )
+                RETURNING 1
+            )
+            SELECT COUNT(*) FROM deleted_rows
+            """
+            )
+        )
+        .scalar_one()
+    )
+    print(
+        "Deleted "
+        f"{deleted_duplicate_beliefs} timed_belief rows that would have become "
+        "duplicate primary keys after converting belief_horizon to integer seconds."
+    )
+    op.alter_column(
+        "timed_belief",
+        "belief_horizon",
+        existing_type=postgresql.INTERVAL(),
+        type_=sa.Integer(),
+        postgresql_using="FLOOR(EXTRACT(EPOCH FROM belief_horizon))::INTEGER",
+    )
+
+
+def downgrade():
+    op.alter_column(
+        "timed_belief",
+        "belief_horizon",
+        existing_type=sa.Integer(),
+        type_=postgresql.INTERVAL(),
+        postgresql_using="belief_horizon * interval '1 second'",
+    )

flexmeasures/data/services/sensors.py

@@ -830,7 +830,7 @@ def filtered_agg(func):
             sa.func.max(
                 TimedBelief.event_start
                 + sensor.event_resolution
-                - TimedBelief.belief_horizon
+                - TimedBelief.belief_horizon * sa.literal_column("interval '1 second'")
             ).label("max_belief_time"),
             filtered_agg(sa.func.min).label("min_event_value"),
             filtered_agg(sa.func.max).label("max_event_value"),

flexmeasures/data/tests/test_scheduling_sequential.py

@@ -88,7 +88,6 @@ def test_create_sequential_jobs(db, app, flex_description_sequential, smart_buil
     assert battery_power.empty
 
     # Work on jobs
-    queued_jobs[0].perform()
     work_on_rq(queue, handle_scheduling_exception)
 
     # Check that the jobs completed successfully