feat(postgres): Move to array storage

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"

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,

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) {

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
 }
 

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