diff --git a/documentation/changelog.rst b/documentation/changelog.rst index f2ce661c91..ca77a390c4 100644 --- a/documentation/changelog.rst +++ b/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 `_] * 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 `_] * 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 `_] +* 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 `_] * 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 `_ and `PR #2360 `_] Bugfixes diff --git a/flexmeasures/data/migrations/versions/9f2c1d7b3a44_shrink_timed_belief_values_to_float4.py b/flexmeasures/data/migrations/versions/9f2c1d7b3a44_shrink_timed_belief_values_to_float4.py new file mode 100644 index 0000000000..1be57e83f5 --- /dev/null +++ b/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, + ) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index 537f5031ac..9aadf9b243 100644 --- a/flexmeasures/data/models/time_series.py +++ b/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) diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index 9b20d4a946..2536730335 100644 --- a/flexmeasures/data/services/sensors.py +++ b/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) diff --git a/flexmeasures/data/services/time_series.py b/flexmeasures/data/services/time_series.py index bec17970da..9b8f6a121a 100644 --- a/flexmeasures/data/services/time_series.py +++ b/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 diff --git a/flexmeasures/data/tests/test_belief_value_precision.py b/flexmeasures/data/tests/test_belief_value_precision.py new file mode 100644 index 0000000000..4f18afd50f --- /dev/null +++ b/flexmeasures/data/tests/test_belief_value_precision.py @@ -0,0 +1,168 @@ +"""Tests for storing belief values at single precision (float4). + +See the column overrides on flexmeasures.data.models.time_series.TimedBelief and the +``9f2c1d7b3a44`` migration. +""" + +import pandas as pd +import pytest +from sqlalchemy import Float +from timely_beliefs import BeliefsDataFrame + +from flexmeasures.data.models.data_sources import DataSource +from flexmeasures.data.models.time_series import TimedBelief +from flexmeasures.data.services.sensors import get_sensor_stats +from flexmeasures.data.services.time_series import round_to_stored_precision +from flexmeasures.data.utils import save_to_db +from flexmeasures.tests.utils import get_test_sensor + + +@pytest.mark.parametrize("column_name", ["cumulative_probability", "event_value"]) +def test_value_columns_are_single_precision(column_name): + """The two value columns must be declared as float4, not float8. + + This is what the ``9f2c1d7b3a44`` migration puts in the database, so the ORM has to + agree -- otherwise a schema built by create_all() (as the test suite does) silently + diverges from every migrated deployment. It also guards against a future + timely_beliefs release quietly reclaiming these columns. + """ + column = TimedBelief.__table__.columns[column_name] + assert isinstance(column.type, Float) + assert column.type.precision == 24, ( + f"{column_name} is no longer declared as float4 " + f"(precision={column.type.precision})" + ) + + +def test_rounding_follows_the_column_rather_than_a_hardcoded_precision(monkeypatch): + """No rounding may happen while the column is still float8. + + There is a window between deploying this code and running the ``9f2c1d7b3a44`` + migration in which the column is still double precision. Rounding to float4 during + that window would classify a genuine update as "unchanged" and silently drop it, so + the precision has to be read from the column rather than assumed. + """ + from flexmeasures.data.services import time_series as time_series_service + + # Pretend the column is still double precision, as it is pre-migration + monkeypatch.setattr(time_series_service, "_stored_dtype", lambda field: None) + + high_precision_value = 1234567.891 + df = pd.DataFrame( + {"event_value": [high_precision_value], "cumulative_probability": [0.5]} + ) + rounded = time_series_service.round_to_stored_precision(df) + + assert ( + rounded["event_value"][0] == high_precision_value + ), "values were rounded even though the column is not narrowed" + + +def test_stored_dtype_reads_the_mapped_column(): + """The stored dtype must be derived from the column, and be float32 today.""" + from flexmeasures.data.services.time_series import _stored_dtype + + _stored_dtype.cache_clear() + assert _stored_dtype("event_value") == "float32" + assert _stored_dtype("cumulative_probability") == "float32" + + +def test_round_to_stored_precision_leaves_input_untouched(): + """The helper must not mutate the frame it is handed.""" + df = pd.DataFrame( + {"event_value": [1234567.891], "cumulative_probability": [0.5], "other": [1.0]} + ) + rounded = round_to_stored_precision(df) + + assert df["event_value"][0] == 1234567.891, "input frame was mutated" + assert rounded["event_value"][0] != 1234567.891, "value was not rounded" + assert rounded["event_value"][0] == pytest.approx(1234567.891, rel=1e-6) + # Columns that are not stored as float4 must pass through untouched + assert rounded["other"][0] == 1.0 + + +def test_resubmitting_a_high_precision_value_is_dropped_as_unchanged(setup_beliefs, db): + """Re-submitting a value needing more than float4 precision must be a no-op. + + The database rounds the stored value to float4, so a candidate carrying the original + double-precision value no longer compares equal to it. Without rounding the + comparison to the precision the value is actually stored at, the candidate looks + like a changed belief on every submission -- and since its belief time is unchanged + too, saving it raises a duplicate key violation instead of being quietly skipped. + """ + sensor = get_test_sensor(db) + source = DataSource(name="High precision source", type="demo script") + db.session.add(source) + db.session.commit() + + event_start = pd.Timestamp("2021-03-28 16:00:00+00:00") + belief_time = pd.Timestamp("2021-03-27 08:00:00+00:00") + # Needs 10 significant digits, so float4 (~7) cannot represent it exactly + high_precision_value = 1234567.891 + assert float(pd.Series([high_precision_value]).astype("float32")[0]) != ( + high_precision_value + ), "test value is representable in float4, so it proves nothing" + + def candidate() -> BeliefsDataFrame: + return BeliefsDataFrame( + [ + TimedBelief( + sensor=sensor, + source=source, + event_start=event_start, + belief_time=belief_time, + event_value=high_precision_value, + ) + ] + ) + + save_to_db(candidate()) + db.session.commit() + num_beliefs = len(sensor.search_beliefs(most_recent_beliefs_only=False)) + + # Submitting the very same value again must not raise, and must not store anything + save_to_db(candidate()) + db.session.commit() + + assert len(sensor.search_beliefs(most_recent_beliefs_only=False)) == num_beliefs + + +def test_sum_over_values_does_not_accumulate_in_single_precision(setup_beliefs, db): + """Sensor stats must sum in double precision, even though values are stored float4. + + PostgreSQL's ``sum(real)`` accumulates in ``real``. Once the running total reaches + 2^24, adding 1 to it is a no-op, so a sum over many rows silently loses whole units. + Both values used here are exactly representable in float4, so any error observed is + the accumulator's, not the storage's. + """ + sensor = get_test_sensor(db) + source = DataSource(name="Accumulator source", type="demo script") + db.session.add(source) + db.session.commit() + + large_value = 2.0**24 # 16777216, where the float4 gap between values is 2 + num_ones = 100 + belief_time = pd.Timestamp("2021-03-27 08:00:00+00:00") + event_start = pd.Timestamp("2021-03-28 16:00:00+00:00") + + beliefs = [ + TimedBelief( + sensor=sensor, + source=source, + event_start=event_start + pd.Timedelta(hours=i), + belief_time=belief_time, + event_value=large_value if i == 0 else 1.0, + ) + for i in range(num_ones + 1) + ] + save_to_db(BeliefsDataFrame(beliefs)) + db.session.commit() + + stats = get_sensor_stats(sensor, "", "", from_cache=False) + our_stats = [v for k, v in stats.items() if f"(ID: {source.id})" in k] + assert len(our_stats) == 1, f"expected one row for our source, got {stats.keys()}" + + # Accumulating in float4 would return large_value, having dropped every single 1.0 + assert our_stats[0]["Sum over values"] == pytest.approx( + large_value + num_ones, abs=0.5 + )