Shrink timed_belief value columns from float8 to float4

documentation/changelog.rst

@@ -12,6 +12,8 @@ v1.0.0 | July XX, 2026
 
 .. warning:: Upgrading to this version requires running ``flexmeasures db upgrade`` (you can create a backup first with ``flexmeasures db-ops dump``).
 
+.. warning:: This release stores sensor data at single precision (float4, ~7 significant decimal digits) instead of double precision. The migration rewrites the whole ``timed_belief`` table and rebuilds its primary key, so on a large database it is a heavy, table-locking operation — plan a maintenance window. Stored values are rounded to the nearest single-precision value, and this is not reversible: downgrading restores the column type, not the lost digits. Check your data first if you record values whose magnitude leaves little headroom at 7 significant digits, such as cumulative meter readings or currency totals in the millions.
+
 New features
 -------------
 
@@ -70,6 +72,7 @@ Infrastructure / Support
 * Add ``FLEXMEASURES_DEFAULT_JOB_TIMEOUT`` and ``FLEXMEASURES_JOB_TIMEOUT`` settings for configuring RQ job timeouts globally and per queue, and log actionable guidance when a forecasting job times out [see `PR #2318 <https://github.com/FlexMeasures/flexmeasures/pull/2318>`_]
 * Stop manual runs of the Docker publishing workflow from overwriting the ``latest`` image tag, and let them opt in to it explicitly [see `PR #2316 <https://www.github.com/FlexMeasures/flexmeasures/pull/2316>`_]
 * Add a pre-commit hook that blocks image files (png, jpg, gif, bmp, tiff, webp, ico, psd) from being committed outside of ``flexmeasures/ui/static/`` and ``documentation/``, to protect the git history from binary bloat; screenshots belong in the ``FlexMeasures/screenshots`` repo instead [see `PR #2315 <https://www.github.com/FlexMeasures/flexmeasures/pull/2315>`_]
+* Store the ``timed_belief`` ``cumulative_probability`` and ``event_value`` columns as single-precision floats (float4) instead of double-precision (float8), shrinking the table's heap by roughly 9% (about 4% of its total on-disk footprint once its indexes are counted, as those do not shrink) [see `PR #2332 <https://www.github.com/FlexMeasures/flexmeasures/pull/2332>`_]
 * Schedulers track devices via a typed device inventory, which classifies every flex-model entry once and serves as the single source of truth for device roles and canonical device indices [see `PR #2321 <https://www.github.com/FlexMeasures/flexmeasures/pull/2321>`_ and `PR #2360 <https://www.github.com/FlexMeasures/flexmeasures/pull/2360>`_]
 
 Bugfixes

flexmeasures/data/migrations/versions/9f2c1d7b3a44_shrink_timed_belief_values_to_float4.py

@@ -0,0 +1,69 @@
+"""shrink timed_belief value columns from float8 to float4
+
+Change ``cumulative_probability`` and ``event_value`` on ``timed_belief`` from
+double precision (float8, 8 bytes) to single precision (float4/REAL, 4 bytes).
+These are the two per-row numeric value columns on what is typically the
+largest table.
+
+Measured on 12.9M rows of production data (sensors 5 and 14 of the ems
+production dump of 2026-07-28), this saves 8 bytes of an 84.5-byte row: the
+heap shrinks by 9.4%, but the indexes do not shrink at all, so the table's
+total on-disk footprint drops by only ~3.9%. The 8 bytes are a smaller share
+of the row than they look, because of the 23-byte tuple header and the
+16-byte belief_horizon interval.
+
+Single precision keeps ~7 significant decimal digits, which is plenty for
+sensor readings and for a cumulative probability in [0, 1]. Values already
+stored are rounded to the nearest float4 by the implicit cast.
+
+Note: ``ALTER COLUMN ... TYPE`` rewrites the whole table (and rebuilds the
+primary key, since ``cumulative_probability`` is part of it), so on a large
+``timed_belief`` this is a heavy, table-locking operation — run it during a
+maintenance window.
+
+Note: the downgrade restores the column type, not the precision. Digits lost by
+the upgrade's rounding are gone for good.
+
+Revision ID: 9f2c1d7b3a44
+Revises: 3c2f9e5a1d47
+Create Date: 2026-07-20
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "9f2c1d7b3a44"
+down_revision = "3c2f9e5a1d47"
+branch_labels = None
+depends_on = None
+
+# Single precision (float4/REAL) is a Float with precision <= 24; double
+# precision (float8) is a bare Float. See the matching ORM columns on
+# flexmeasures.data.models.time_series.TimedBelief.
+FLOAT4 = sa.Float(precision=24)
+FLOAT8 = sa.Float()
+
+
+def upgrade():
+    for column in ("cumulative_probability", "event_value"):
+        op.alter_column(
+            "timed_belief",
+            column,
+            type_=FLOAT4,
+            existing_type=FLOAT8,
+            existing_nullable=False,
+        )
+
+
+def downgrade():
+    for column in ("cumulative_probability", "event_value"):
+        op.alter_column(
+            "timed_belief",
+            column,
+            type_=FLOAT8,
+            existing_type=FLOAT4,
+            existing_nullable=False,
+        )

flexmeasures/data/models/time_series.py

@@ -925,6 +925,31 @@ class TimedBelief(db.Model, tb.TimedBeliefDBMixin):
     It also records the source of the belief, and the sensor that the event pertains to.
     """
 
+    # Store these as single-precision floats (float4/REAL, 4 bytes) rather than the
+    # double-precision (float8, 8 bytes) that timely_beliefs' TimedBeliefDBMixin uses.
+    # This halves the width of the two per-row value columns, saving 8 bytes of an
+    # 84.5-byte row: measured on production data, the timed_belief heap shrinks by
+    # ~9.4%, and its total on-disk footprint by ~3.9% (indexes do not shrink).
+    # ~7 significant digits of single precision are plenty for sensor readings and for
+    # a cumulative probability in [0, 1]. Keep in sync with the matching migration.
+    #
+    # TODO: replace this override with timely_beliefs' value_column_type hook
+    #       (SeitaBV/timely-beliefs#242) once a release carrying it is out:
+    #
+    #           value_column_type = db.Float(precision=24)
+    #
+    #       Redeclaring the columns here works, but because mixin columns are ordered by
+    #       creation order it moves both to the front of the table and of the primary
+    #       key, making cumulative_probability (nearly always 0.5) the leading key
+    #       column. That only affects schemas built by create_all() -- i.e. the test
+    #       schema, since deployments run `flexmeasures db upgrade` -- so it is a
+    #       tolerable interim, but it does mean tests exercise a different index shape
+    #       than production. The upstream hook keeps the ordering identical.
+    cumulative_probability = db.Column(
+        db.Float(precision=24), nullable=False, primary_key=True, default=0.5
+    )
+    event_value = db.Column(db.Float(precision=24), nullable=False)
+
     @declared_attr
     def source_id(cls):
         return db.Column(db.Integer, db.ForeignKey("data_source.id"), primary_key=True)

flexmeasures/data/services/sensors.py

@@ -838,8 +838,16 @@ def _get_sensor_stats(
     # We pass it to aggregate FILTER clauses so that the planner can compute all aggregates in a single pass over the belief rows.
     not_nan = TimedBelief.event_value != float("nan")
 
-    def filtered_agg(func):
-        return func(TimedBelief.event_value).filter(not_nan)
+    # event_value is stored as float4 (see TimedBelief), and PostgreSQL's sum() over a
+    # float4 column both returns and *accumulates in* float4. Over a sensor with many
+    # rows the running total outgrows the magnitude of the individual values being added
+    # to it, and the sum drifts visibly. So sum over an explicit float8 cast.
+    # (avg() already accumulates in float8, and min()/max() pick an existing value, so
+    # only sum() needs this.)
+    def filtered_agg(func, column=TimedBelief.event_value):
+        return func(column).filter(not_nan)
+
+    event_value_as_float8 = sa.cast(TimedBelief.event_value, sa.Float())
 
     q = (
         sa.select(
@@ -854,7 +862,7 @@ def filtered_agg(func):
             filtered_agg(sa.func.min).label("min_event_value"),
             filtered_agg(sa.func.max).label("max_event_value"),
             filtered_agg(sa.func.avg).label("avg_event_value"),
-            filtered_agg(sa.func.sum).label("sum_event_value"),
+            filtered_agg(sa.func.sum, event_value_as_float8).label("sum_event_value"),
             sa.func.count(TimedBelief.event_value).label("count_event_value"),
         )
         .select_from(TimedBelief)

flexmeasures/data/services/time_series.py

@@ -2,6 +2,7 @@
 
 from typing import Any
 from datetime import timedelta
+from functools import lru_cache
 
 import inflect
 from flask import current_app
@@ -13,6 +14,57 @@
 
 p = inflect.engine()
 
+#: The ``timed_belief`` columns whose stored precision we need to account for when
+#: deciding whether a belief has changed.
+VALUE_FIELDS = ("cumulative_probability", "event_value")
+
+
+@lru_cache(maxsize=None)
+def _stored_dtype(field: str) -> str | None:
+    """Return the numpy dtype ``field`` is stored at, or None if it is not narrowed.
+
+    Read from the mapped column rather than hardcoded, so this keeps telling the truth
+    if the column type changes -- including during the window between deploying code
+    that expects float4 and actually running the migration that narrows the column.
+    Getting that backwards is not a missed optimization: rounding to float4 while the
+    database is still float8 would classify a real update as "unchanged" and silently
+    drop it.
+
+    This is an attribute lookup on already-loaded table metadata, not a query.
+    """
+    # Imported here because flexmeasures.data.models.time_series imports this module
+    from flexmeasures.data.models.time_series import TimedBelief
+
+    precision = TimedBelief.__table__.columns[field].type.precision
+    # A Float with precision <= 24 is single precision (float4/REAL) on PostgreSQL;
+    # anything wider (precision None means a bare Float, i.e. float8) needs no rounding.
+    if precision is not None and precision <= 24:
+        return "float32"
+    return None
+
+
+def round_to_stored_precision(df: pd.DataFrame) -> pd.DataFrame:
+    """Round the value fields of a reset-index belief frame to the precision they are stored at.
+
+    Beliefs read back from the database have been rounded to whatever the column stores,
+    while candidate beliefs still carry the full double precision they were submitted or
+    computed with. Where the column is narrower than float8, comparing the two directly
+    would report any value needing more than ~7 significant digits as "changed" on every
+    single submission, so the same data would be re-saved forever.
+
+    Rounding both sides to the precision they will actually be stored at makes the
+    comparison ask the question we mean to ask: will storing this candidate change what
+    is in the database?
+
+    Operates on a copy; ``df`` is left untouched.
+    """
+    df = df.copy()
+    for field in VALUE_FIELDS:
+        dtype = _stored_dtype(field)
+        if dtype is not None and field in df.columns:
+            df[field] = df[field].astype(dtype).astype("float64")
+    return df
+
 
 def aggregate_values(bdf_dict: dict[Any, tb.BeliefsDataFrame]) -> tb.BeliefsDataFrame:
     # todo: test this function rigorously, e.g. with empty bdfs in bdf_dict
@@ -81,17 +133,17 @@ def drop_unchanged_beliefs(bdf: tb.BeliefsDataFrame) -> tb.BeliefsDataFrame:
         bdf = pd.concat([ex_ante_bdf, ex_post_bdf])
         return bdf
 
-    # Remove unchanged beliefs from within the new data itself
+    # Remove unchanged beliefs from within the new data itself.
+    # Deduplicate at stored precision, for the same reason we compare against the
+    # database at stored precision, but keep the caller's original values in the rows
+    # we do retain.
     index_names = bdf.index.names
-    bdf = (
-        bdf.sort_index()
-        .reset_index()
-        .drop_duplicates(
-            ["event_start", "source", "cumulative_probability", "event_value"],
-            keep="first",
-        )
-        .set_index(index_names)
+    flat_bdf = bdf.sort_index().reset_index()
+    already_seen = round_to_stored_precision(flat_bdf).duplicated(
+        ["event_start", "source", "cumulative_probability", "event_value"],
+        keep="first",
     )
+    bdf = flat_bdf[~already_seen].set_index(index_names)
 
     # Remove unchanged beliefs with respect to what is already stored in the database
     if bdf.belief_horizons[0] > timedelta(0):
@@ -184,8 +236,11 @@ def _drop_unchanged_beliefs_compared_to_db(
         "cumulative_probability",
         "event_value",
     ]
-    a = a_df.set_index(compare_fields)
-    b = b_df.set_index(compare_fields)
+    # Compare at the precision the values are stored at, not the precision they arrived
+    # with (see round_to_stored_precision). Only the comparison is rounded: the frame we
+    # return still carries the caller's original values.
+    a = round_to_stored_precision(a_df).set_index(compare_fields)
+    b = round_to_stored_precision(b_df).set_index(compare_fields)
     dropped = a.drop(b.index, errors="ignore", axis=0)
 
     # Keep whole probabilistic beliefs, not just the parts that changed