Speed up queries via materialized view

documentation/changelog.rst

@@ -20,6 +20,7 @@ New features
 
 Infrastructure / Support
 ----------------------
+* Speed up queries for the most recent beliefs with a materialized view, kept fresh by scheduling the new ``flexmeasures db-ops refresh-materialized-views`` command; also introduces the ``FLEXMEASURES_MVIEW_ALWAYS_INCLUDE_LIVE_TAIL`` setting and the ``include_live_tail`` chart data API parameter [see `PR #1671 <https://github.com/FlexMeasures/flexmeasures/pull/1671>`_]
 * 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>`_]

documentation/configuration.rst

@@ -809,6 +809,30 @@ Allow to override the default sunset link for your clients.
 
 Default: ``None`` (defaults are set internally for each sunset API version, e.g. ``"https://flexmeasures.readthedocs.io/en/v0.13.0/api/v2_0.html"`` for v2.0)
 
+Performance optimizations
+----------------------------
+
+.. _mview-live-tail-config:
+
+FLEXMEASURES_MVIEW_ALWAYS_INCLUDE_LIVE_TAIL
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+FlexMeasures can serve queries for the most recent beliefs from a materialized database view,
+which is considerably faster than computing them from the beliefs table (see :ref:`mview-data`).
+The view only knows about beliefs recorded before its last refresh, though.
+
+This setting controls whether such queries also look up events recorded since the last refresh
+(the "live tail"), so results are always complete, at some extra query cost.
+If set to ``False``, queries are served from the view alone (events recorded since the last
+refresh are not shown), and users can request inclusion of the latest data per query,
+via a toggle on the asset graphs page or the ``include_live_tail`` API parameter.
+
+Note that in either case, *revised* beliefs about events recorded before the last refresh
+only show up after the next refresh. How stale results can get is bounded by how often you
+schedule the ``flexmeasures db-ops refresh-materialized-views`` command (see :ref:`mview-data`).
+
+Default: ``True``
+
 .. _fallback-redirect-config:
 
 FLEXMEASURES_FALLBACK_REDIRECT

documentation/host/data.rst

@@ -408,6 +408,35 @@ To log out:
    \q
 
 
+.. _mview-data:
+
+Speeding up queries with a materialized view
+---------------------------------------------
+
+FlexMeasures stores multiple beliefs per event (e.g. subsequent forecasts, as well as measurements, from multiple sources).
+The most common query — the most recent belief about each event — therefore requires an expensive aggregation over the beliefs table.
+FlexMeasures maintains a materialized view (``most_recent_beliefs_mview``, created by a database migration) that pre-computes this aggregation,
+speeding up such queries considerably on large databases.
+
+The view is used automatically, but only if it is being refreshed regularly. Set up a cron job to refresh it, for example, every 10 minutes:
+
+.. code-block:: bash
+
+   */10 * * * * flexmeasures db-ops refresh-materialized-views --concurrent
+
+The ``--concurrent`` option avoids locking reads during the refresh, at the cost of higher resource usage.
+
+The time of the last successful refresh is recorded in the ``latest_task_run`` table
+(so the cron schedule is the only place where you control the cadence),
+and can be monitored with ``flexmeasures monitor latest-run --task refresh-materialized-views``.
+If the last successful refresh is older than 24 hours, FlexMeasures stops trusting the view and queries fall back to the beliefs table.
+
+Events recorded since the last refresh are looked up in the beliefs table by default,
+so query results are complete (see :ref:`mview-live-tail-config`).
+Note that *revised* beliefs about previously recorded events only show up after the next refresh,
+so the refresh cadence bounds how stale query results can get.
+
+
 Transaction management
 -----------------------
 

flexmeasures/api/dev/sensors.py

@@ -93,6 +93,8 @@ def get_chart(self, id: int, sensor: Sensor, **kwargs):
             "most_recent_beliefs_only": fields.Boolean(
                 required=False, load_default=True
             ),
+            "use_materialized_view": fields.Boolean(required=False, load_default=True),
+            "include_live_tail": fields.Boolean(required=False, load_default=None),
             "compress_json": fields.Boolean(required=False),
         },
         location="query",
@@ -113,6 +115,8 @@ def get_chart_data(self, id: int, sensor: Sensor, **kwargs):
         - "beliefs_before" (see the `timely-beliefs documentation <https://github.com/SeitaBV/timely-beliefs/blob/main/timely_beliefs/docs/timing.md/#events-and-sensors>`_)
         - "resolution" (see [docs about describing timing](https://flexmeasures.readthedocs.io/latest/api/notation.html#frequency-and-resolution))
         - "most_recent_beliefs_only" (if true, returns the most recent belief for each event; if false, returns each belief for each event; defaults to true)
+        - "use_materialized_view" (if true, the most recent beliefs may be looked up in a materialized view, if available and recently refreshed; defaults to true)
+        - "include_live_tail" (whether queries served by the materialized view also look up events recorded since its last refresh; defaults to the FLEXMEASURES_MVIEW_ALWAYS_INCLUDE_LIVE_TAIL setting)
         """
         return sensor.search_beliefs(as_json=True, **kwargs)
 

flexmeasures/api/v3_0/assets.py

@@ -960,6 +960,8 @@ def get_chart(self, id: int, asset: GenericAsset, **kwargs):
                 required=False, load_default=False
             ),
             "most_recent_beliefs_only": fields.Boolean(required=False),
+            "use_materialized_view": fields.Boolean(required=False, load_default=True),
+            "include_live_tail": fields.Boolean(required=False, load_default=None),
             "compress_json": fields.Boolean(required=False),
         },
         location="query",

flexmeasures/cli/db_ops.py

@@ -9,6 +9,7 @@
 import click
 
 from flexmeasures.cli.utils import MsgStyle
+from flexmeasures.data import db
 
 
 @click.group("db-ops")
@@ -95,4 +96,65 @@ def restore(file: str):
         click.secho("db restore unsuccessful", **MsgStyle.ERROR)
 
 
+@fm_db_ops.command("refresh-materialized-views")
+@with_appcontext
+@click.option(
+    "--concurrent",
+    is_flag=True,
+    default=False,
+    help="Refresh without locking reads on the materialized view, at the cost of higher resource usage."
+    " Requires the unique index created by the corresponding migration.",
+)
+def refresh_materialized_views(concurrent: bool):
+    """
+    Refresh the materialized view that caches the most recent beliefs (for faster queries).
+
+    By default, this locks the materialized view for the duration of the refresh.
+    Use the --concurrent option to avoid locking reads, at the cost of higher resource usage
+    (this requires the unique index that the corresponding migration created).
+
+    Run this periodically (e.g. from a cron job) to bound how stale the cached data can get.
+    The time of the last successful run is recorded in the latest_task_run table, which serves
+    as the queries' cutoff between trusting the view and reading recent events from the beliefs
+    table, and can be monitored with ``flexmeasures monitor latest-run``.
+    """
+    import time
+
+    from sqlalchemy import text
+    from timely_beliefs.beliefs.materialized_views import refresh_mview_ddl
+
+    from flexmeasures.data.transactional import task_with_status_report
+    from flexmeasures.data.models.task_runs import LatestTaskRun
+    from flexmeasures.data.services.materialized_views import MVIEW_REFRESH_TASK_NAME
+
+    @task_with_status_report(MVIEW_REFRESH_TASK_NAME)
+    def _refresh():
+        ddl = refresh_mview_ddl(concurrently=concurrent)
+        if concurrent:
+            # REFRESH MATERIALIZED VIEW CONCURRENTLY cannot run inside a transaction block
+            with db.engine.connect().execution_options(
+                isolation_level="AUTOCOMMIT"
+            ) as connection:
+                connection.execute(text(ddl))
+        else:
+            db.session.execute(text(ddl))
+
+    start_time = time.time()
+    click.secho(
+        f"Refreshing materialized view{' concurrently' if concurrent else ''}...",
+        **MsgStyle.WARN,
+    )
+    _refresh()
+
+    # The task decorator recorded success/failure in the db; convey it as an exit status, too
+    task_run = db.session.get(LatestTaskRun, MVIEW_REFRESH_TASK_NAME)
+    if task_run is None or not task_run.status:
+        click.secho("Refreshing the materialized view failed.", **MsgStyle.ERROR)
+        raise click.Abort()
+    click.secho(
+        f"Materialized view refreshed in {time.time() - start_time:.2f} seconds.",
+        **MsgStyle.SUCCESS,
+    )
+
+
 app.cli.add_command(fm_db_ops)

flexmeasures/cli/tests/test_db_ops.py

@@ -0,0 +1,94 @@
+import pytest
+from sqlalchemy import select, text
+from timely_beliefs.beliefs.materialized_views import (
+    MOST_RECENT_BELIEFS_MVIEW,
+    create_mview_ddl,
+    create_mview_indexes_ddl,
+    drop_mview_ddl,
+)
+
+from flexmeasures.data.models.task_runs import LatestTaskRun
+from flexmeasures.data.models.time_series import TimedBelief
+from flexmeasures.data.services.materialized_views import MVIEW_REFRESH_TASK_NAME
+
+
+@pytest.fixture(scope="function")
+def setup_mview(fresh_db):
+    """Create the materialized view (as the corresponding migration would) and let the app know it exists."""
+    from flexmeasures.data import config as data_config
+
+    fresh_db.session.execute(text(drop_mview_ddl()))
+    fresh_db.session.execute(text(create_mview_ddl(TimedBelief)))
+    fresh_db.session.execute(text(create_mview_indexes_ddl()))
+
+    # The app checks for the view at startup, i.e. before this fixture created it
+    original_mview = data_config.most_recent_beliefs_mview
+    data_config.most_recent_beliefs_mview = MOST_RECENT_BELIEFS_MVIEW
+    yield
+    data_config.most_recent_beliefs_mview = original_mview
+    # Commit the drop, because the refresh CLI command commits, thereby persisting the view
+    fresh_db.session.execute(text(drop_mview_ddl()))
+    fresh_db.session.commit()
+
+
+def test_refresh_materialized_views(app, fresh_db, setup_mview):
+    """The CLI command should refresh the view and record its run in the latest_task_run table."""
+    from flexmeasures.cli.db_ops import refresh_materialized_views
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(refresh_materialized_views)
+    assert result.exit_code == 0
+
+    task_run = fresh_db.session.get(LatestTaskRun, MVIEW_REFRESH_TASK_NAME)
+    assert task_run is not None
+    assert task_run.status is True
+
+
+def test_refresh_materialized_views_failure(app, fresh_db):
+    """Without the view in place, the CLI command should fail and record the failed run."""
+    from flexmeasures.cli.db_ops import refresh_materialized_views
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(refresh_materialized_views)
+    assert result.exit_code != 0
+
+    task_run = fresh_db.session.get(LatestTaskRun, MVIEW_REFRESH_TASK_NAME)
+    assert task_run is not None
+    assert task_run.status is False
+
+
+def test_search_beliefs_with_mview(app, fresh_db, setup_mview, setup_beliefs_fresh_db):
+    """After a refresh, searching via the materialized view should match searching the beliefs table."""
+    from flexmeasures.cli.db_ops import refresh_materialized_views
+    import pandas as pd
+
+    from flexmeasures.data.models.time_series import Sensor
+
+    # Commit the beliefs before refreshing, so the view can see them
+    fresh_db.session.commit()
+
+    runner = app.test_cli_runner()
+    result = runner.invoke(refresh_materialized_views)
+    assert result.exit_code == 0
+
+    sensor = fresh_db.session.execute(
+        select(Sensor).filter_by(name="epex_da")
+    ).scalar_one()
+
+    bdf_live = sensor.search_beliefs(
+        most_recent_beliefs_only=True, use_materialized_view=False
+    )
+    assert not bdf_live.empty
+
+    bdf_mview = sensor.search_beliefs(
+        most_recent_beliefs_only=True, use_materialized_view=True
+    )
+    pd.testing.assert_frame_equal(bdf_mview, bdf_live)
+
+    # Also without the live tail (i.e. from the view alone), because all data was recorded before the refresh
+    bdf_mview_only = sensor.search_beliefs(
+        most_recent_beliefs_only=True,
+        use_materialized_view=True,
+        include_live_tail=False,
+    )
+    pd.testing.assert_frame_equal(bdf_mview_only, bdf_live)

flexmeasures/data/config.py

@@ -16,6 +16,7 @@
 )
 Base = None  # type: ignore
 session_options = None
+most_recent_beliefs_mview = None
 
 
 def init_db():
@@ -32,7 +33,7 @@ def init_db():
 def configure_db_for(app: Flask):
     """Call this to configure the database and the tools we use on it for the Flask app.
     This should only be called once in the app's lifetime."""
-    global db, Base
+    global db, Base, most_recent_beliefs_mview
 
     with app.app_context():
         db.init_app(app)
@@ -51,6 +52,19 @@ def configure_db_for(app: Flask):
             forecasting,
         )  # noqa: F401
 
+        from timely_beliefs.beliefs.materialized_views import (
+            get_most_recent_beliefs_mview,
+        )
+
+        try:
+            most_recent_beliefs_mview = get_most_recent_beliefs_mview(db.session)
+        except Exception:
+            app.logger.warning(
+                "Could not determine whether the most_recent_beliefs_mview materialized view exists."
+                " Beliefs will be retrieved from the beliefs table instead of the materialized view.",
+            )
+            db.session.rollback()
+
         # This would create db structure based on models, but you should use `flask db upgrade` for that.
         # Base.metadata.create_all(bind=db.engine)
 

flexmeasures/data/migrations/versions/timed_beliefs_materialized_views.py

@@ -0,0 +1,63 @@
+"""Add materialized view caching the most recent belief per event per source
+
+The SQL below is frozen output of the DDL generators in timely_beliefs.beliefs.materialized_views
+(generated against timely-beliefs' TimedBeliefDBMixin for FlexMeasures' timed_belief table),
+so that this migration stays immutable while timely-beliefs owns the canonical view definition.
+
+Revision ID: c98798csds8c
+Revises: 55d8936a55f9
+Create Date: 2025-08-08 04:55:33.722545
+
+"""
+
+from alembic import op
+
+# revision identifiers
+revision = "c98798csds8c"
+down_revision = "55d8936a55f9"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute(
+        """
+        CREATE MATERIALIZED VIEW most_recent_beliefs_mview AS
+        SELECT
+            sensor_id,
+            event_start,
+            source_id,
+            MIN(belief_horizon) AS most_recent_belief_horizon
+        FROM timed_belief
+        GROUP BY
+            sensor_id,
+            event_start,
+            source_id;
+    """
+    )
+
+    op.execute(
+        """
+        CREATE INDEX idx_most_recent_beliefs_mview_sensor_event
+            ON most_recent_beliefs_mview (sensor_id, event_start);
+    """
+    )
+
+    op.execute(
+        """
+        CREATE INDEX idx_most_recent_beliefs_mview_event_start
+            ON most_recent_beliefs_mview (event_start);
+    """
+    )
+
+    # A unique index is required to allow concurrent refreshes
+    op.execute(
+        """
+        CREATE UNIQUE INDEX idx_most_recent_beliefs_mview_unique
+            ON most_recent_beliefs_mview (sensor_id, event_start, source_id);
+    """
+    )
+
+
+def downgrade():
+    op.execute("DROP MATERIALIZED VIEW IF EXISTS most_recent_beliefs_mview;")

flexmeasures/data/models/generic_assets.py

@@ -1128,6 +1128,8 @@ def search_beliefs(  # noqa C901
         as_json: bool = False,
         compress_json: bool = False,
         resolution: timedelta | None = None,
+        use_materialized_view: bool = True,
+        include_live_tail: bool | None = None,
     ) -> BeliefsDataFrame | str:
         """Search all beliefs about events for all sensors of this asset
 
@@ -1172,6 +1174,8 @@ def search_beliefs(  # noqa C901
                 most_recent_events_only=most_recent_events_only,
                 one_deterministic_belief_per_event_per_source=True,
                 resolution=resolution,
+                use_materialized_view=use_materialized_view,
+                include_live_tail=include_live_tail,
             )
         if as_json and not compress_json:
             from flexmeasures.data.services.time_series import simplify_index