diff --git a/internal/server/postgres/.sqlc.yaml b/internal/server/postgres/.sqlc.yaml index 0f29781..b4cdad4 100644 --- a/internal/server/postgres/.sqlc.yaml +++ b/internal/server/postgres/.sqlc.yaml @@ -16,6 +16,81 @@ sql: emit_interface: true emit_exported_queries: true overrides: + - column: "winning_predictions.p02_sip" + go_type: + type: "int16" + pointer: true + - column: "winning_predictions.p10_sip" + go_type: + type: "int16" + pointer: true + - column: "winning_predictions.p25_sip" + go_type: + type: "int16" + pointer: true + - column: "winning_predictions.p75_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" + go_type: + type: "int16" + pointer: true + - column: "ListPredictionsAtTimeForLocations.p90_sip" + go_type: + type: "int16" + pointer: true + - column: "ListPredictionsAtTimeForLocations.p98_sip" + go_type: + type: "int16" + pointer: true - db_type: "uuid" go_type: import: "github.com/google/uuid" diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index b353afa..5c0cd98 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -64,17 +64,12 @@ func (s *DataPlatformDataServiceServerImpl) CreateForecast( Msg("found source") // Check the forecast values have monotonically increasing horizons - resolution_mins := req.Values[1].HorizonMins - req.Values[0].HorizonMins - for i, value := range req.Values { - if i > 0 { - if resolution_mins != value.HorizonMins-req.Values[i-1].HorizonMins || - value.HorizonMins <= req.Values[i-1].HorizonMins { - return nil, status.Error( - codes.InvalidArgument, - "Forecast horizon values must be monotonically spaced in time.", - ) - } - } + err = validateForecastValues(req.Values) + if err != nil { + return nil, status.Error( + codes.InvalidArgument, + fmt.Sprintf("invalid forecast values: %v", err), + ) } // Check the forecaster exists @@ -372,6 +367,10 @@ func (s *DataPlatformDataServiceServerImpl) StreamForecastData( rows, err := pool.Query( stream.Context(), db.ListPredictionsForForecasts, + fNames, + fVersions, + locationUuid, + int16(req.EnergySource.Number()), pgtype.Timestamp{ Time: req.TimeWindow.StartTimestampUtc.AsTime(), Valid: true, @@ -380,10 +379,6 @@ func (s *DataPlatformDataServiceServerImpl) StreamForecastData( Time: req.TimeWindow.EndTimestampUtc.AsTime(), Valid: true, }, - locationUuid, - int16(req.EnergySource.Number()), - fNames, - fVersions, ) if err != nil { return fmt.Errorf("failed to stream predictions: %w", err) @@ -1564,6 +1559,13 @@ func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts( return fmt.Errorf("error receiving from stream: %w", err) } + if err := validateForecastValues(req.Values); err != nil { + return status.Error( + codes.InvalidArgument, + fmt.Sprintf("invalid forecast values: %v", err), + ) + } + fKey := forecasterKey{ name: req.Forecaster.ForecasterName, version: req.Forecaster.ForecasterVersion, diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index efa7070..a69c26e 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -2033,11 +2033,37 @@ func TestCreateForecast(t *testing.T) { } } + yieldsPartial := make([]*pb.CreateForecastRequest_ForecastValue, 10) + for i := range yieldsPartial { + statFractions := map[string]float32{"p10": 0.1, "p90": 0.9} + // p25 is only populated on some values, making it partial + if i%2 == 0 { + statFractions["p25"] = 0.25 + } + + yieldsPartial[i] = &pb.CreateForecastRequest_ForecastValue{ + HorizonMins: uint32(i * 30), + P50Fraction: 0.5, + OtherStatisticsFractions: statFractions, + } + } + testcases := []struct { name string req *pb.CreateForecastRequest shouldErr bool }{ + { + name: "Shouldn't create forecast with partially populated statistic", + req: &pb.CreateForecastRequest{ + LocationUuid: siteResp.LocationUuid, + Forecaster: fc, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + InitTimeUtc: timestamppb.New(pivotTime), + Values: yieldsPartial, + }, + shouldErr: true, + }, { name: "Should create forecast with populated values", req: &pb.CreateForecastRequest{ @@ -2461,6 +2487,20 @@ func TestStreamCreateForecasts(t *testing.T) { yields := generateTestForecastValues(10, 30) + yieldsPartial := make([]*pb.CreateForecastRequest_ForecastValue, 10) + for i := range yieldsPartial { + statFractions := map[string]float32{"p10": 0.1, "p90": 0.9} + if i%2 == 0 { + statFractions["p25"] = 0.25 + } + + yieldsPartial[i] = &pb.CreateForecastRequest_ForecastValue{ + HorizonMins: uint32(i * 30), + P50Fraction: 0.5, + OtherStatisticsFractions: statFractions, + } + } + testcases := []struct { name string setupStream func(ctx context.Context) (pb.DataPlatformDataService_StreamCreateForecastsClient, error) @@ -2470,6 +2510,25 @@ func TestStreamCreateForecasts(t *testing.T) { expectedErrCode codes.Code expectedUuidsCount int }{ + { + name: "Shouldn't create forecast stream with partially populated statistic", + setupStream: func(ctx context.Context) (pb.DataPlatformDataService_StreamCreateForecastsClient, error) { + return dc.StreamCreateForecasts(ctx) + }, + sendCount: 1, + getReq: func(i int) *pb.CreateForecastRequest { + return &pb.CreateForecastRequest{ + LocationUuid: siteResp.LocationUuid, + Forecaster: fc, + EnergySource: pb.EnergySource_ENERGY_SOURCE_SOLAR, + InitTimeUtc: timestamppb.New(pivotTime.Add(time.Duration(i) * time.Hour)), + Values: yieldsPartial, + } + }, + shouldErr: true, + expectedErrCode: codes.InvalidArgument, + expectedUuidsCount: 0, + }, { name: "Valid stream under limit", setupStream: func(ctx context.Context) (pb.DataPlatformDataService_StreamCreateForecastsClient, error) { diff --git a/internal/server/postgres/mappers.go b/internal/server/postgres/mappers.go index bcf935a..57bcc1a 100644 --- a/internal/server/postgres/mappers.go +++ b/internal/server/postgres/mappers.go @@ -1,6 +1,7 @@ package postgres import ( + "errors" "fmt" "time" @@ -70,11 +71,86 @@ func extractSIPStatPtrFromMap(m map[string]float32, key string) *int16 { 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). +// Callers must have run validateForecastValues first: sqlc maps SMALLINT[] to []int16, which +// cannot express element-level nulls, so partial coverage would silently be written as zeros. +func extractSIPStatSlice(values []*pb.CreateForecastRequest_ForecastValue, key string) []int16 { + out := make([]int16, len(values)) + present := false + + for i, v := range values { + if f, ok := v.OtherStatisticsFractions[key]; ok { + out[i] = int16(f * 30000.0) + present = true + } + } + + if !present { + return nil + } + + return out +} + +// extractP50Slice builds the p50 array. P50 is a top-level field on ForecastValue rather than a +// key in OtherStatisticsFractions, and is always present. +func extractP50Slice(values []*pb.CreateForecastRequest_ForecastValue) []int16 { + out := make([]int16, len(values)) + for i, v := range values { + out[i] = int16(v.P50Fraction * 30000.0) + } + + return out +} + // sipToFraction converts a SIP value to a fraction. func sipToFraction(sip int16) float32 { return float32(sip) / 30000.0 } +// validateForecastValues checks the invariants the array storage layout depends on: +// at least two values, strictly increasing horizons, evenly spaced, and each optional statistic +// either present on every value or on none. +func validateForecastValues(values []*pb.CreateForecastRequest_ForecastValue) error { + if len(values) < 2 { + return errors.New("a forecast must contain at least two values") + } + + resolution := int32(values[1].HorizonMins) - int32(values[0].HorizonMins) + if resolution <= 0 { + return errors.New("forecast horizons must be monotonically increasing") + } + + for i := 1; i < len(values); i++ { + if int32(values[i].HorizonMins)-int32(values[i-1].HorizonMins) != resolution { + return errors.New("forecast horizons must be evenly spaced in time") + } + } + + // SMALLINT[] maps to []int16, which has no way to represent a null element, so a statistic + // supplied for only some horizons would be written as zeros (a valid 0% reading) for the rest. + for _, key := range []string{"p02", "p10", "p25", "p75", "p90", "p98"} { + count := 0 + + for _, v := range values { + if _, ok := v.OtherStatisticsFractions[key]; ok { + count++ + } + } + + if count != 0 && count != len(values) { + return fmt.Errorf( + "statistic '%s' must be present for all values or none, got %d of %d", + key, count, len(values), + ) + } + } + + return nil +} + // buildOtherStatsMap constructs a map of other statistics from optional SIP pointers. // Only keys that are not nil will be included in the returned map. func buildOtherStatsMap(p02, p10, p25, p75, p90, p98 *int16) map[string]float32 { @@ -163,6 +239,13 @@ func mapCreateForecast( TargetPeriod: targetPeriod, Metadata: req.Metadata, CreatedAtUtc: createdTime, + P02Sips: extractSIPStatSlice(req.Values, "p02"), + P10Sips: extractSIPStatSlice(req.Values, "p10"), + P25Sips: extractSIPStatSlice(req.Values, "p25"), + P50Sips: extractP50Slice(req.Values), + P75Sips: extractSIPStatSlice(req.Values, "p75"), + P90Sips: extractSIPStatSlice(req.Values, "p90"), + P98Sips: extractSIPStatSlice(req.Values, "p98"), }, nil } diff --git a/internal/server/postgres/sql/.sqlfluff.toml b/internal/server/postgres/sql/.sqlfluff.toml index d2a2488..b1bb416 100644 --- a/internal/server/postgres/sql/.sqlfluff.toml +++ b/internal/server/postgres/sql/.sqlfluff.toml @@ -4,6 +4,7 @@ max_line_length = 120 rules = aliasing, ambiguous, capitalisation, convention, layout, structure, references exclude_rules = ST07, AL03, AL07, LT08, RF02, RF03 processes = 4 +large_file_skip_byte_limit = 25000 [sqlfluff:indentation] indented_using_on = False diff --git a/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql b/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql new file mode 100644 index 0000000..2aad77c --- /dev/null +++ b/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql @@ -0,0 +1,75 @@ +-- +goose Up +DROP FUNCTION IF EXISTS uuidv7_extract_timestamp(UUID); + +-- +goose StatementBegin +CREATE FUNCTION uuidv7_extract_timestamp(u UUID) RETURNS TIMESTAMP + LANGUAGE sql + IMMUTABLE STRICT PARALLEL SAFE + RETURN uuid_extract_timestamp(u) AT TIME ZONE 'UTC'; +-- +goose StatementEnd + +/* + * Moves predicted values from separate table into arrays. + * + * Array index i (1-based) corresponds to target time: + * target_time = LOWER(target_period) + (i - 1) * value_resolution_mins + * Only works if a forecast has evenly spaced target times. + */ + +ALTER TABLE pred.forecasts + ADD COLUMN p02_sips SMALLINT [], + ADD COLUMN p10_sips SMALLINT [], + ADD COLUMN p25_sips SMALLINT [], + ADD COLUMN p50_sips SMALLINT [], + ADD COLUMN p75_sips SMALLINT [], + ADD COLUMN p90_sips SMALLINT [], + ADD COLUMN p98_sips SMALLINT []; + +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; + +/* + * I want init time to be VIRTUAL, but sqlc doesn't support it yet. + * See https://github.com/sqlc-dev/sqlc/issues/4322. Until then it stays a plain NOT NULL + * column written by the application, so a rebuild that forgets to carry it over fails loudly + * rather than silently nulling the column every read query derives horizons from. + * + * The recency check is dropped: it referenced CURRENT_TIMESTAMP, which is not + * immutable, so it could not be revalidated and breaks ATTACH PARTITION. + */ +ALTER TABLE pred.forecasts + DROP CONSTRAINT IF EXISTS init_time_utc_recency_check; + +-- +goose Down +ALTER TABLE pred.forecasts + ADD CONSTRAINT init_time_utc_recency_check CHECK ( + init_time_utc >= '2000-01-01 00:00:00'::TIMESTAMP + AND init_time_utc < CURRENT_TIMESTAMP + MAKE_INTERVAL(days => 30) + ) NOT VALID; + +ALTER TABLE pred.forecasts + DROP CONSTRAINT IF EXISTS plevel_lengths_match_check, + DROP COLUMN p02_sips, DROP COLUMN p10_sips, DROP COLUMN p25_sips, + DROP COLUMN p50_sips, DROP COLUMN p75_sips, DROP COLUMN p90_sips, + DROP COLUMN p98_sips; + +DROP FUNCTION IF EXISTS uuidv7_extract_timestamp(UUID); + +-- +goose StatementBegin +CREATE FUNCTION uuidv7_extract_timestamp(UUID) RETURNS TIMESTAMPTZ +AS $$ + SELECT to_timestamp( + right(substring(uuid_send($1) from 1 for 6)::text, -1)::bit(48)::int8 + /1000.0); +$$ LANGUAGE sql immutable strict parallel safe; +-- +goose StatementEnd diff --git a/internal/server/postgres/sql/migrations/00012_rebuild_forecast_partitions.sql b/internal/server/postgres/sql/migrations/00012_rebuild_forecast_partitions.sql new file mode 100644 index 0000000..396a0b3 --- /dev/null +++ b/internal/server/postgres/sql/migrations/00012_rebuild_forecast_partitions.sql @@ -0,0 +1,211 @@ +-- +goose Up + +-- +goose StatementBegin +/* + * Rebuilds one pred.forecasts partition with its values folded into arrays, and its rows + * physically ordered to match idx_forecasts_filter. + * + * A rebuild rather than an UPDATE, because rows grow ~3.6x: an in-place update cannot keep the + * new tuple on its page, so every row would be a non-HOT update leaving a dead tuple and a new + * entry in every index. Rebuilding also lets us choose physical order for free, and packs + * indexes at full density. + * + * 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. + * + * The aggregation is chunked into an unlogged staging table and committed per chunk, so a week's + * worth of values is never sorted in one go. + * + * 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. + * + * pred.predicted_generation_values carries a foreign key to pred.forecasts, and PostgreSQL + * refuses to detach a partition that is still referenced. The matching values partition is + * therefore detached and retired in the same transaction as the swap - which is the correct + * coupling anyway, since once a week's forecasts hold arrays its value rows are dead. It is + * renamed rather than dropped so the rebuild stays verifiable and reversible. The parent's + * foreign key is left intact for every partition that has not yet been rebuilt. + * + * Retired tables are left on disk as pred.predicted_generation_values_pXXXXXXXX_retired. They + * are no longer partitions, so the cleanup deployment's DROP TABLE will not remove them - drop + * them explicitly once the rebuild has been verified. + * + * This must be driven one partition at a time rather than looped unattended: DETACH and ATTACH + * each take a brief ACCESS EXCLUSIVE lock on pred.forecasts, and ATTACH validates the partition + * bound and the foreign keys. + * + * CALL pred.rebuild_forecast_partition('forecasts_p20260803') + */ +CREATE OR REPLACE PROCEDURE pred.rebuild_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_fk 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; + + v_lo := partman.uuid7_time_decoder( + (regexp_match(v_bounds, $re$FROM \('([^']+)'\)$re$))[1]::UUID) AT TIME ZONE 'UTC'; + v_hi := partman.uuid7_time_decoder( + (regexp_match(v_bounds, $re$TO \('([^']+)'\)$re$))[1]::UUID) 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; + + /* Reverts on commit of the transaction the swap runs in. */ + EXECUTE format('SET LOCAL work_mem = %L', p_work_mem); + + 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; + + RAISE NOTICE 'rebuilt %: % rows (% forecasts had no values and were dropped)', + p_partition, v_dst, v_all - v_dst; + + /* 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 migrated_check CHECK (p50_sips IS NOT NULL)', v_new); + + EXECUTE format('ANALYZE pred.%I', v_new); + + /* Detaching leaves a standalone copy of the foreign key behind on the values partition, which + * would still block the forecasts detach below, so it has to go too. */ + 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.forecasts ATTACH PARTITION pred.%I %s', v_new, v_bounds); + EXECUTE format('DROP TABLE pred.%I', p_partition); + EXECUTE format('ALTER TABLE pred.%I RENAME TO %I', v_new, p_partition); + + DROP TABLE pred.fc_staging; + + RAISE NOTICE 'done: % in % (old values retained as pred.%_retired, drop once verified)', + p_partition, clock_timestamp() - v_started, v_values; +END; +$$; +-- +goose StatementEnd + +-- +goose Down +DROP PROCEDURE IF EXISTS pred.rebuild_forecast_partition(TEXT, INTERVAL, TEXT); diff --git a/internal/server/postgres/sql/queries/locations.sql b/internal/server/postgres/sql/queries/locations.sql index 41e34a4..59ef8c9 100644 --- a/internal/server/postgres/sql/queries/locations.sql +++ b/internal/server/postgres/sql/queries/locations.sql @@ -12,14 +12,20 @@ INSERT INTO loc.geometries AS l ( ST_CENTROID(ST_GEOMFROMTEXT(sqlc.arg(geom)::TEXT, 4326)) ) ) RETURNING - l.geometry_uuid, l.geometry_name, ST_X(l.associated_point)::REAL AS longitude, ST_Y(l.associated_point)::REAL AS latitude; + l.geometry_uuid, + l.geometry_name, + ST_X(l.associated_point)::REAL AS longitude, + ST_Y(l.associated_point)::REAL AS latitude; -- name: RenameGeometry :one UPDATE loc.geometries AS l SET geometry_name = LOWER(sqlc.arg(new_geometry_name)::TEXT) WHERE l.geometry_uuid = $1 RETURNING - l.geometry_uuid, l.geometry_name, ST_X(l.associated_point)::REAL AS longitude, ST_Y(l.associated_point)::REAL AS latitude; + l.geometry_uuid, + l.geometry_name, + ST_X(l.associated_point)::REAL AS longitude, + ST_Y(l.associated_point)::REAL AS latitude; -- name: ReownGeometry :one /* ReownGeometry assigns a new owning_entity_id to a geometry. @@ -57,7 +63,7 @@ SELECT geometry_uuid, ST_ASBINARY(geom)::BYTEA AS geom_wkb FROM loc.geometries -WHERE geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID []); +WHERE geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID[]); -- name: GetGeometryGeoJSON :one /* GetLocationGeoJSON returns a GeoJSON FeatureCollection for the given geometries. @@ -79,7 +85,7 @@ FROM ( l.geometry_type_id, ST_SIMPLIFYPRESERVETOPOLOGY(l.geom, sqlc.arg(simplification_level)::REAL) AS geom_simple FROM loc.geometries AS l - WHERE l.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID []) + WHERE l.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID[]) ) AS sl; /*- Queries for the sources table -------------------------------------*/ @@ -215,12 +221,12 @@ WHERE OR us.source_type_id = sqlc.narg(source_type_id)::SMALLINT ) AND ( - ARRAY_LENGTH(sqlc.arg(geometry_uuids)::UUID [], 1) IS NULL - OR us.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID []) + ARRAY_LENGTH(sqlc.arg(geometry_uuids)::UUID[], 1) IS NULL + OR us.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID[]) ) AND ( - ARRAY_LENGTH(sqlc.arg(geometry_names)::TEXT [], 1) IS NULL - OR us.geometry_name = ANY(sqlc.arg(geometry_names)::TEXT []) + ARRAY_LENGTH(sqlc.arg(geometry_names)::TEXT[], 1) IS NULL + OR us.geometry_name = ANY(sqlc.arg(geometry_names)::TEXT[]) ) AND ( sqlc.narg(geometry_type_id)::SMALLINT IS NULL @@ -270,12 +276,12 @@ WHERE OR us.source_type_id = sqlc.narg(source_type_id)::SMALLINT ) AND ( - ARRAY_LENGTH(sqlc.arg(geometry_uuids)::UUID [], 1) IS NULL - OR us.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID []) + ARRAY_LENGTH(sqlc.arg(geometry_uuids)::UUID[], 1) IS NULL + OR us.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID[]) ) AND ( - ARRAY_LENGTH(sqlc.arg(geometry_names)::TEXT [], 1) IS NULL - OR us.geometry_name = ANY(sqlc.arg(geometry_names)::TEXT []) + ARRAY_LENGTH(sqlc.arg(geometry_names)::TEXT[], 1) IS NULL + OR us.geometry_name = ANY(sqlc.arg(geometry_names)::TEXT[]) ) AND ( sqlc.narg(geometry_type_id)::SMALLINT IS NULL @@ -325,8 +331,8 @@ WHERE OR us.source_type_id = sqlc.narg(source_type_id)::SMALLINT ) AND ( - ARRAY_LENGTH(sqlc.arg(geometry_uuids)::UUID [], 1) IS NULL - OR us.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID []) + ARRAY_LENGTH(sqlc.arg(geometry_uuids)::UUID[], 1) IS NULL + OR us.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID[]) ) AND ( sqlc.narg(geometry_type_id)::SMALLINT IS NULL diff --git a/internal/server/postgres/sql/queries/observations.sql b/internal/server/postgres/sql/queries/observations.sql index 0dc816f..8f05021 100644 --- a/internal/server/postgres/sql/queries/observations.sql +++ b/internal/server/postgres/sql/queries/observations.sql @@ -8,8 +8,8 @@ SELECT observer_name FROM obs.observers WHERE ( - ARRAY_LENGTH(sqlc.arg(observer_names)::TEXT [], 1) IS NULL - OR observer_name = ANY(sqlc.arg(observer_names)::TEXT []) + ARRAY_LENGTH(sqlc.arg(observer_names)::TEXT[], 1) IS NULL + OR observer_name = ANY(sqlc.arg(observer_names)::TEXT[]) ); -- name: GetObserverByName :one @@ -86,7 +86,7 @@ WHERE * It uses lateral joins to perform a reverse index scan for efficiency. */ WITH target_locations AS ( - SELECT UNNEST(sqlc.arg(geometry_uuids)::UUID []) AS geometry_uuid + SELECT UNNEST(sqlc.arg(geometry_uuids)::UUID[]) AS geometry_uuid ), target_observer AS ( SELECT observer_uuid @@ -138,7 +138,7 @@ SELECT FROM obs.observed_generation_values AS og INNER JOIN loc.sources_mv AS sh USING (geometry_uuid, source_type_id) WHERE - og.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID []) + og.geometry_uuid = ANY(sqlc.arg(geometry_uuids)::UUID[]) AND og.source_type_id = $1 AND og.observer_uuid = $2 AND og.observation_timestamp_utc = sqlc.arg(target_timestamp_utc)::TIMESTAMP diff --git a/internal/server/postgres/sql/queries/predictions.sql b/internal/server/postgres/sql/queries/predictions.sql index 08080af..4df56ee 100644 --- a/internal/server/postgres/sql/queries/predictions.sql +++ b/internal/server/postgres/sql/queries/predictions.sql @@ -53,8 +53,8 @@ SELECT created_at_utc FROM ranked_forecasters WHERE ( - ARRAY_LENGTH(sqlc.arg(forecaster_names)::TEXT [], 1) IS NULL - OR forecaster_name = ANY(sqlc.arg(forecaster_names)::TEXT []) + ARRAY_LENGTH(sqlc.arg(forecaster_names)::TEXT[], 1) IS NULL + OR forecaster_name = ANY(sqlc.arg(forecaster_names)::TEXT[]) ) AND ( NOT sqlc.arg(latest_version_only)::BOOLEAN OR rn = 1 @@ -73,9 +73,16 @@ INSERT INTO pred.forecasts ( value_resolution_mins, target_period, metadata, - created_at_utc + created_at_utc, + p02_sips, + p10_sips, + p25_sips, + p50_sips, + p75_sips, + p90_sips, + p98_sips ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16 ); -- name: DeleteForecastByUUID :exec @@ -110,11 +117,15 @@ INSERT INTO pred.predicted_generation_values ( /* 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 - UNNEST(sqlc.arg(forecaster_names)::TEXT []) AS fname, - UNNEST(sqlc.arg(forecaster_versions)::TEXT []) AS fversion + UNNEST(sqlc.arg(forecaster_names)::TEXT[]) AS fname, + UNNEST(sqlc.arg(forecaster_versions)::TEXT[]) AS fversion ), matched_forecasters AS ( SELECT @@ -125,44 +136,118 @@ matched_forecasters AS ( INNER JOIN requested_forecasters AS rf ON f.forecaster_name = LOWER(rf.fname) AND f.forecaster_version = LOWER(rf.fversion) +), +matched_forecasts AS ( + SELECT + f.forecast_uuid, + f.geometry_uuid, + f.source_type_id, + f.created_at_utc, + f.metadata, + f.init_time_utc, + f.value_resolution_mins, + f.p02_sips, + f.p10_sips, + f.p25_sips, + f.p50_sips, + f.p75_sips, + f.p90_sips, + f.p98_sips, + mf.forecaster_name, + mf.forecaster_version, + LOWER(f.target_period) AS first_target_utc + FROM pred.forecasts AS f + INNER JOIN matched_forecasters AS mf USING (forecaster_id) + WHERE f.geometry_uuid = sqlc.arg(geometry_uuid)::UUID + AND f.source_type_id = sqlc.arg(source_type_id)::SMALLINT + 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 ( + SELECT + mfc.forecaster_name, + mfc.forecaster_version, + mfc.created_at_utc, + mfc.metadata, + mfc.geometry_uuid, + mfc.source_type_id, + mfc.init_time_utc, + ( + EXTRACT(EPOCH FROM (mfc.first_target_utc - mfc.init_time_utc)) / 60 + + (o.ord - 1) * mfc.value_resolution_mins + )::SMALLINT AS horizon_mins, + (mfc.first_target_utc + MAKE_INTERVAL( + 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, + mfc.p75_sips[o.ord] AS p75_sip, + mfc.p90_sips[o.ord] AS p90_sip, + 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. */ SELECT - mf.forecaster_name, - mf.forecaster_version, - f.created_at_utc, - pg.horizon_mins, - pg.p02_sip, - pg.p10_sip, - pg.p25_sip, - pg.p50_sip, - pg.p75_sip, - pg.p90_sip, - pg.p98_sip, + e.forecaster_name, + e.forecaster_version, + e.created_at_utc, + e.horizon_mins, + e.p02_sip, + e.p10_sip, + e.p25_sip, + e.p50_sip, + e.p75_sip, + e.p90_sip, + e.p98_sip, sv.capacity_watts, - f.metadata, - UUIDV7_EXTRACT_TIMESTAMP(f.forecast_uuid)::TIMESTAMP AS init_time_utc, - ( - UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER) - )::TIMESTAMP AS target_time_utc -FROM pred.forecasts AS f - INNER JOIN matched_forecasters AS mf USING (forecaster_id) - INNER JOIN pred.predicted_generation_values AS pg - ON f.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') + e.metadata, + e.init_time_utc, + e.target_time_utc +FROM expanded AS e LEFT OUTER JOIN LATERAL ( SELECT capacity_watts FROM loc.sources_mv AS s - WHERE s.geometry_uuid = f.geometry_uuid - AND s.source_type_id = f.source_type_id - AND s.sys_period - @> (UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER)) - LIMIT 1 - ) AS sv ON TRUE -WHERE f.geometry_uuid = sqlc.arg(geometry_uuid)::UUID - AND f.source_type_id = sqlc.arg(source_type_id)::SMALLINT - 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'); + WHERE s.geometry_uuid = e.geometry_uuid + AND s.source_type_id = e.source_type_id + AND s.sys_period @> e.target_time_utc + LIMIT 1 --noqa + ) AS sv ON TRUE; -- name: GetLatestForecastsAtHorizonSincePivot :many /* GetLatestForecastAtHorizonSincePivot retrieves the latest forecasts for a given location @@ -194,6 +279,11 @@ FROM pred.forecasters AS fr WHERE geometry_uuid = $1 AND source_type_id = $2 AND forecaster_id = fr.forecaster_id + -- Without a lower bound the range is (-infinity, pivot), so MergeAppend has to open + -- every partition including the historical default one. + AND forecast_uuid >= UUIDV7_BOUNDARY( + sqlc.arg(pivot_timestamp)::TIMESTAMP - INTERVAL '7 days' + ) AND forecast_uuid < UUIDV7_BOUNDARY( sqlc.arg(pivot_timestamp)::TIMESTAMP - MAKE_INTERVAL( mins => sqlc.arg(horizon_mins)::INTEGER @@ -206,23 +296,30 @@ FROM pred.forecasters AS fr ORDER BY fr.forecaster_name ASC, f.init_time_utc DESC; -- name: ListPredictionsForLocation :many -/* ListPredictionsForLocation retrieves predicted generation values as a timeseries. - * Multiple overlapping forecasts can make up the timeseries, so predictions with the same target time - * are filtered by lowest allowable horizon (i.e. predicted closest to their target time). - * Predicted values are smallint percentages (sip) of capcity; - * with 0 representing 0% and 30000 representing 100% of capacity. +/* ListPredictionsForLocation retrieves all predicted generation values for a given location, + * source type, and forecaster within a time window. * - * Note that the 3 day intervals are due to our forecasts only going out to 2 days. - * If we increase that horizon, these will need to be increased. + * 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_overlapping_window AS ( +WITH allowed_forecasts AS ( SELECT f.forecast_uuid, f.geometry_uuid, f.source_type_id, f.created_at_utc, f.metadata, - UUIDV7_EXTRACT_TIMESTAMP(f.forecast_uuid)::TIMESTAMP AS init_time_utc + f.value_resolution_mins, + f.init_time_utc, + f.p02_sips, + f.p10_sips, + f.p25_sips, + f.p50_sips, + f.p75_sips, + f.p90_sips, + f.p98_sips, + LOWER(f.target_period) AS first_target_utc FROM pred.forecasts AS f WHERE f.geometry_uuid = $1 AND f.source_type_id = $2 @@ -238,41 +335,109 @@ WITH allowed_forecasts_overlapping_window AS ( AND f.created_at_utc <= COALESCE(sqlc.narg(pivot_timestamp)::TIMESTAMP, CURRENT_TIMESTAMP) AND f.target_period && TSRANGE( sqlc.arg(start_timestamp_utc)::TIMESTAMP, - sqlc.arg(end_timestamp_utc)::TIMESTAMP, - '[]' + sqlc.arg(end_timestamp_utc)::TIMESTAMP, '[]' ) ), -winning_predictions AS ( - SELECT DISTINCT ON ( - UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER) - ) - fow.forecast_uuid, - fow.init_time_utc, - fow.created_at_utc, - fow.geometry_uuid, - fow.source_type_id, +sliced AS ( + /* 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. */ + SELECT + af.*, + GREATEST(1, CEIL(EXTRACT(EPOCH FROM (GREATEST( + sqlc.arg(start_timestamp_utc)::TIMESTAMP, + af.init_time_utc + MAKE_INTERVAL(mins => sqlc.arg(horizon_mins)::INTEGER) + ) - af.first_target_utc)) / 60.0 / af.value_resolution_mins)::INTEGER + 1) AS lo, + LEAST(ARRAY_LENGTH(af.p50_sips, 1), FLOOR(EXTRACT(EPOCH FROM ( + 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 ( + /* Expand the sliced arrays into rows, with each row representing a single + * target time and its associated predicted values. */ + SELECT + s.forecast_uuid, + s.init_time_utc, + s.created_at_utc, + s.metadata, + s.geometry_uuid, + s.source_type_id, + (s.first_target_utc + MAKE_INTERVAL( + mins => + ((s.lo + o.ord - 2) * s.value_resolution_mins)::INTEGER + ))::TIMESTAMP + AS target_time_utc, + (EXTRACT(EPOCH FROM ( + 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 + 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.p25_sip, pg.p10_sip, - pg.p50_sip, + pg.p25_sip, pg.p75_sip, pg.p90_sip, - pg.p98_sip, - fow.metadata, - ( - UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER) - )::TIMESTAMP AS target_time_utc - FROM allowed_forecasts_overlapping_window AS fow - INNER JOIN pred.predicted_generation_values AS pg USING (forecast_uuid) - WHERE ( - UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + 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 - -- Sorting by decreasing init time ensures the DISTINCT captures the lowest allowed horizon - ORDER BY - (UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER)) ASC, - fow.init_time_utc DESC + 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. */ + SELECT DISTINCT ON (target_time_utc) * + FROM expanded + ORDER BY target_time_utc ASC, forecast_uuid DESC ) SELECT wp.horizon_mins, @@ -297,33 +462,46 @@ WHERE sv.sys_period @> wp.target_time_utc ORDER BY wp.target_time_utc ASC; -- name: ListPredictionsAtTimeForLocations :many -/* ListPredictionsAtTimeForLocations retrieves predicted generation values as percentages - * of capacity for a specific time and horizon. - * This is useful for comparing predictions across multiple locations. - * Predicted values are 16-bit integers, with 0 representing 0% and 30000 representing 100% of capacity. - * - * Note that the 3 day intervals are due to our forecasts only going out to 2 days. - * If we increase that horizon, these will need to be increased. +/* PostgreSQL returns NULL on an out of bounds array index. As such, ARRAY_LENGTH is used + * to guard against this. */ --- name: ListPredictionsAtTimeForLocations :many WITH target_locations AS ( - SELECT UNNEST(sqlc.arg(geometry_uuids)::UUID []) AS geometry_uuid + SELECT UNNEST(sqlc.arg(geometry_uuids)::UUID[]) AS geometry_uuid ), latest_allowed_forecast_per_location AS ( SELECT lf.forecast_uuid, - tl.geometry_uuid::UUID AS geometry_uuid, -- again, SQLC complains without this + tl.geometry_uuid::UUID AS geometry_uuid, lf.source_type_id, lf.created_at_utc, lf.metadata, - UUIDV7_EXTRACT_TIMESTAMP(lf.forecast_uuid)::TIMESTAMP AS init_time_utc + lf.init_time_utc, + lf.value_resolution_mins, + lf.p02_sips, + lf.p10_sips, + lf.p25_sips, + lf.p50_sips, + lf.p75_sips, + lf.p90_sips, + lf.p98_sips, + LOWER(lf.target_period) AS first_target_utc FROM target_locations AS tl CROSS JOIN LATERAL ( SELECT f.forecast_uuid, f.source_type_id, f.created_at_utc, - f.metadata + f.metadata, + f.init_time_utc, + f.value_resolution_mins, + f.target_period, + f.p02_sips, + f.p10_sips, + f.p25_sips, + f.p50_sips, + f.p75_sips, + f.p90_sips, + f.p98_sips FROM pred.forecasts AS f WHERE f.geometry_uuid = tl.geometry_uuid AND f.source_type_id = $1 @@ -342,35 +520,72 @@ latest_allowed_forecast_per_location AS ( ORDER BY f.forecast_uuid DESC LIMIT 1 ) AS lf +), +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. */ + SELECT + laf.*, + (EXTRACT(EPOCH FROM ( + sqlc.arg(target_timestamp_utc)::TIMESTAMP - laf.first_target_utc + )) / 60 / laf.value_resolution_mins)::INTEGER + 1 AS idx, + (EXTRACT(EPOCH FROM ( + sqlc.arg(target_timestamp_utc)::TIMESTAMP - laf.init_time_utc + )) / 60)::SMALLINT AS horizon_mins + FROM latest_allowed_forecast_per_location AS laf ) SELECT - laf.forecast_uuid, - laf.geometry_uuid, - laf.source_type_id, - pg.horizon_mins, - pg.p02_sip, - pg.p10_sip, - pg.p25_sip, - pg.p50_sip, - pg.p75_sip, - pg.p90_sip, - pg.p98_sip, - laf.created_at_utc, - laf.init_time_utc, + i.forecast_uuid, + i.geometry_uuid, + i.source_type_id, + i.horizon_mins, + i.created_at_utc, + i.init_time_utc, sv.capacity_watts, sv.latitude, sv.longitude, sv.geometry_name, - laf.metadata, + 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, sqlc.arg(target_timestamp_utc)::TIMESTAMP AS target_time_utc -FROM latest_allowed_forecast_per_location AS laf - INNER JOIN pred.predicted_generation_values AS pg USING (forecast_uuid) +FROM indexed AS i INNER JOIN loc.sources_mv AS sv USING (geometry_uuid, source_type_id) -WHERE - (UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER)) - = sqlc.arg(target_timestamp_utc)::TIMESTAMP - AND sv.sys_period @> sqlc.arg(target_timestamp_utc)::TIMESTAMP; - + /* 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; -- 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. @@ -383,29 +598,55 @@ WITH relevant_forecasts AS ( f.forecast_uuid, f.source_type_id, f.geometry_uuid, - f.forecaster_id + f.forecaster_id, + f.init_time_utc, + f.value_resolution_mins, + f.p50_sips, + LOWER(f.target_period) AS first_target_utc FROM pred.forecasts AS f WHERE f.geometry_uuid = $4 AND f.source_type_id = $1 AND f.forecaster_id = $2 AND f.forecast_uuid >= UUIDV7_BOUNDARY(sqlc.arg(pivot_timestamp)::TIMESTAMP - INTERVAL '8 days') AND f.forecast_uuid < UUIDV7_BOUNDARY(sqlc.arg(pivot_timestamp)::TIMESTAMP + INTERVAL '1 millisecond') - AND UUIDV7_EXTRACT_TIMESTAMP(f.forecast_uuid)::TIME = sqlc.arg(pivot_timestamp)::TIMESTAMP::TIME + AND f.init_time_utc::TIME = sqlc.arg(pivot_timestamp)::TIMESTAMP::TIME ), -relevant_predicted_values AS MATERIALIZED ( +expanded_array AS ( + SELECT + rf.geometry_uuid, + rf.source_type_id, + ( + 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, - ( - UUIDV7_EXTRACT_TIMESTAMP(pg.forecast_uuid)::TIMESTAMP + MAKE_INTERVAL(mins => pg.horizon_mins::INTEGER) - )::TIMESTAMP AS target_time_utc + (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 pg.forecast_uuid >= UUIDV7_BOUNDARY(sqlc.arg(pivot_timestamp)::TIMESTAMP - INTERVAL '8 days') + 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 geometry_uuid,