From 80c7a197dcd6135afb1c012eff62449538445fcd Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Thu, 13 Aug 2026 07:31:03 +0100 Subject: [PATCH] fix(sql): Cleanup array migration --- internal/server/postgres/.sqlc.yaml | 62 +- internal/server/postgres/dataserverimpl.go | 51 -- internal/server/postgres/mappers.go | 13 - internal/server/postgres/mappers_test.go | 42 -- .../migrations/00013_split_rebuild_swap.sql | 604 +++++++++--------- .../sql/migrations/00014_drop_row_storage.sql | 201 ++++++ .../postgres/sql/queries/predictions.sql | 267 +++----- internal/server/postgres/testdata/seeding.sql | 46 +- 8 files changed, 608 insertions(+), 678 deletions(-) create mode 100644 internal/server/postgres/sql/migrations/00014_drop_row_storage.sql diff --git a/internal/server/postgres/.sqlc.yaml b/internal/server/postgres/.sqlc.yaml index b4cdad4..9f2de2d 100644 --- a/internal/server/postgres/.sqlc.yaml +++ b/internal/server/postgres/.sqlc.yaml @@ -16,78 +16,30 @@ sql: emit_interface: true emit_exported_queries: true overrides: - - column: "winning_predictions.p02_sip" + - column: "*.p02_sip" go_type: type: "int16" pointer: true - - column: "winning_predictions.p10_sip" + - column: "*.p10_sip" go_type: type: "int16" pointer: true - - column: "winning_predictions.p25_sip" + - column: "*.p25_sip" go_type: type: "int16" pointer: true - - column: "winning_predictions.p75_sip" + - column: "*.p50_sip" go_type: type: "int16" - pointer: true - - column: "winning_predictions.p90_sip" - go_type: - type: "int16" - pointer: true - - column: "winning_predictions.p98_sip" - go_type: - type: "int16" - pointer: true - - column: "expanded.p02_sip" - go_type: - type: "int16" - pointer: true - - column: "expanded.p10_sip" - go_type: - type: "int16" - pointer: true - - column: "expanded.p25_sip" - go_type: - type: "int16" - pointer: true - - column: "expanded.p50_sip" - go_type: - type: "int16" - - column: "expanded.p75_sip" - go_type: - type: "int16" - pointer: true - - column: "expanded.p90_sip" - go_type: - type: "int16" - pointer: true - - column: "expanded.p98_sip" - go_type: - type: "int16" - pointer: true - - column: "ListPredictionsAtTimeForLocations.p02_sip" - go_type: - type: "int16" - pointer: true - - column: "ListPredictionsAtTimeForLocations.p10_sip" - go_type: - type: "int16" - pointer: true - - column: "ListPredictionsAtTimeForLocations.p25_sip" - go_type: - type: "int16" - pointer: true - - column: "ListPredictionsAtTimeForLocations.p75_sip" + - column: "*.p75_sip" go_type: type: "int16" pointer: true - - column: "ListPredictionsAtTimeForLocations.p90_sip" + - column: "*.p90_sip" go_type: type: "int16" pointer: true - - column: "ListPredictionsAtTimeForLocations.p98_sip" + - column: "*.p98_sip" go_type: type: "int16" pointer: true diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index 5c0cd98..6b64d31 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -109,31 +109,6 @@ func (s *DataPlatformDataServiceServerImpl) CreateForecast( return nil, fmt.Errorf("invalid forecast: %w", err) } - // Create the forecast data - paramsList := make([]db.CreatePredictedValuesParams, len(req.Values)) - for i, value := range req.Values { - paramsList[i] = db.CreatePredictedValuesParams{ - HorizonMins: int16(value.HorizonMins), - P02Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p02"), - P10Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p10"), - P25Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p25"), - P50Sip: int16(value.P50Fraction * 30000.0), - P75Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p75"), - P90Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p90"), - P98Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p98"), - ForecastUuid: fParams.ForecastUuid, - } - } - - count, err := querier.CreatePredictedValues(ctx, paramsList) - if err != nil || count < int64(len(req.Values)) { - if err == nil { - err = errors.New("inserted count less than requested") - } - - return nil, fmt.Errorf("invalid predicted generation values: %w", err) - } - l.Debug(). Str("dp.forecast.uuid", fParams.ForecastUuid.String()). Str("dp.geometry.uuid", fParams.GeometryUuid.String()). @@ -1481,7 +1456,6 @@ func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts( var ( forecastParams []db.CreateForecastsParams - valueParams []db.CreatePredictedValuesParams createdUuids []string batchUuids []string ) @@ -1519,20 +1493,10 @@ func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts( return fmt.Errorf("failed to insert forecasts batch: %w", err) } - countV, err := querier.CreatePredictedValues(ctx, valueParams) - if err != nil || countV < int64(len(valueParams)) { - if err == nil { - err = errors.New("inserted predicted values count less than requested") - } - - return fmt.Errorf("failed to insert predicted values batch: %w", err) - } - createdUuids = append(createdUuids, batchUuids...) // Reset batch buffers forecastParams = forecastParams[:0] - valueParams = valueParams[:0] batchUuids = batchUuids[:0] batchesProcessed++ @@ -1633,21 +1597,6 @@ func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts( } forecastParams = append(forecastParams, fParams) - - for _, value := range req.Values { - valueParams = append(valueParams, db.CreatePredictedValuesParams{ - HorizonMins: int16(value.HorizonMins), - P02Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p02"), - P10Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p10"), - P25Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p25"), - P50Sip: int16(value.P50Fraction * 30000.0), - P75Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p75"), - P90Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p90"), - P98Sip: extractSIPStatPtrFromMap(value.OtherStatisticsFractions, "p98"), - ForecastUuid: fParams.ForecastUuid, - }) - } - batchUuids = append(batchUuids, fParams.ForecastUuid.String()) // Flush if we hit the batch size limit diff --git a/internal/server/postgres/mappers.go b/internal/server/postgres/mappers.go index 57bcc1a..9ab13b1 100644 --- a/internal/server/postgres/mappers.go +++ b/internal/server/postgres/mappers.go @@ -58,19 +58,6 @@ func timeptrToPgTimestamp(t *timestamppb.Timestamp) pgtype.Timestamp { return pgtype.Timestamp{Time: t.AsTime().UTC(), Valid: true} } -// extractSIPStatPtrFromMap gets a key's value from a map as a pointer, and converts it -// to a smallint percentage. If it doesn't exist, it returns nil. -func extractSIPStatPtrFromMap(m map[string]float32, key string) *int16 { - val, exists := m[key] - if !exists { - return nil - } - - sip_val := int16(val * 30000.0) - - return &sip_val -} - // extractSIPStatSlice builds the array for a single p-level from a forecast's values. // Returns nil if no value in the series carries this statistic, so the column is stored as a // NULL array rather than a materialised array of nulls (~3 bytes per forecast against ~130). diff --git a/internal/server/postgres/mappers_test.go b/internal/server/postgres/mappers_test.go index ec38127..ac0acc9 100644 --- a/internal/server/postgres/mappers_test.go +++ b/internal/server/postgres/mappers_test.go @@ -83,48 +83,6 @@ func Test_timeptrToPgTimestamp(t *testing.T) { } } -func Test_extractSIPStatPtrFromMap(t *testing.T) { - m := map[string]float32{"p10": 0.1, "p90": 0.9} - - tests := []struct { - name string - inputMap map[string]float32 - key string - expected *int16 - }{ - { - name: "nil map returns nil", - inputMap: nil, - key: "p10", - expected: nil, - }, - { - name: "missing key returns nil", - inputMap: m, - key: "p50", - expected: nil, - }, - { - name: "existing key returns sip value", - inputMap: m, - key: "p10", - expected: func() *int16 { v := int16(3000); return &v }(), - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - res := extractSIPStatPtrFromMap(tt.inputMap, tt.key) - if tt.expected == nil { - require.Nil(t, res) - } else { - require.NotNil(t, res) - require.Equal(t, *tt.expected, *res) - } - }) - } -} - func Test_sipToFraction(t *testing.T) { tests := []struct { name string diff --git a/internal/server/postgres/sql/migrations/00013_split_rebuild_swap.sql b/internal/server/postgres/sql/migrations/00013_split_rebuild_swap.sql index 97c2b53..ba314eb 100644 --- a/internal/server/postgres/sql/migrations/00013_split_rebuild_swap.sql +++ b/internal/server/postgres/sql/migrations/00013_split_rebuild_swap.sql @@ -3,321 +3,313 @@ -- +goose StatementBegin /* * Splits pred.rebuild_forecast_partition into a build phase and a swap phase. - * - * 00012 did both in one transaction, and deadlocked against the live read path: - * - * the reader held AccessShare on pred.forecasters - the read queries resolve the forecaster - * id before they touch pred.forecasts - and waited for AccessShare on pred.forecasts; - * the rebuild held AccessExclusive on pred.forecasts from its DETACH, and waited for - * AccessExclusive on pred.forecasters, which DROP TABLE needs in order to remove the - * partition's foreign key triggers from the referenced side. - * - * DROP TABLE was the only statement in the swap that reached beyond pred.forecasts and its - * values sibling: pred.forecasts references pred.forecasters, loc.geometries and - * loc.source_types, and dropping a referencing table takes AccessExclusive on all three. Every - * other step touches those tables at ShareRowExclusive at most (ATTACH clones the parent's - * foreign keys onto the new partition), which readers never block. So the old partition is now - * renamed to _retired rather than dropped - matching how the values partition is already - * treated - and dropped out of band once the rebuild has been verified. - * - * The swap also takes the locks it needs up front, pred.forecasts before - * pred.predicted_generation_values, which is the order every reader and writer takes them in. - * Holding nothing else at that point, a conflict is a lock_timeout on the first statement rather - * than a cycle discovered twelve minutes in. - * - * Splitting the phases is what makes losing that race cheap: the staging fill and the ordered - * insert are committed before the swap begins, so a swap can be retried on its own. - * - * CALL pred.build_forecast_partition('forecasts_p20260803'); - * CALL pred.swap_forecast_partition('forecasts_p20260803'); */ + CREATE OR REPLACE PROCEDURE pred.build_forecast_partition( - p_partition TEXT, - p_chunk INTERVAL DEFAULT INTERVAL '1 hour', - p_work_mem TEXT DEFAULT '256MB' -) -LANGUAGE plpgsql AS $$ -DECLARE - v_values TEXT; - v_new TEXT := p_partition || '_v2'; - v_bounds TEXT; - v_lo TIMESTAMP; - v_hi TIMESTAMP; - v_t TIMESTAMP; - v_n BIGINT; - v_total BIGINT := 0; - v_all BIGINT; - v_src BIGINT; - v_dst BIGINT; - v_started TIMESTAMPTZ := clock_timestamp(); -BEGIN - SELECT pg_get_expr(c.relpartbound, c.oid) INTO v_bounds - FROM pg_class AS c INNER JOIN pg_namespace AS n ON n.oid = c.relnamespace - WHERE n.nspname = 'pred' AND c.relname = p_partition; - - IF v_bounds IS NULL THEN - RAISE EXCEPTION 'not an attached partition of pred.forecasts: pred.%', p_partition; - END IF; - - /* pg_partman names siblings _p, so the values partition covering the same - * uuid range differs only in its prefix. */ - v_values := 'predicted_generation_values_' || substring(p_partition FROM '^forecasts_(p.+)$'); - - IF v_values IS NULL OR to_regclass('pred.' || quote_ident(v_values)) IS NULL THEN - RAISE EXCEPTION 'no values partition matching pred.%: expected pred.%', - p_partition, v_values; - END IF; - - /* A rebuilt table left by a run whose swap failed holds verified rows: swapping it in is - * cheaper and safer than rebuilding it, so refuse rather than silently starting over. */ - IF to_regclass('pred.' || quote_ident(v_new)) IS NOT NULL THEN - RAISE EXCEPTION 'pred.% already exists', - v_new - USING HINT = format( - 'CALL pred.swap_forecast_partition(%L) to finish that run, or DROP TABLE pred.%I to start over', - p_partition, v_new); - END IF; - - v_lo := partman.uuid7_time_decoder( - (regexp_match(v_bounds, $re$FROM \('([^']+)'\)$re$))[1]::UUID::TEXT) AT TIME ZONE 'UTC'; - v_hi := partman.uuid7_time_decoder( - (regexp_match(v_bounds, $re$TO \('([^']+)'\)$re$))[1]::UUID::TEXT) AT TIME ZONE 'UTC'; - - RAISE NOTICE 'pred.% covers % .. % (% chunks of %)', - p_partition, v_lo, v_hi, - CEIL(EXTRACT(EPOCH FROM (v_hi - v_lo)) / EXTRACT(EPOCH FROM p_chunk)), p_chunk; - - CREATE UNLOGGED TABLE IF NOT EXISTS pred.fc_staging ( - forecast_uuid UUID PRIMARY KEY, - p02_sips SMALLINT [], p10_sips SMALLINT [], p25_sips SMALLINT [], - p50_sips SMALLINT [], p75_sips SMALLINT [], p90_sips SMALLINT [], - p98_sips SMALLINT [] - ); - - /* An earlier run that failed after filling would otherwise leave rows behind that the - * ON CONFLICT DO NOTHING below would silently keep. */ - TRUNCATE pred.fc_staging; - - v_t := v_lo; - WHILE v_t < v_hi LOOP - /* The CASE WHEN bool_or(...) guard keeps an unused p-level as a NULL array rather than a - * materialised array of nulls: ~3 bytes per forecast against ~130. */ - EXECUTE format($q$ - INSERT INTO pred.fc_staging ( - forecast_uuid, p02_sips, p10_sips, p25_sips, - p50_sips, p75_sips, p90_sips, p98_sips) - SELECT - forecast_uuid, - CASE WHEN bool_or(p02_sip IS NOT NULL) THEN array_agg(p02_sip ORDER BY horizon_mins) END, - CASE WHEN bool_or(p10_sip IS NOT NULL) THEN array_agg(p10_sip ORDER BY horizon_mins) END, - CASE WHEN bool_or(p25_sip IS NOT NULL) THEN array_agg(p25_sip ORDER BY horizon_mins) END, - array_agg(p50_sip ORDER BY horizon_mins), - CASE WHEN bool_or(p75_sip IS NOT NULL) THEN array_agg(p75_sip ORDER BY horizon_mins) END, - CASE WHEN bool_or(p90_sip IS NOT NULL) THEN array_agg(p90_sip ORDER BY horizon_mins) END, - CASE WHEN bool_or(p98_sip IS NOT NULL) THEN array_agg(p98_sip ORDER BY horizon_mins) END - FROM pred.%I - WHERE forecast_uuid >= uuidv7_boundary(%L::TIMESTAMP AT TIME ZONE 'UTC') - AND forecast_uuid < uuidv7_boundary(%L::TIMESTAMP AT TIME ZONE 'UTC') - GROUP BY forecast_uuid - ON CONFLICT (forecast_uuid) DO NOTHING - $q$, v_values, v_t, v_t + p_chunk); - - GET DIAGNOSTICS v_n = ROW_COUNT; - v_total := v_total + v_n; - - COMMIT; - - RAISE NOTICE '% .. % +% (total %, elapsed %)', - v_t, v_t + p_chunk, v_n, v_total, clock_timestamp() - v_started; - - v_t := v_t + p_chunk; - END LOOP; - - RAISE NOTICE 'staged % forecasts in %', v_total, clock_timestamp() - v_started; - - EXECUTE format('SET LOCAL work_mem = %L', p_work_mem); - - /* Ordering by (geometry_uuid, source_type_id, forecaster_id, forecast_uuid DESC) makes one - * location's forecasts contiguous and matches idx_forecasts_filter, so an index scan walks - * the heap in physical order. Every hot-path query filters on geometry_uuid first, so - * nothing loses. StreamForecastData is the only broad scan and is explicitly rare. - * - * Forecasts with no rows in the values partition are dropped: the INNER JOIN against staging - * excludes them, and the row count check below is written to expect that. */ - EXECUTE format('CREATE TABLE pred.%I (LIKE pred.forecasts INCLUDING ALL)', v_new); - - EXECUTE format($q$ - INSERT INTO pred.%I ( - forecast_uuid, geometry_uuid, source_type_id, forecaster_id, init_time_utc, - value_resolution_mins, target_period, metadata, created_at_utc, - p02_sips, p10_sips, p25_sips, p50_sips, p75_sips, p90_sips, p98_sips - ) - SELECT f.forecast_uuid, f.geometry_uuid, f.source_type_id, f.forecaster_id, f.init_time_utc, - f.value_resolution_mins, f.target_period, f.metadata, f.created_at_utc, - s.p02_sips, s.p10_sips, s.p25_sips, s.p50_sips, s.p75_sips, s.p90_sips, s.p98_sips - FROM pred.%I AS f - INNER JOIN pred.fc_staging AS s USING (forecast_uuid) - ORDER BY f.geometry_uuid, f.source_type_id, f.forecaster_id, f.forecast_uuid DESC - $q$, v_new, p_partition); - - EXECUTE format('SELECT count(*) FROM pred.%I', p_partition) INTO v_all; - EXECUTE format( - 'SELECT count(*) FROM pred.%I AS f - WHERE EXISTS (SELECT 1 FROM pred.%I AS v WHERE v.forecast_uuid = f.forecast_uuid)', - p_partition, v_values) INTO v_src; - EXECUTE format('SELECT count(*) FROM pred.%I', v_new) INTO v_dst; - - IF v_src <> v_dst THEN - RAISE EXCEPTION 'row count mismatch for %: % source forecasts with values -> % rebuilt rows', - p_partition, v_src, v_dst; - END IF; - - /* Marks the partition as migrated. Note this does not buy constraint exclusion on the read - * queries' legacy branch: they filter p50_sips inside a CTE rather than on a direct scan of - * pred.forecasts, so the planner cannot use it to prune. It is an integrity check and an - * operational marker for which partitions are done. Added before ATTACH so the fresh, - * exclusively-locked table is scanned rather than a live partition. */ - EXECUTE format( - 'ALTER TABLE pred.%I ADD CONSTRAINT p50_sips_not_null CHECK (p50_sips IS NOT NULL)', v_new); - - EXECUTE format('ANALYZE pred.%I', v_new); - - DROP TABLE pred.fc_staging; - - RAISE NOTICE 'built pred.%: % rows (% forecasts had no values and were dropped) in %', - v_new, v_dst, v_all - v_dst, clock_timestamp() - v_started; - RAISE NOTICE 'nothing is swapped yet: CALL pred.swap_forecast_partition(''%'')', p_partition; -END; -$$; + p_partition TEXT, + p_chunk INTERVAL DEFAULT INTERVAL '1 hour', + p_work_mem TEXT DEFAULT '256MB', + p_batch_rows BIGINT DEFAULT 50000 + ) + LANGUAGE plpgsql AS $$ + DECLARE + v_values TEXT; + v_new TEXT := p_partition || '_v2'; + v_bounds TEXT; + v_default BOOLEAN; + v_lo TIMESTAMP; + v_hi TIMESTAMP; + v_t TIMESTAMP; + v_last_uuid UUID; + v_next_uuid UUID; + v_n BIGINT; + v_total BIGINT := 0; + v_all BIGINT; + v_src BIGINT; + v_dst BIGINT; + v_started TIMESTAMPTZ := clock_timestamp(); + BEGIN + SELECT pg_get_expr(c.relpartbound, c.oid) INTO v_bounds + FROM pg_class AS c INNER JOIN pg_namespace AS n ON n.oid = c.relnamespace + WHERE n.nspname = 'pred' AND c.relname = p_partition; + + IF v_bounds IS NULL THEN + RAISE EXCEPTION 'not an attached partition of pred.forecasts: pred.%', p_partition; + END IF; + + v_default := (v_bounds = 'DEFAULT'); + + /* Sibling naming is the same prefix swap for every partition, including the default one: + * forecasts_pXXXXXXXX -> predicted_generation_values_pXXXXXXXX, + * forecasts_default -> predicted_generation_values_default. */ + v_values := 'predicted_generation_values_' || substring(p_partition FROM '^forecasts_(.+)$'); + + IF v_values IS NULL OR to_regclass('pred.' || quote_ident(v_values)) IS NULL THEN + RAISE EXCEPTION 'no values partition matching pred.%: expected pred.%', + p_partition, v_values; + END IF; + + IF to_regclass('pred.' || quote_ident(v_new)) IS NOT NULL THEN + RAISE EXCEPTION 'pred.% already exists', + v_new + USING HINT = format( + 'CALL pred.swap_forecast_partition(%L) to finish that run, or DROP TABLE pred.%I to start over', + p_partition, v_new); + END IF; + + CREATE UNLOGGED TABLE IF NOT EXISTS pred.fc_staging ( + forecast_uuid UUID PRIMARY KEY, + p02_sips SMALLINT [], p10_sips SMALLINT [], p25_sips SMALLINT [], + p50_sips SMALLINT [], p75_sips SMALLINT [], p90_sips SMALLINT [], + p98_sips SMALLINT [] + ); + TRUNCATE pred.fc_staging; + + IF v_default THEN + /* No time bounds to chunk over - forecast_uuid in this partition can be anything that + * fell outside every managed weekly range. Chunk by distinct forecast_uuid instead: + * find the batch boundary first, then aggregate up to it, so a forecast's horizon rows + * never straddle two batches. */ + v_last_uuid := '00000000-0000-0000-0000-000000000000'::UUID; + + LOOP + EXECUTE format($q$ + SELECT forecast_uuid FROM ( + SELECT DISTINCT forecast_uuid FROM pred.%I + WHERE forecast_uuid > %L + ORDER BY forecast_uuid + LIMIT %L + ) AS batch ORDER BY forecast_uuid DESC LIMIT 1 + $q$, v_values, v_last_uuid, p_batch_rows) INTO v_next_uuid; + + EXIT WHEN v_next_uuid IS NULL; + + EXECUTE format($q$ + INSERT INTO pred.fc_staging ( + forecast_uuid, p02_sips, p10_sips, p25_sips, + p50_sips, p75_sips, p90_sips, p98_sips) + SELECT + forecast_uuid, + CASE WHEN bool_or(p02_sip IS NOT NULL) THEN array_agg(p02_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p10_sip IS NOT NULL) THEN array_agg(p10_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p25_sip IS NOT NULL) THEN array_agg(p25_sip ORDER BY horizon_mins) END, + array_agg(p50_sip ORDER BY horizon_mins), + CASE WHEN bool_or(p75_sip IS NOT NULL) THEN array_agg(p75_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p90_sip IS NOT NULL) THEN array_agg(p90_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p98_sip IS NOT NULL) THEN array_agg(p98_sip ORDER BY horizon_mins) END + FROM pred.%I + WHERE forecast_uuid > %L AND forecast_uuid <= %L + GROUP BY forecast_uuid + ON CONFLICT (forecast_uuid) DO NOTHING + $q$, v_values, v_last_uuid, v_next_uuid); + + GET DIAGNOSTICS v_n = ROW_COUNT; + v_total := v_total + v_n; + v_last_uuid := v_next_uuid; + + COMMIT; + + RAISE NOTICE '... % +% (total %, elapsed %)', + v_last_uuid, v_n, v_total, clock_timestamp() - v_started; + END LOOP; + ELSE + v_lo := partman.uuid7_time_decoder( + (regexp_match(v_bounds, $re$FROM \('([^']+)'\)$re$))[1]::UUID::TEXT) AT TIME ZONE 'UTC'; + v_hi := partman.uuid7_time_decoder( + (regexp_match(v_bounds, $re$TO \('([^']+)'\)$re$))[1]::UUID::TEXT) AT TIME ZONE 'UTC'; + + RAISE NOTICE 'pred.% covers % .. % (% chunks of %)', + p_partition, v_lo, v_hi, + CEIL(EXTRACT(EPOCH FROM (v_hi - v_lo)) / EXTRACT(EPOCH FROM p_chunk)), p_chunk; + + v_t := v_lo; + WHILE v_t < v_hi LOOP + EXECUTE format($q$ + INSERT INTO pred.fc_staging ( + forecast_uuid, p02_sips, p10_sips, p25_sips, + p50_sips, p75_sips, p90_sips, p98_sips) + SELECT + forecast_uuid, + CASE WHEN bool_or(p02_sip IS NOT NULL) THEN array_agg(p02_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p10_sip IS NOT NULL) THEN array_agg(p10_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p25_sip IS NOT NULL) THEN array_agg(p25_sip ORDER BY horizon_mins) END, + array_agg(p50_sip ORDER BY horizon_mins), + CASE WHEN bool_or(p75_sip IS NOT NULL) THEN array_agg(p75_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p90_sip IS NOT NULL) THEN array_agg(p90_sip ORDER BY horizon_mins) END, + CASE WHEN bool_or(p98_sip IS NOT NULL) THEN array_agg(p98_sip ORDER BY horizon_mins) END + FROM pred.%I + WHERE forecast_uuid >= uuidv7_boundary(%L::TIMESTAMP AT TIME ZONE 'UTC') + AND forecast_uuid < uuidv7_boundary(%L::TIMESTAMP AT TIME ZONE 'UTC') + GROUP BY forecast_uuid + ON CONFLICT (forecast_uuid) DO NOTHING + $q$, v_values, v_t, v_t + p_chunk); + + GET DIAGNOSTICS v_n = ROW_COUNT; + v_total := v_total + v_n; + + COMMIT; + + RAISE NOTICE '% .. % +% (total %, elapsed %)', + v_t, v_t + p_chunk, v_n, v_total, clock_timestamp() - v_started; + + v_t := v_t + p_chunk; + END LOOP; + END IF; + + RAISE NOTICE 'staged % forecasts in %', v_total, clock_timestamp() - v_started; + + EXECUTE format('SET LOCAL work_mem = %L', p_work_mem); + + EXECUTE format('CREATE TABLE pred.%I (LIKE pred.forecasts INCLUDING ALL)', v_new); + + /* LEFT JOIN rather than INNER, with COALESCE onto the source row's own arrays: a forecast + * that already has arrays (written after the array-write deploy landed) has no row in + * fc_staging and must not be dropped. Only a forecast with neither a staging match nor its + * own arrays is truly value-less and gets excluded by the WHERE below. */ + EXECUTE format($q$ + INSERT INTO pred.%I ( + forecast_uuid, geometry_uuid, source_type_id, forecaster_id, init_time_utc, + value_resolution_mins, target_period, metadata, created_at_utc, + p02_sips, p10_sips, p25_sips, p50_sips, p75_sips, p90_sips, p98_sips + ) + SELECT f.forecast_uuid, f.geometry_uuid, f.source_type_id, f.forecaster_id, f.init_time_utc, + f.value_resolution_mins, f.target_period, f.metadata, f.created_at_utc, + COALESCE(s.p02_sips, f.p02_sips), COALESCE(s.p10_sips, f.p10_sips), + COALESCE(s.p25_sips, f.p25_sips), COALESCE(s.p50_sips, f.p50_sips), + COALESCE(s.p75_sips, f.p75_sips), COALESCE(s.p90_sips, f.p90_sips), + COALESCE(s.p98_sips, f.p98_sips) + FROM pred.%I AS f + LEFT JOIN pred.fc_staging AS s USING (forecast_uuid) + WHERE s.forecast_uuid IS NOT NULL OR f.p50_sips IS NOT NULL + ORDER BY f.geometry_uuid, f.source_type_id, f.forecaster_id, f.forecast_uuid DESC + $q$, v_new, p_partition); + + EXECUTE format('SELECT count(*) FROM pred.%I', p_partition) INTO v_all; + EXECUTE format( + 'SELECT count(*) FROM pred.%I AS f + WHERE f.p50_sips IS NOT NULL + OR EXISTS (SELECT 1 FROM pred.%I AS v WHERE v.forecast_uuid = f.forecast_uuid)', + p_partition, v_values) INTO v_src; + EXECUTE format('SELECT count(*) FROM pred.%I', v_new) INTO v_dst; + + IF v_src <> v_dst THEN + RAISE EXCEPTION 'row count mismatch for %: % source forecasts with values -> % rebuilt rows', + p_partition, v_src, v_dst; + END IF; + + EXECUTE format( + 'ALTER TABLE pred.%I ADD CONSTRAINT p50_sips_not_null CHECK (p50_sips IS NOT NULL)', v_new); + + EXECUTE format('ANALYZE pred.%I', v_new); + + DROP TABLE pred.fc_staging; + + RAISE NOTICE 'built pred.%: % rows (% forecasts had no values and were dropped) in %', + v_new, v_dst, v_all - v_dst, clock_timestamp() - v_started; + RAISE NOTICE 'nothing is swapped yet: CALL pred.swap_forecast_partition(''%'')', p_partition; + END; + $$; -- +goose StatementEnd -- +goose StatementBegin /* * Swaps a table built by pred.build_forecast_partition in for its partition. - * - * Metadata only, so this is seconds rather than minutes - the one exception is ATTACH, which - * validates the partition bound and clones the parent's foreign keys onto the new partition. - * - * pred.predicted_generation_values carries a foreign key to pred.forecasts, and PostgreSQL - * refuses to detach a partition that is still referenced, so the matching values partition is - * detached first. Detaching leaves a standalone copy of the foreign key behind on it, which - * would still block the forecasts detach, so that constraint is dropped too - it takes - * AccessExclusive on pred.forecasts, which is already held. - * - * Both old tables are renamed rather than dropped. Dropping them takes AccessExclusive on every - * table they reference (pred.forecasters, loc.geometries, loc.source_types), which is what - * deadlocked against the read path in 00012, and it is the point of no return for the week. - * Drop them out of band once the rebuild is verified. - * - * CALL pred.swap_forecast_partition('forecasts_p20260803') */ CREATE OR REPLACE PROCEDURE pred.swap_forecast_partition( - p_partition TEXT, - p_lock_timeout TEXT DEFAULT '5s' -) -LANGUAGE plpgsql AS $$ -DECLARE - v_values TEXT; - v_new TEXT := p_partition || '_v2'; - v_old TEXT := p_partition || '_retired'; - v_bounds TEXT; - v_fk TEXT; - v_names JSONB; - v_target TEXT; - r RECORD; - v_started TIMESTAMPTZ := clock_timestamp(); -BEGIN - SELECT pg_get_expr(c.relpartbound, c.oid) INTO v_bounds - FROM pg_class AS c INNER JOIN pg_namespace AS n ON n.oid = c.relnamespace - WHERE n.nspname = 'pred' AND c.relname = p_partition; - - IF v_bounds IS NULL THEN - RAISE EXCEPTION 'not an attached partition of pred.forecasts: pred.%', p_partition; - END IF; - - IF to_regclass('pred.' || quote_ident(v_new)) IS NULL THEN - RAISE EXCEPTION 'no rebuilt table pred.%', v_new - USING HINT = format('CALL pred.build_forecast_partition(%L) first', p_partition); - END IF; - - v_values := 'predicted_generation_values_' || substring(p_partition FROM '^forecasts_(p.+)$'); - - IF v_values IS NULL OR to_regclass('pred.' || quote_ident(v_values)) IS NULL THEN - RAISE EXCEPTION 'no values partition matching pred.%: expected pred.%', - p_partition, v_values; - END IF; - - /* Everything below needs these two, so take them now, while nothing else is held, in the - * order the read and write paths take them: forecasts before predicted_generation_values. - * A conflict then fails here, before any DDL has run, and the call can simply be repeated. - * Both are partitioned parents and LOCK recurses, so this covers the partitions too. */ - EXECUTE format('SET LOCAL lock_timeout = %L', p_lock_timeout); - LOCK TABLE pred.forecasts IN ACCESS EXCLUSIVE MODE; - LOCK TABLE pred.predicted_generation_values IN ACCESS EXCLUSIVE MODE; - - EXECUTE format( - 'ALTER TABLE pred.predicted_generation_values DETACH PARTITION pred.%I', v_values); - - SELECT conname INTO v_fk - FROM pg_constraint - WHERE conrelid = ('pred.' || quote_ident(v_values))::regclass - AND contype = 'f' - AND confrelid = 'pred.forecasts'::regclass; - - IF v_fk IS NOT NULL THEN - EXECUTE format('ALTER TABLE pred.%I DROP CONSTRAINT %I', v_values, v_fk); - END IF; - - EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', v_values, v_values || '_retired'); - - EXECUTE format('ALTER TABLE pred.forecasts DETACH PARTITION pred.%I', p_partition); - EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', p_partition, v_old); - EXECUTE format('ALTER TABLE pred.forecasts ATTACH PARTITION pred.%I %s', v_new, v_bounds); - EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', v_new, p_partition); - - /* Renaming a table does not rename its indexes, so the swapped-in partition would keep the - * _v2 names it was built under - and the retired table would keep the names it should have. - * Move the retired ones aside, then name each new index after the one it replaces. Matching - * is on the index definition rather than the name because PostgreSQL truncates names to 63 - * bytes, which bites at a different point for the longer _v2 table name. Renaming an index - * renames the constraint behind it, so primary keys are covered by the same loop. */ - SELECT jsonb_object_agg(s.definition, s.index_name) INTO v_names - FROM ( - SELECT regexp_replace(pg_get_indexdef(i.indexrelid), - '^CREATE (UNIQUE )?INDEX \S+ ON \S+ ', '') AS definition, - c.relname AS index_name - FROM pg_index AS i INNER JOIN pg_class AS c ON c.oid = i.indexrelid - WHERE i.indrelid = ('pred.' || quote_ident(v_old))::REGCLASS - ) AS s; - - FOR r IN - SELECT c.oid, c.relname - FROM pg_index AS i INNER JOIN pg_class AS c ON c.oid = i.indexrelid - WHERE i.indrelid = ('pred.' || quote_ident(v_old))::REGCLASS - LOOP - EXECUTE format('ALTER INDEX pred.%I RENAME TO %I', r.relname, 'retired_' || r.oid); - END LOOP; - - FOR r IN - SELECT c.relname, - regexp_replace(pg_get_indexdef(i.indexrelid), - '^CREATE (UNIQUE )?INDEX \S+ ON \S+ ', '') AS definition - FROM pg_index AS i INNER JOIN pg_class AS c ON c.oid = i.indexrelid - WHERE i.indrelid = ('pred.' || quote_ident(p_partition))::REGCLASS - LOOP - /* An index with no counterpart on the retired table - the parent's index set has moved - * on since that partition was made - just loses the _v2 from the name it was built with. */ - v_target := COALESCE(v_names ->> r.definition, replace(r.relname, v_new, p_partition)); - - IF v_target <> r.relname THEN - EXECUTE format('ALTER INDEX pred.%I RENAME TO %I', r.relname, v_target); - END IF; - END LOOP; - - RAISE NOTICE 'swapped % in % (old rows retained as pred.%_retired and pred.%_retired)', - p_partition, clock_timestamp() - v_started, p_partition, v_values; -END; -$$; + p_partition TEXT, + p_lock_timeout TEXT DEFAULT '5s' + ) + LANGUAGE plpgsql AS $$ + DECLARE + v_values TEXT; + v_new TEXT := p_partition || '_v2'; + v_old TEXT := p_partition || '_retired'; + v_bounds TEXT; + v_fk TEXT; + v_names JSONB; + v_target TEXT; + r RECORD; + v_started TIMESTAMPTZ := clock_timestamp(); + BEGIN + SELECT pg_get_expr(c.relpartbound, c.oid) INTO v_bounds + FROM pg_class AS c INNER JOIN pg_namespace AS n ON n.oid = c.relnamespace + WHERE n.nspname = 'pred' AND c.relname = p_partition; + + IF v_bounds IS NULL THEN + RAISE EXCEPTION 'not an attached partition of pred.forecasts: pred.%', p_partition; + END IF; + + IF to_regclass('pred.' || quote_ident(v_new)) IS NULL THEN + RAISE EXCEPTION 'no rebuilt table pred.%', v_new + USING HINT = format('CALL pred.build_forecast_partition(%L) first', p_partition); + END IF; + + v_values := 'predicted_generation_values_' || substring(p_partition FROM '^forecasts_(.+)$'); + + IF v_values IS NULL OR to_regclass('pred.' || quote_ident(v_values)) IS NULL THEN + RAISE EXCEPTION 'no values partition matching pred.%: expected pred.%', + p_partition, v_values; + END IF; + + EXECUTE format('SET LOCAL lock_timeout = %L', p_lock_timeout); + LOCK TABLE pred.forecasts IN ACCESS EXCLUSIVE MODE; + LOCK TABLE pred.predicted_generation_values IN ACCESS EXCLUSIVE MODE; + + EXECUTE format( + 'ALTER TABLE pred.predicted_generation_values DETACH PARTITION pred.%I', v_values); + + SELECT conname INTO v_fk + FROM pg_constraint + WHERE conrelid = ('pred.' || quote_ident(v_values))::regclass + AND contype = 'f' + AND confrelid = 'pred.forecasts'::regclass; + + IF v_fk IS NOT NULL THEN + EXECUTE format('ALTER TABLE pred.%I DROP CONSTRAINT %I', v_values, v_fk); + END IF; + + EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', v_values, v_values || '_retired'); + + EXECUTE format('ALTER TABLE pred.forecasts DETACH PARTITION pred.%I', p_partition); + EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', p_partition, v_old); + EXECUTE format('ALTER TABLE pred.forecasts ATTACH PARTITION pred.%I %s', v_new, v_bounds); + EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', v_new, p_partition); + + SELECT jsonb_object_agg(s.definition, s.index_name) INTO v_names + FROM ( + SELECT regexp_replace(pg_get_indexdef(i.indexrelid), + '^CREATE (UNIQUE )?INDEX \S+ ON \S+ ', '') AS definition, + c.relname AS index_name + FROM pg_index AS i INNER JOIN pg_class AS c ON c.oid = i.indexrelid + WHERE i.indrelid = ('pred.' || quote_ident(v_old))::REGCLASS + ) AS s; + + FOR r IN + SELECT c.oid, c.relname + FROM pg_index AS i INNER JOIN pg_class AS c ON c.oid = i.indexrelid + WHERE i.indrelid = ('pred.' || quote_ident(v_old))::REGCLASS + LOOP + EXECUTE format('ALTER INDEX pred.%I RENAME TO %I', r.relname, 'retired_' || r.oid); + END LOOP; + + FOR r IN + SELECT c.relname, + regexp_replace(pg_get_indexdef(i.indexrelid), + '^CREATE (UNIQUE )?INDEX \S+ ON \S+ ', '') AS definition + FROM pg_index AS i INNER JOIN pg_class AS c ON c.oid = i.indexrelid + WHERE i.indrelid = ('pred.' || quote_ident(p_partition))::REGCLASS + LOOP + v_target := COALESCE(v_names ->> r.definition, replace(r.relname, v_new, p_partition)); + + IF v_target <> r.relname THEN + EXECUTE format('ALTER INDEX pred.%I RENAME TO %I', r.relname, v_target); + END IF; + END LOOP; + + RAISE NOTICE 'swapped % in % (old rows retained as pred.%_retired and pred.%_retired)', + p_partition, clock_timestamp() - v_started, p_partition, v_values; + END; + $$; -- +goose StatementEnd -- +goose StatementBegin diff --git a/internal/server/postgres/sql/migrations/00014_drop_row_storage.sql b/internal/server/postgres/sql/migrations/00014_drop_row_storage.sql new file mode 100644 index 0000000..58d381a --- /dev/null +++ b/internal/server/postgres/sql/migrations/00014_drop_row_storage.sql @@ -0,0 +1,201 @@ +-- +goose Up + +/* + * Consolidates the array storage migration (00011-00013). + * + * Makes p50_sips NOT NULL, drops pred.predicted_generation_values and the machinery that + * rebuilt partitions into it, and leaves pred.forecasts as the single source of predicted + * values. + * + * This is not reversible. The Down below restores the schema so that a code revert works, + * but the row data is gone. + */ + +-- +goose StatementBegin +DO $$ +DECLARE + v_unmigrated BIGINT; +BEGIN + SELECT count(*) INTO v_unmigrated FROM pred.forecasts WHERE p50_sips IS NULL; + + IF v_unmigrated > 0 THEN + RAISE EXCEPTION 'refusing to drop row storage: % forecasts still have no arrays', + v_unmigrated + USING HINT = 'find them with: SELECT tableoid::REGCLASS, count(*) FROM pred.forecasts ' + 'WHERE p50_sips IS NULL GROUP BY 1'; + END IF; +END $$; +-- +goose StatementEnd + +/* Every partition rebuilt by 00013 already carries a validated + * CHECK (p50_sips IS NOT NULL) named p50_sips_not_null, which lets SET NOT NULL skip its + * scan. Partitions pg_partman created after the array deploy carry no such constraint - + * they are array-native by construction but unproven, so prove them here. Adding NOT VALID + * and validating separately keeps the scan under SHARE UPDATE EXCLUSIVE instead of + * ACCESS EXCLUSIVE. */ +-- +goose StatementBegin +DO $$ +DECLARE + r RECORD; +BEGIN + FOR r IN + SELECT c.relname + FROM pg_class AS c + INNER JOIN pg_inherits AS i ON i.inhrelid = c.oid + INNER JOIN pg_class AS p ON p.oid = i.inhparent + WHERE p.relname = 'forecasts' + AND NOT EXISTS ( + SELECT 1 FROM pg_constraint AS k + WHERE k.conrelid = c.oid + AND k.conname = 'p50_sips_not_null' + AND k.convalidated + ) + LOOP + RAISE NOTICE 'proving p50_sips on pred.%', r.relname; + EXECUTE format( + 'ALTER TABLE pred.%I ADD CONSTRAINT p50_sips_not_null ' + 'CHECK (p50_sips IS NOT NULL) NOT VALID', r.relname); + EXECUTE format( + 'ALTER TABLE pred.%I VALIDATE CONSTRAINT p50_sips_not_null', r.relname); + END LOOP; +END $$; +-- +goose StatementEnd + +ALTER TABLE pred.forecasts ALTER COLUMN p50_sips SET NOT NULL; + +/* The per-partition CHECKs were the marker for which partitions had been rebuilt, and the + * proof that let SET NOT NULL skip its scan. The column constraint now subsumes both. */ +-- +goose StatementBegin +DO $$ +DECLARE + r RECORD; +BEGIN + FOR r IN + SELECT c.relname + FROM pg_class AS c + INNER JOIN pg_inherits AS i ON i.inhrelid = c.oid + INNER JOIN pg_class AS p ON p.oid = i.inhparent + INNER JOIN pg_constraint AS k + ON k.conrelid = c.oid AND k.conname = 'p50_sips_not_null' + WHERE p.relname = 'forecasts' + LOOP + EXECUTE format( + 'ALTER TABLE pred.%I DROP CONSTRAINT p50_sips_not_null', r.relname); + END LOOP; +END $$; +-- +goose StatementEnd + +/* p50_sips can no longer be NULL, so the guard is dead. Left NOT VALID, as it has been + * since 00011 - validating it is a full scan and buys nothing the write path does not + * already enforce. */ +ALTER TABLE pred.forecasts DROP CONSTRAINT plevel_lengths_match_check; + +ALTER TABLE pred.forecasts + ADD CONSTRAINT plevel_lengths_match_check CHECK ( + ARRAY_LENGTH(p50_sips, 1) > 0 + AND COALESCE(ARRAY_LENGTH(p02_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p10_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p25_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p75_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p90_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p98_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + ) NOT VALID; + +/* pg_partman config goes before the table, or the next run_maintenance errors on a parent + * that no longer exists. */ +DELETE FROM partman.part_config_sub +WHERE sub_parent = 'pred.predicted_generation_values'; + +DELETE FROM partman.part_config +WHERE parent_table = 'pred.predicted_generation_values'; + +/* Takes ACCESS EXCLUSIVE on pred.forecasters, loc.geometries and loc.source_types to remove + * this table's foreign key triggers from the referenced side - a lock set that can deadlock + * against the read path. Safe only because goose runs with the API down. */ +SET lock_timeout = '30s'; + +DROP TABLE pred.predicted_generation_values; +DROP TABLE IF EXISTS pred.predicted_generation_values_template; +DROP TABLE IF EXISTS partman.template_pred_predicted_generation_values; + +RESET lock_timeout; + +/* Only reference was the other_stats_fractions constraint on the table just dropped. */ +DROP FUNCTION IF EXISTS pred.check_all_jsonb_values_are_valid_stat_fractions(JSONB); + +DROP PROCEDURE IF EXISTS pred.swap_forecast_partition(TEXT, TEXT); +DROP PROCEDURE IF EXISTS pred.build_forecast_partition(TEXT, INTERVAL, TEXT); +DROP TABLE IF EXISTS pred.fc_staging; + +-- +goose Down + +/* + * Restores the schema, not the data. The row storage this migration dropped is gone; the + * arrays on pred.forecasts remain authoritative. This exists so that reverting the binary + * leaves a database whose shape the old code recognises - the legacy read branches will + * simply find no rows, which is the correct answer now that every forecast has arrays. + */ + +ALTER TABLE pred.forecasts ALTER COLUMN p50_sips DROP NOT NULL; + +ALTER TABLE pred.forecasts DROP CONSTRAINT plevel_lengths_match_check; + +ALTER TABLE pred.forecasts + ADD CONSTRAINT plevel_lengths_match_check CHECK ( + p50_sips IS NULL OR ( + ARRAY_LENGTH(p50_sips, 1) > 0 + AND COALESCE(ARRAY_LENGTH(p02_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p10_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p25_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p75_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p90_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + AND COALESCE(ARRAY_LENGTH(p98_sips, 1), ARRAY_LENGTH(p50_sips, 1)) = ARRAY_LENGTH(p50_sips, 1) + ) + ) NOT VALID; + +CREATE TABLE pred.predicted_generation_values ( + horizon_mins SMALLINT NOT NULL, + CONSTRAINT horizon_mins_nonnegative_check CHECK (horizon_mins >= 0), + CONSTRAINT horizon_mins_fiveminutely_check CHECK (horizon_mins % 5 = 0), + p50_sip SMALLINT NOT NULL, + CONSTRAINT p50_sip_nonnegative_check CHECK (p50_sip >= 0), + p10_sip SMALLINT, + CONSTRAINT p10_sip_nonnegative_check CHECK (p10_sip >= 0), + p90_sip SMALLINT, + CONSTRAINT p90_sip_nonnegative_check CHECK (p90_sip >= 0), + p02_sip SMALLINT, + CONSTRAINT p02_sip_nonnegative_check CHECK (p02_sip >= 0), + p25_sip SMALLINT, + CONSTRAINT p25_sip_nonnegative_check CHECK (p25_sip >= 0), + p75_sip SMALLINT, + CONSTRAINT p75_sip_nonnegative_check CHECK (p75_sip >= 0), + p98_sip SMALLINT, + CONSTRAINT p98_sip_nonnegative_check CHECK (p98_sip >= 0), + forecast_uuid UUID NOT NULL + REFERENCES pred.forecasts (forecast_uuid) + ON DELETE CASCADE + ON UPDATE CASCADE, + PRIMARY KEY (forecast_uuid, horizon_mins) +) +PARTITION BY RANGE (forecast_uuid); + +SELECT partman.create_parent( + p_parent_table => 'pred.predicted_generation_values', + p_control => 'forecast_uuid', + p_type => 'range', + p_interval => '1 week', + p_automatic_maintenance => 'on', + p_jobmon => FALSE, + p_time_encoder => 'partman.uuid7_time_encoder', + p_time_decoder => 'partman.uuid7_time_decoder', + p_premake => 7 +); + +UPDATE partman.part_config +SET + retention_keep_table = TRUE, + retention_keep_index = TRUE, + infinite_time_partitions = TRUE +WHERE parent_table = 'pred.predicted_generation_values'; + +SELECT partman.run_maintenance('pred.predicted_generation_values'); diff --git a/internal/server/postgres/sql/queries/predictions.sql b/internal/server/postgres/sql/queries/predictions.sql index 4df56ee..c19d4b8 100644 --- a/internal/server/postgres/sql/queries/predictions.sql +++ b/internal/server/postgres/sql/queries/predictions.sql @@ -101,26 +101,10 @@ WITH forecasts_to_delete AS ( DELETE FROM pred.forecasts WHERE forecast_uuid IN (SELECT forecast_uuid FROM forecasts_to_delete); --- name: CreatePredictedValues :copyfrom -/* CreatePredictedValues inserts predicted generation values using - * postgres COPY protocol, making it the fastest way to perform large inserts of predictions. - * Input p-values are expected as smallint percentages (sip) of capacity, - * with 0 representing 0% and 30000 representing 100% of capacity. - */ -INSERT INTO pred.predicted_generation_values ( - horizon_mins, p50_sip, p10_sip, p90_sip, forecast_uuid, p02_sip, p98_sip, p25_sip, p75_sip -) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9 -); - -- name: ListPredictionsForForecasts :many /* ListPredictionsForForecasts retrieves all predicted generation values for a given location, * source type, and dynamic list of forecasters within a time window. * Note that this does not return ordered results for speed. Ordering is up to the client. - * - * Currently this is two queries in one, as the application in this version's state can have - * values stored either in arrays or in the legacy predicted_generation_values table. - * When everything is migrated to arrays, the second branch can be removed. */ WITH requested_forecasters AS ( SELECT @@ -163,7 +147,7 @@ matched_forecasts AS ( AND f.forecast_uuid >= UUIDV7_BOUNDARY(sqlc.arg(start_timestamp)::TIMESTAMP) AND f.forecast_uuid < UUIDV7_BOUNDARY(sqlc.arg(end_timestamp)::TIMESTAMP + INTERVAL '1 millisecond') ), -expanded_array AS ( +expanded AS ( SELECT mfc.forecaster_name, mfc.forecaster_version, @@ -172,6 +156,7 @@ expanded_array AS ( mfc.geometry_uuid, mfc.source_type_id, mfc.init_time_utc, + o.p50_sip, ( EXTRACT(EPOCH FROM (mfc.first_target_utc - mfc.init_time_utc)) / 60 + (o.ord - 1) * mfc.value_resolution_mins @@ -180,7 +165,6 @@ expanded_array AS ( mins => ((o.ord - 1) * mfc.value_resolution_mins)::INTEGER ))::TIMESTAMP AS target_time_utc, - o.p50_sip, mfc.p02_sips[o.ord] AS p02_sip, mfc.p10_sips[o.ord] AS p10_sip, mfc.p25_sips[o.ord] AS p25_sip, @@ -189,40 +173,9 @@ expanded_array AS ( mfc.p98_sips[o.ord] AS p98_sip FROM matched_forecasts AS mfc CROSS JOIN LATERAL UNNEST(mfc.p50_sips) WITH ORDINALITY AS o (p50_sip, ord) - WHERE mfc.p50_sips IS NOT NULL -), -expanded_legacy AS ( - SELECT - mfc.forecaster_name, - mfc.forecaster_version, - mfc.created_at_utc, - mfc.metadata, - mfc.geometry_uuid, - mfc.source_type_id, - mfc.init_time_utc, - pg.horizon_mins, - (mfc.init_time_utc + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER))::TIMESTAMP - AS target_time_utc, - pg.p50_sip, - pg.p02_sip, - pg.p10_sip, - pg.p25_sip, - pg.p75_sip, - pg.p90_sip, - pg.p98_sip - FROM matched_forecasts AS mfc - INNER JOIN pred.predicted_generation_values AS pg - ON mfc.forecast_uuid = pg.forecast_uuid - AND pg.forecast_uuid >= UUIDV7_BOUNDARY(sqlc.arg(start_timestamp)::TIMESTAMP) - AND pg.forecast_uuid < UUIDV7_BOUNDARY(sqlc.arg(end_timestamp)::TIMESTAMP + INTERVAL '1 millisecond') - WHERE mfc.p50_sips IS NULL -), -expanded AS ( - SELECT * FROM expanded_array - UNION ALL - SELECT * FROM expanded_legacy ) -/* Column order here is load-bearing: StreamForecastData scans these positionally. */ +/* Column order here is load-bearing: StreamForecastData bypasses sqlc and scans these + * positionally, so reordering silently corrupts the stream. */ SELECT e.forecaster_name, e.forecaster_version, @@ -298,10 +251,6 @@ ORDER BY fr.forecaster_name ASC, f.init_time_utc DESC; -- name: ListPredictionsForLocation :many /* ListPredictionsForLocation retrieves all predicted generation values for a given location, * source type, and forecaster within a time window. - * - * Currently this is two queries in one, as the application in this version's state can have - * values stored either in arrays or in the legacy predicted_generation_values table. - * When everything is migrated to arrays, the second branch can be removed. */ WITH allowed_forecasts AS ( SELECT @@ -338,10 +287,15 @@ WITH allowed_forecasts AS ( sqlc.arg(end_timestamp_utc)::TIMESTAMP, '[]' ) ), -sliced AS ( +sliced AS MATERIALIZED ( /* Convert the target-time window and minimum horizon into an array index range. * This means that a forecast with a window that only partially overlaps the requested - * window will only be partially expanded. */ + * window will only be partially expanded. + * + * MATERIALIZED pins lo and hi to one evaluation per forecast. Inlined, the planner + * substitutes their NUMERIC-heavy definitions into every expression downstream that + * derives from them - target_time_utc, horizon_mins, and both sort keys - and recomputes + * them per expanded row rather than per forecast. Measured at ~2x on the read path. */ SELECT af.*, GREATEST(1, CEIL(EXTRACT(EPOCH FROM (GREATEST( @@ -352,11 +306,12 @@ sliced AS ( sqlc.arg(end_timestamp_utc)::TIMESTAMP - af.first_target_utc )) / 60.0 / af.value_resolution_mins)::INTEGER + 1) AS hi FROM allowed_forecasts AS af - WHERE af.p50_sips IS NOT NULL ), -expanded_array AS ( +expanded AS ( /* Expand the sliced arrays into rows, with each row representing a single - * target time and its associated predicted values. */ + * target time and its associated predicted values. + * UNNEST renumbers from 1 regardless of the slice's lower bound, hence the + * s.lo + o.ord - 1 indexing into the sibling p-level arrays. */ SELECT s.forecast_uuid, s.init_time_utc, @@ -364,6 +319,13 @@ expanded_array AS ( s.metadata, s.geometry_uuid, s.source_type_id, + o.p50_sip::SMALLINT AS p50_sip, + s.p02_sips[s.lo + o.ord - 1]::SMALLINT AS p02_sip, + s.p10_sips[s.lo + o.ord - 1]::SMALLINT AS p10_sip, + s.p25_sips[s.lo + o.ord - 1]::SMALLINT AS p25_sip, + s.p75_sips[s.lo + o.ord - 1]::SMALLINT AS p75_sip, + s.p90_sips[s.lo + o.ord - 1]::SMALLINT AS p90_sip, + s.p98_sips[s.lo + o.ord - 1]::SMALLINT AS p98_sip, (s.first_target_utc + MAKE_INTERVAL( mins => ((s.lo + o.ord - 2) * s.value_resolution_mins)::INTEGER @@ -373,65 +335,13 @@ expanded_array AS ( s.first_target_utc + MAKE_INTERVAL(mins => ((s.lo + o.ord - 2) * s.value_resolution_mins)::INTEGER) - s.init_time_utc - )) / 60)::SMALLINT AS horizon_mins, - o.p50_sip::SMALLINT AS p50_sip, - s.p02_sips[s.lo + o.ord - 1]::SMALLINT AS p02_sip, - s.p10_sips[s.lo + o.ord - 1]::SMALLINT AS p10_sip, - s.p25_sips[s.lo + o.ord - 1]::SMALLINT AS p25_sip, - s.p75_sips[s.lo + o.ord - 1]::SMALLINT AS p75_sip, - s.p90_sips[s.lo + o.ord - 1]::SMALLINT AS p90_sip, - s.p98_sips[s.lo + o.ord - 1]::SMALLINT AS p98_sip + )) / 60)::SMALLINT AS horizon_mins FROM sliced AS s CROSS JOIN LATERAL UNNEST(s.p50_sips[s.lo:s.hi]) WITH ORDINALITY AS o (p50_sip, ord) WHERE s.hi >= s.lo ), -expanded_legacy AS ( - /* Forecasts whose partition has not yet been rebuilt into arrays. Column order must match - * expanded_array exactly - UNION ALL matches by position, not by name. */ - SELECT - af.forecast_uuid, - af.init_time_utc, - af.created_at_utc, - af.metadata, - af.geometry_uuid, - af.source_type_id, - (af.init_time_utc + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER))::TIMESTAMP - AS target_time_utc, - pg.horizon_mins, - pg.p50_sip, - pg.p02_sip, - pg.p10_sip, - pg.p25_sip, - pg.p75_sip, - pg.p90_sip, - pg.p98_sip - FROM allowed_forecasts AS af - INNER JOIN pred.predicted_generation_values AS pg - ON af.forecast_uuid = pg.forecast_uuid - /* Repeating the bounds from allowed_forecasts lets the planner prune partitions of - * predicted_generation_values statically. Without them the equijoin alone only prunes - * at runtime, and only if a nested loop is chosen over a hash join. */ - AND pg.forecast_uuid >= UUIDV7_BOUNDARY( - sqlc.arg(start_timestamp_utc)::TIMESTAMP - INTERVAL '3 days' - ) - AND pg.forecast_uuid < UUIDV7_BOUNDARY( - sqlc.arg(end_timestamp_utc)::TIMESTAMP - - MAKE_INTERVAL(mins => sqlc.arg(horizon_mins)::INTEGER) - + INTERVAL '1 millisecond' - ) - WHERE af.p50_sips IS NULL - AND (af.init_time_utc + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER)) - BETWEEN sqlc.arg(start_timestamp_utc)::TIMESTAMP - AND sqlc.arg(end_timestamp_utc)::TIMESTAMP - AND pg.horizon_mins >= sqlc.arg(horizon_mins)::INTEGER -), -expanded AS ( - SELECT * FROM expanded_array - UNION ALL - SELECT * FROM expanded_legacy -), winning_predictions AS ( /* ordering by descending forecast_uuid means the lowest horizons are selected first, * since init times are encoded within it. */ @@ -462,8 +372,12 @@ WHERE sv.sys_period @> wp.target_time_utc ORDER BY wp.target_time_utc ASC; -- name: ListPredictionsAtTimeForLocations :many -/* PostgreSQL returns NULL on an out of bounds array index. As such, ARRAY_LENGTH is used - * to guard against this. +/* ListPredictionsAtTimeForLocations retrieves the latest prediction for a single target time + * across many locations - the map query. + * + * PostgreSQL returns NULL on an out of bounds array index, so ARRAY_LENGTH guards the + * subscript: if target_period and the array length ever disagree, the row is omitted rather + * than silently returning NULL values. */ WITH target_locations AS ( SELECT UNNEST(sqlc.arg(geometry_uuids)::UUID[]) AS geometry_uuid @@ -523,8 +437,7 @@ latest_allowed_forecast_per_location AS ( ), indexed AS ( /* target_time = first_target_utc + (i - 1) * value_resolution_mins, - * so i = (target - first_target) / resolution + 1. Only meaningful when p50_sips is - * populated; the legacy branch below keys on horizon_mins instead. */ + * so i = (target - first_target) / resolution + 1. */ SELECT laf.*, (EXTRACT(EPOCH FROM ( @@ -534,64 +447,62 @@ indexed AS ( sqlc.arg(target_timestamp_utc)::TIMESTAMP - laf.init_time_utc )) / 60)::SMALLINT AS horizon_mins FROM latest_allowed_forecast_per_location AS laf +), +picked AS ( + SELECT + i.forecast_uuid, + i.geometry_uuid, + i.source_type_id, + i.horizon_mins, + i.created_at_utc, + i.init_time_utc, + i.metadata, + i.p02_sips[i.idx] AS p02_sip, + i.p10_sips[i.idx] AS p10_sip, + i.p25_sips[i.idx] AS p25_sip, + i.p50_sips[i.idx] AS p50_sip, + i.p75_sips[i.idx] AS p75_sip, + i.p90_sips[i.idx] AS p90_sip, + i.p98_sips[i.idx] AS p98_sip + FROM indexed AS i + WHERE i.idx BETWEEN 1 AND ARRAY_LENGTH(i.p50_sips, 1) + /* Rejects a target time that falls between array slots. The row model got this for free + * from the join on horizon_mins; without it a misaligned request silently returns a + * neighbouring value instead of no row. */ + AND MOD( + (EXTRACT(EPOCH FROM (sqlc.arg(target_timestamp_utc)::TIMESTAMP - i.first_target_utc)) / 60)::NUMERIC, + i.value_resolution_mins::NUMERIC + ) = 0 ) SELECT - i.forecast_uuid, - i.geometry_uuid, - i.source_type_id, - i.horizon_mins, - i.created_at_utc, - i.init_time_utc, + p.forecast_uuid, + p.geometry_uuid, + p.source_type_id, + p.horizon_mins, + p.created_at_utc, + p.init_time_utc, sv.capacity_watts, sv.latitude, sv.longitude, sv.geometry_name, - i.metadata, - COALESCE(i.p02_sips[i.idx], legacy.p02_sip) AS p02_sip, - COALESCE(i.p10_sips[i.idx], legacy.p10_sip) AS p10_sip, - COALESCE(i.p25_sips[i.idx], legacy.p25_sip) AS p25_sip, - COALESCE(i.p50_sips[i.idx], legacy.p50_sip) AS p50_sip, - COALESCE(i.p75_sips[i.idx], legacy.p75_sip) AS p75_sip, - COALESCE(i.p90_sips[i.idx], legacy.p90_sip) AS p90_sip, - COALESCE(i.p98_sips[i.idx], legacy.p98_sip) AS p98_sip, + p.metadata, + p.p02_sip, + p.p10_sip, + p.p25_sip, + p.p50_sip, + p.p75_sip, + p.p90_sip, + p.p98_sip, sqlc.arg(target_timestamp_utc)::TIMESTAMP AS target_time_utc -FROM indexed AS i +FROM picked AS p INNER JOIN loc.sources_mv AS sv USING (geometry_uuid, source_type_id) - /* The p50_sips IS NULL test sits inside the subquery, not in an ON clause: a LEFT JOIN's ON - * condition filters the result but does not stop the subquery being evaluated, so putting it - * there would probe predicted_generation_values for migrated forecasts too. */ - LEFT OUTER JOIN LATERAL ( - SELECT - pg.p02_sip, - pg.p10_sip, - pg.p25_sip, - pg.p50_sip, - pg.p75_sip, - pg.p90_sip, - pg.p98_sip - FROM pred.predicted_generation_values AS pg - WHERE i.p50_sips IS NULL - AND pg.forecast_uuid = i.forecast_uuid - AND pg.horizon_mins = i.horizon_mins - ) AS legacy ON TRUE -WHERE ( - ( - i.p50_sips IS NOT NULL AND i.idx BETWEEN 1 AND ARRAY_LENGTH(i.p50_sips, 1) - AND MOD( - (EXTRACT(EPOCH FROM (sqlc.arg(target_timestamp_utc)::TIMESTAMP - i.first_target_utc)) / 60)::NUMERIC, - i.value_resolution_mins::NUMERIC - ) - = 0 - ) - OR (i.p50_sips IS NULL AND legacy.p50_sip IS NOT NULL) -) -AND sv.sys_period @> sqlc.arg(target_timestamp_utc)::TIMESTAMP; +WHERE sv.sys_period @> sqlc.arg(target_timestamp_utc)::TIMESTAMP; + -- name: GetWeekAverageDeltasForLocations :many -/* GetWeekAverageDeltasForLocations retrieves the average deltas between predicted and observed generation values - * for a given source type, forecaster, and observer, across a week of forecasts made with the same init time. - * The pivot timestamp is used to determine the week and init time of interest. The results are - * grouped by location and horizon. MATERIALIZED is used because the count assumptions made by postgres on the - * CTEs are unreliable thanks to the UUIDV7_EXTRACT_TIMESTAMP function call. +/* GetWeekAverageDeltasForLocations retrieves the average deltas between predicted and observed + * generation values for a given source type, forecaster, and observer, across a week of + * forecasts made with the same init time. The pivot timestamp determines the week and init time + * of interest. Results are grouped by location and horizon. */ WITH relevant_forecasts AS ( SELECT @@ -611,41 +522,21 @@ WITH relevant_forecasts AS ( AND f.forecast_uuid < UUIDV7_BOUNDARY(sqlc.arg(pivot_timestamp)::TIMESTAMP + INTERVAL '1 millisecond') AND f.init_time_utc::TIME = sqlc.arg(pivot_timestamp)::TIMESTAMP::TIME ), -expanded_array AS ( +relevant_predicted_values AS MATERIALIZED ( SELECT rf.geometry_uuid, rf.source_type_id, + o.p50_sip, ( EXTRACT(EPOCH FROM (rf.first_target_utc - rf.init_time_utc)) / 60 + (o.ord - 1) * rf.value_resolution_mins )::SMALLINT AS horizon_mins, - o.p50_sip, (rf.first_target_utc + MAKE_INTERVAL( mins => ((o.ord - 1) * rf.value_resolution_mins)::INTEGER ))::TIMESTAMP AS target_time_utc FROM relevant_forecasts AS rf CROSS JOIN LATERAL UNNEST(rf.p50_sips) WITH ORDINALITY AS o (p50_sip, ord) - WHERE rf.p50_sips IS NOT NULL -), -expanded_legacy AS ( - SELECT - rf.geometry_uuid, - rf.source_type_id, - pg.horizon_mins, - pg.p50_sip, - (rf.init_time_utc + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER))::TIMESTAMP - AS target_time_utc - FROM relevant_forecasts AS rf - INNER JOIN pred.predicted_generation_values AS pg USING (forecast_uuid) - WHERE rf.p50_sips IS NULL - AND pg.forecast_uuid >= UUIDV7_BOUNDARY(sqlc.arg(pivot_timestamp)::TIMESTAMP - INTERVAL '8 days') - AND pg.forecast_uuid < UUIDV7_BOUNDARY(sqlc.arg(pivot_timestamp)::TIMESTAMP + INTERVAL '1 millisecond') -), -relevant_predicted_values AS MATERIALIZED ( - SELECT * FROM expanded_array - UNION ALL - SELECT * FROM expanded_legacy ), relevant_observations AS MATERIALIZED ( SELECT diff --git a/internal/server/postgres/testdata/seeding.sql b/internal/server/postgres/testdata/seeding.sql index a9d9de2..ac8a5a6 100644 --- a/internal/server/postgres/testdata/seeding.sql +++ b/internal/server/postgres/testdata/seeding.sql @@ -54,33 +54,33 @@ BEGIN FROM unnest(geo_list) AS u(geo_id) CROSS JOIN generate_series(0, (history_window_mins / forecast_freq_mins) - 1) AS s(idx) ORDER BY init_time_utc ASC - ), - inserted_forecasts AS ( - INSERT INTO pred.forecasts - (forecast_uuid, source_type_id, geometry_uuid, forecaster_id, init_time_utc, value_resolution_mins, target_period) - SELECT - UUIDV7(init_time_utc), 1, geo_id, - (SELECT forecaster_id FROM pred.forecasters WHERE forecaster_name = name_prefix || '_forecaster_1'), - init_time_utc, pgv_res_mins::SMALLINT, - TSRANGE(init_time_utc, init_time_utc + (forecast_len_mins * INTERVAL '1 minute')) - FROM generated_data - RETURNING forecast_uuid, init_time_utc ) - INSERT INTO pred.predicted_generation_values - (horizon_mins, p50_sip, p10_sip, p90_sip, forecast_uuid) + INSERT INTO pred.forecasts + (forecast_uuid, source_type_id, geometry_uuid, forecaster_id, init_time_utc, value_resolution_mins, target_period, + p50_sips, p10_sips, p90_sips) SELECT - gs.h, - (random() * 30000)::SMALLINT, - 3000::SMALLINT, - 27000::SMALLINT, - inf.forecast_uuid - FROM inserted_forecasts inf - CROSS JOIN LATERAL generate_series(0, forecast_len_mins - pgv_res_mins, pgv_res_mins) AS gs(h) - ORDER BY inf.init_time_utc ASC; + UUIDV7(gd.init_time_utc), 1, gd.geo_id, + (SELECT forecaster_id FROM pred.forecasters WHERE forecaster_name = name_prefix || '_forecaster_1'), + gd.init_time_utc, pgv_res_mins::SMALLINT, + TSRANGE(gd.init_time_utc, gd.init_time_utc + (forecast_len_mins * INTERVAL '1 minute')), + v.p50_sips, + array_fill(3000::SMALLINT, ARRAY[forecast_len_mins / pgv_res_mins]), + array_fill(27000::SMALLINT, ARRAY[forecast_len_mins / pgv_res_mins]) + FROM generated_data gd + -- Correlated on init_time_utc so each forecast gets its own values; an uncorrelated + -- subquery is an InitPlan, evaluated once, giving every forecast the same array. + CROSS JOIN LATERAL ( + SELECT array_agg((random() * 30000)::SMALLINT) AS p50_sips + FROM generate_series( + gd.init_time_utc, + gd.init_time_utc + ((forecast_len_mins - pgv_res_mins) * INTERVAL '1 minute'), + (pgv_res_mins * INTERVAL '1 minute') + ) + ) AS v; -- Spoof the table size so Postgres uses indexes rather than seq scan in testing - UPDATE pg_class SET reltuples = 346000000, relpages = 5000000 WHERE relname = 'predicted_generation_values'; - UPDATE pg_class SET reltuples = 346000000, relpages = 5000000 WHERE relname = 'predicted_generation_values_pkey'; + UPDATE pg_class SET reltuples = 14400000, relpages = 1600000 WHERE relname = 'forecasts'; + UPDATE pg_class SET reltuples = 14400000, relpages = 1600000 WHERE relname = 'idx_forecasts_filter'; REFRESH MATERIALIZED VIEW loc.sources_mv;