diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 4fc2f13886..8eb9fbe187 100644 --- a/documentation/changelog.rst +++ b/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 `_] * Upgraded dependencies [see `PR #1485 `_, `PR #2215 `_ and `PR #2243 `_] * Prepare the ``device_scheduler`` to deal with commitments per device group [see `PR #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 `_] diff --git a/flexmeasures/api/v3_0/tests/test_sensors_api.py b/flexmeasures/api/v3_0/tests/test_sensors_api.py index d8d033ca16..dac4356c00 100644 --- a/flexmeasures/api/v3_0/tests/test_sensors_api.py +++ b/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"] diff --git a/flexmeasures/data/migrations/alembic.ini b/flexmeasures/data/migrations/alembic.ini index f8ed4801f7..ec9d45c26a 100644 --- a/flexmeasures/data/migrations/alembic.ini +++ b/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,) diff --git a/flexmeasures/data/migrations/versions/6d5c9e9d62cf_change_timed_belief_horizon_to_integer_seconds.py b/flexmeasures/data/migrations/versions/6d5c9e9d62cf_change_timed_belief_horizon_to_integer_seconds.py new file mode 100644 index 0000000000..a447fe4ffd --- /dev/null +++ b/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'", + ) diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index 82cb7f46e5..445cae82a3 100644 --- a/flexmeasures/data/services/sensors.py +++ b/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"), diff --git a/flexmeasures/data/tests/test_scheduling_sequential.py b/flexmeasures/data/tests/test_scheduling_sequential.py index 68d5689f2d..207018ba7c 100644 --- a/flexmeasures/data/tests/test_scheduling_sequential.py +++ b/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