From 896bbdbb603fa3003c0ba5700abc1f5b15807b85 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:00:22 +0100 Subject: [PATCH 1/3] feat(postgres): Move to array storage --- internal/server/postgres/.sqlc.yaml | 75 ++++ internal/server/postgres/dataserverimpl.go | 29 +- .../server/postgres/dataserverimpl_test.go | 57 +++ internal/server/postgres/mappers.go | 83 ++++ .../00011_forecast_value_arrays.sql | 90 ++++ .../00012_rebuild_forecast_partitions.sql | 211 +++++++++ .../postgres/sql/queries/predictions.sql | 414 ++++++++++++------ 7 files changed, 802 insertions(+), 157 deletions(-) create mode 100644 internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql create mode 100644 internal/server/postgres/sql/migrations/00012_rebuild_forecast_partitions.sql 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..aa86aa3 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,10 @@ 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..c162ae2 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -2033,11 +2033,36 @@ 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 +2486,19 @@ 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 +2508,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..0df20ca 100644 --- a/internal/server/postgres/mappers.go +++ b/internal/server/postgres/mappers.go @@ -70,11 +70,87 @@ 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 fmt.Errorf("a forecast must contain at least two values") + } + + resolution := int32(values[1].HorizonMins) - int32(values[0].HorizonMins) + if resolution <= 0 { + return fmt.Errorf("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 fmt.Errorf("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/migrations/00011_forecast_value_arrays.sql b/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql new file mode 100644 index 0000000..e633230 --- /dev/null +++ b/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql @@ -0,0 +1,90 @@ +-- +goose Up + +/* + * Replaces uuidv7_extract_timestamp with an immutable implementation returning UTC wall time + * directly, rather than a TIMESTAMPTZ that every call site then cast using the session TimeZone. + * Measured at ~215ns/row against ~378ns/row for the version it replaces. + * + * PG18's native uuid_extract_timestamp() is faster still (~96ns/row), but returns NULL for any + * UUID that is not version 1 or 7 - including pg_partman's partition bounds, which zero the + * version nibble ('019f15d3-5800-0000-...'). Decoding the first 48 bits directly keeps the + * original function's contract of working on any UUID, which the partition rebuild in 00012 + * depends on. The speed difference is immaterial now that no hot-path query calls this. + */ + +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 TIMESTAMP 'epoch' + ( + ('x' || encode(substring(uuid_send(u) FROM 1 FOR 6), 'hex'))::BIT(48)::BIGINT +) * INTERVAL '1 millisecond'; +-- +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..b83b427 --- /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 := uuidv7_extract_timestamp( + (regexp_match(v_bounds, $re$FROM \('([^']+)'\)$re$))[1]::UUID); + v_hi := uuidv7_extract_timestamp( + (regexp_match(v_bounds, $re$TO \('([^']+)'\)$re$))[1]::UUID); + + 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/predictions.sql b/internal/server/postgres/sql/queries/predictions.sql index 08080af..c7dd43f 100644 --- a/internal/server/postgres/sql/queries/predictions.sql +++ b/internal/server/postgres/sql/queries/predictions.sql @@ -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,6 +117,10 @@ 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 @@ -125,44 +136,86 @@ 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, + LOWER(f.target_period) AS first_target_utc, + 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 + 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)) + 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 - ) 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'); + ) AS sv ON TRUE; -- name: GetLatestForecastsAtHorizonSincePivot :many /* GetLatestForecastAtHorizonSincePivot retrieves the latest forecasts for a given location @@ -194,6 +247,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,124 +264,151 @@ 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.forecast_uuid, f.geometry_uuid, f.source_type_id, f.created_at_utc, f.metadata, + f.value_resolution_mins, f.init_time_utc, + LOWER(f.target_period) AS first_target_utc, + 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 = $1 AND f.source_type_id = $2 AND f.forecaster_id = $3 AND f.forecast_uuid >= UUIDV7_BOUNDARY( - sqlc.arg(start_timestamp_utc)::TIMESTAMP - INTERVAL '3 days' - ) + sqlc.arg(start_timestamp_utc)::TIMESTAMP - INTERVAL '3 days') AND f.forecast_uuid < UUIDV7_BOUNDARY( sqlc.arg(end_timestamp_utc)::TIMESTAMP - MAKE_INTERVAL(mins => sqlc.arg(horizon_mins)::INTEGER) - + INTERVAL '1 millisecond' - ) + + INTERVAL '1 millisecond') 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.p02_sip, - pg.p25_sip, - pg.p10_sip, - pg.p50_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.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. */ + SELECT DISTINCT ON (target_time_utc) * + FROM expanded + ORDER BY target_time_utc ASC, forecast_uuid DESC ) SELECT - wp.horizon_mins, - wp.p02_sip, - wp.p25_sip, - wp.p10_sip, - wp.p50_sip, - wp.p75_sip, - wp.p90_sip, - wp.p98_sip, - wp.target_time_utc, - wp.metadata, - wp.init_time_utc, - wp.created_at_utc, - sv.capacity_watts, - sv.latitude, - sv.longitude, - sv.geometry_name + wp.horizon_mins, wp.p02_sip, wp.p25_sip, wp.p10_sip, wp.p50_sip, + wp.p75_sip, wp.p90_sip, wp.p98_sip, + wp.target_time_utc, wp.metadata, wp.init_time_utc, wp.created_at_utc, + sv.capacity_watts, sv.latitude, sv.longitude, sv.geometry_name FROM winning_predictions AS wp INNER JOIN loc.sources_mv AS sv USING (geometry_uuid, source_type_id) 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 ), 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, + LOWER(lf.target_period) AS first_target_utc, + lf.p02_sips, lf.p10_sips, lf.p25_sips, lf.p50_sips, + lf.p75_sips, lf.p90_sips, lf.p98_sips FROM target_locations AS tl CROSS JOIN LATERAL ( SELECT - f.forecast_uuid, - f.source_type_id, - f.created_at_utc, - f.metadata + f.forecast_uuid, f.source_type_id, f.created_at_utc, 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 +427,58 @@ 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, + 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, + i.created_at_utc, + i.init_time_utc, sv.capacity_watts, sv.latitude, sv.longitude, sv.geometry_name, - laf.metadata, + i.metadata, 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 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 +491,51 @@ 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, + LOWER(f.target_period) AS first_target_utc, + f.p50_sips 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, From 38721d8dc9bef65e1e8756377a09b080082bd393 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:25:43 +0100 Subject: [PATCH 2/3] fix(sql): fix linting --- Makefile | 1 + internal/server/postgres/dataserverimpl.go | 5 +- .../server/postgres/dataserverimpl_test.go | 2 + internal/server/postgres/mappers.go | 10 +- internal/server/postgres/sql/.sqlfluff.toml | 1 + .../00011_forecast_value_arrays.sql | 21 +- .../00012_rebuild_forecast_partitions.sql | 8 +- .../server/postgres/sql/queries/locations.sql | 34 ++- .../postgres/sql/queries/observations.sql | 8 +- .../postgres/sql/queries/predictions.sql | 255 +++++++++++++----- 10 files changed, 227 insertions(+), 118 deletions(-) diff --git a/Makefile b/Makefile index a3548a9..e68ecf9 100644 --- a/Makefile +++ b/Makefile @@ -55,6 +55,7 @@ lint: @uvx -q sqlfluff fix -q \ --disable-progress-bar \ --config=internal/server/postgres/sql/.sqlfluff.toml \ + --show-lint-violations \ internal/server/postgres/sql/queries .PHONY: bench diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index aa86aa3..5c0cd98 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -1560,7 +1560,10 @@ func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts( } if err := validateForecastValues(req.Values); err != nil { - return status.Error(codes.InvalidArgument, fmt.Sprintf("invalid forecast values: %v", err)) + return status.Error( + codes.InvalidArgument, + fmt.Sprintf("invalid forecast values: %v", err), + ) } fKey := forecasterKey{ diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index c162ae2..a69c26e 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -2040,6 +2040,7 @@ func TestCreateForecast(t *testing.T) { if i%2 == 0 { statFractions["p25"] = 0.25 } + yieldsPartial[i] = &pb.CreateForecastRequest_ForecastValue{ HorizonMins: uint32(i * 30), P50Fraction: 0.5, @@ -2492,6 +2493,7 @@ func TestStreamCreateForecasts(t *testing.T) { if i%2 == 0 { statFractions["p25"] = 0.25 } + yieldsPartial[i] = &pb.CreateForecastRequest_ForecastValue{ HorizonMins: uint32(i * 30), P50Fraction: 0.5, diff --git a/internal/server/postgres/mappers.go b/internal/server/postgres/mappers.go index 0df20ca..57bcc1a 100644 --- a/internal/server/postgres/mappers.go +++ b/internal/server/postgres/mappers.go @@ -1,6 +1,7 @@ package postgres import ( + "errors" "fmt" "time" @@ -93,7 +94,7 @@ func extractSIPStatSlice(values []*pb.CreateForecastRequest_ForecastValue, key s return out } -// extractP50Slice builds the p50 array. p50 is a top-level field on ForecastValue rather than a +// 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)) @@ -104,7 +105,6 @@ func extractP50Slice(values []*pb.CreateForecastRequest_ForecastValue) []int16 { return out } - // sipToFraction converts a SIP value to a fraction. func sipToFraction(sip int16) float32 { return float32(sip) / 30000.0 @@ -115,17 +115,17 @@ func sipToFraction(sip int16) float32 { // either present on every value or on none. func validateForecastValues(values []*pb.CreateForecastRequest_ForecastValue) error { if len(values) < 2 { - return fmt.Errorf("a forecast must contain at least two values") + return errors.New("a forecast must contain at least two values") } resolution := int32(values[1].HorizonMins) - int32(values[0].HorizonMins) if resolution <= 0 { - return fmt.Errorf("forecast horizons must be monotonically increasing") + 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 fmt.Errorf("forecast horizons must be evenly spaced in time") + return errors.New("forecast horizons must be evenly spaced in time") } } 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 index e633230..2aad77c 100644 --- a/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql +++ b/internal/server/postgres/sql/migrations/00011_forecast_value_arrays.sql @@ -1,26 +1,11 @@ -- +goose Up - -/* - * Replaces uuidv7_extract_timestamp with an immutable implementation returning UTC wall time - * directly, rather than a TIMESTAMPTZ that every call site then cast using the session TimeZone. - * Measured at ~215ns/row against ~378ns/row for the version it replaces. - * - * PG18's native uuid_extract_timestamp() is faster still (~96ns/row), but returns NULL for any - * UUID that is not version 1 or 7 - including pg_partman's partition bounds, which zero the - * version nibble ('019f15d3-5800-0000-...'). Decoding the first 48 bits directly keeps the - * original function's contract of working on any UUID, which the partition rebuild in 00012 - * depends on. The speed difference is immaterial now that no hot-path query calls this. - */ - 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 TIMESTAMP 'epoch' + ( - ('x' || encode(substring(uuid_send(u) FROM 1 FOR 6), 'hex'))::BIT(48)::BIGINT -) * INTERVAL '1 millisecond'; + LANGUAGE sql + IMMUTABLE STRICT PARALLEL SAFE + RETURN uuid_extract_timestamp(u) AT TIME ZONE 'UTC'; -- +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 index b83b427..396a0b3 100644 --- a/internal/server/postgres/sql/migrations/00012_rebuild_forecast_partitions.sql +++ b/internal/server/postgres/sql/migrations/00012_rebuild_forecast_partitions.sql @@ -76,10 +76,10 @@ BEGIN p_partition, v_values; END IF; - v_lo := uuidv7_extract_timestamp( - (regexp_match(v_bounds, $re$FROM \('([^']+)'\)$re$))[1]::UUID); - v_hi := uuidv7_extract_timestamp( - (regexp_match(v_bounds, $re$TO \('([^']+)'\)$re$))[1]::UUID); + 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, 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 c7dd43f..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 @@ -124,8 +124,8 @@ INSERT INTO pred.predicted_generation_values ( */ 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 @@ -139,11 +139,23 @@ matched_forecasters AS ( ), 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, - LOWER(f.target_period) AS first_target_utc, - 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 + 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 @@ -153,12 +165,21 @@ matched_forecasts AS ( ), 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, + 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, @@ -167,17 +188,28 @@ expanded_array AS ( 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) + 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, + 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 + 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 @@ -214,7 +246,7 @@ FROM expanded AS e 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 + LIMIT 1 --noqa ) AS sv ON TRUE; -- name: GetLatestForecastsAtHorizonSincePivot :many @@ -273,30 +305,45 @@ ORDER BY fr.forecaster_name ASC, f.init_time_utc DESC; */ WITH allowed_forecasts AS ( SELECT - f.forecast_uuid, f.geometry_uuid, f.source_type_id, f.created_at_utc, f.metadata, - f.value_resolution_mins, f.init_time_utc, - LOWER(f.target_period) AS first_target_utc, - f.p02_sips, f.p10_sips, f.p25_sips, f.p50_sips, f.p75_sips, f.p90_sips, f.p98_sips + f.forecast_uuid, + f.geometry_uuid, + f.source_type_id, + f.created_at_utc, + f.metadata, + 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 AND f.forecaster_id = $3 AND f.forecast_uuid >= UUIDV7_BOUNDARY( - sqlc.arg(start_timestamp_utc)::TIMESTAMP - INTERVAL '3 days') + sqlc.arg(start_timestamp_utc)::TIMESTAMP - INTERVAL '3 days' + ) AND f.forecast_uuid < UUIDV7_BOUNDARY( sqlc.arg(end_timestamp_utc)::TIMESTAMP - MAKE_INTERVAL(mins => sqlc.arg(horizon_mins)::INTEGER) - + INTERVAL '1 millisecond') + + INTERVAL '1 millisecond' + ) 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, '[]' + ) ), 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.*, + 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) @@ -311,10 +358,16 @@ 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 + 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 @@ -329,20 +382,31 @@ expanded_array AS ( 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) + 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.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 + 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 @@ -350,15 +414,17 @@ expanded_legacy AS ( * 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') + 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') + + 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 + 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 ( @@ -374,10 +440,22 @@ winning_predictions AS ( ORDER BY target_time_utc ASC, forecast_uuid DESC ) SELECT - wp.horizon_mins, wp.p02_sip, wp.p25_sip, wp.p10_sip, wp.p50_sip, - wp.p75_sip, wp.p90_sip, wp.p98_sip, - wp.target_time_utc, wp.metadata, wp.init_time_utc, wp.created_at_utc, - sv.capacity_watts, sv.latitude, sv.longitude, sv.geometry_name + wp.horizon_mins, + wp.p02_sip, + wp.p25_sip, + wp.p10_sip, + wp.p50_sip, + wp.p75_sip, + wp.p90_sip, + wp.p98_sip, + wp.target_time_utc, + wp.metadata, + wp.init_time_utc, + wp.created_at_utc, + sv.capacity_watts, + sv.latitude, + sv.longitude, + sv.geometry_name FROM winning_predictions AS wp INNER JOIN loc.sources_mv AS sv USING (geometry_uuid, source_type_id) WHERE sv.sys_period @> wp.target_time_utc @@ -388,7 +466,7 @@ ORDER BY wp.target_time_utc ASC; * to guard against this. */ 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 @@ -399,16 +477,31 @@ latest_allowed_forecast_per_location AS ( lf.metadata, lf.init_time_utc, lf.value_resolution_mins, - LOWER(lf.target_period) AS first_target_utc, - lf.p02_sips, lf.p10_sips, lf.p25_sips, lf.p50_sips, - lf.p75_sips, lf.p90_sips, lf.p98_sips + 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.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 + f.forecast_uuid, + f.source_type_id, + f.created_at_utc, + 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 @@ -432,7 +525,8 @@ 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.*, + SELECT + laf.*, (EXTRACT(EPOCH FROM ( sqlc.arg(target_timestamp_utc)::TIMESTAMP - laf.first_target_utc )) / 60 / laf.value_resolution_mins)::INTEGER + 1 AS idx, @@ -446,13 +540,6 @@ SELECT i.geometry_uuid, i.source_type_id, i.horizon_mins, - 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, i.created_at_utc, i.init_time_utc, sv.capacity_watts, @@ -460,22 +547,42 @@ SELECT 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, sqlc.arg(target_timestamp_utc)::TIMESTAMP AS target_time_utc FROM indexed AS i 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 JOIN LATERAL ( - SELECT pg.p02_sip, pg.p10_sip, pg.p25_sip, pg.p50_sip, pg.p75_sip, pg.p90_sip, pg.p98_sip + 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) + ( + 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; @@ -494,8 +601,8 @@ WITH relevant_forecasts AS ( f.forecaster_id, f.init_time_utc, f.value_resolution_mins, - LOWER(f.target_period) AS first_target_utc, - f.p50_sips + 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 @@ -508,13 +615,17 @@ 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, + ( + 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 + (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) + CROSS JOIN LATERAL UNNEST(rf.p50_sips) WITH ORDINALITY AS o (p50_sip, ord) WHERE rf.p50_sips IS NOT NULL ), expanded_legacy AS ( From 696be169dcc59ea6fbd3f7b28820a13e17789a26 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:30:01 +0100 Subject: [PATCH 3/3] fix(lint): Don't show lint violations --- Makefile | 1 - 1 file changed, 1 deletion(-) diff --git a/Makefile b/Makefile index e68ecf9..a3548a9 100644 --- a/Makefile +++ b/Makefile @@ -55,7 +55,6 @@ lint: @uvx -q sqlfluff fix -q \ --disable-progress-bar \ --config=internal/server/postgres/sql/.sqlfluff.toml \ - --show-lint-violations \ internal/server/postgres/sql/queries .PHONY: bench