filter observations from spanner for existence requests (#1561)

This is to make the results from spanner smaller for existence requests
that don't need them

(note: some of the tests got updated with some temperature data since I
added it to the test instance, but that's unrelated to this change)

Author: n-h-diaz

internal/server/spanner/datasource.go

@@ -103,18 +103,19 @@ func (sds *SpannerDataSource) Observation(ctx context.Context, req *pbv2.Observa
 	date := req.Date
 	var observations []*Observation
 	var err error
+	filterObs := isExistenceRequest(req.Select)
 
 	if entityExpr != "" {
 		containedInPlace, err := v2.ParseContainedInPlace(entityExpr)
 		if err != nil {
 			return nil, fmt.Errorf("error getting observations (contained in): %v", err)
 		}
-		observations, err = sds.client.GetObservationsContainedInPlace(ctx, variables, containedInPlace, date)
+		observations, err = sds.client.GetObservationsContainedInPlace(ctx, variables, containedInPlace, date, filterObs)
 		if err != nil {
 			return nil, fmt.Errorf("error getting observations (contained in): %v", err)
 		}
 	} else {
-		observations, err = sds.client.GetObservations(ctx, variables, entities, date)
+		observations, err = sds.client.GetObservations(ctx, variables, entities, date, filterObs)
 		if err != nil {
 			return nil, fmt.Errorf("error getting observations: %v", err)
 		}

internal/server/spanner/dsutil.go

@@ -216,28 +216,44 @@ func selectFieldsToQueryOptions(selectFields []string) queryOptions {
 	return qo
 }
 
-// Whether to return all observations in the Observation response.
-func queryObs(qo *queryOptions) bool {
+// Whether the queryOptions are for a full observation request.
+func isObservationRequest(qo *queryOptions) bool {
 	return qo.date && qo.value
 }
 
-func buildBaseObsStatement(variables []string, entities []string, date string) spanner.Statement {
+// Whether the queryOptions are for an existence request.
+func isExistenceRequest(selectFields []string) bool {
+	qo := selectFieldsToQueryOptions(selectFields)
+	return !isObservationRequest(&qo) && !qo.facet
+}
+
+func buildBaseObsStatement(variables []string, entities []string, date string, filterObs bool) spanner.Statement {
 	stmt := spanner.Statement{
 		Params: map[string]interface{}{},
 	}
 	filters := []string{}
 
+	var baseStmt string
+	var obsStmt string
 	switch date {
 	case "":
-		stmt.SQL = statements.getAllObs
+		baseStmt = statements.getObs
+		obsStmt = statements.allObs
 	case shared.LATEST:
-		stmt.SQL = statements.getLatestObs
+		baseStmt = statements.getObs
+		obsStmt = statements.latestObs
 	default:
-		stmt.SQL = statements.getDateObs
+		baseStmt = statements.getDateObs
 		stmt.Params["date"] = fmt.Sprintf("$.%s", date)
+		obsStmt = statements.dateObs
 		filters = append(filters, statements.selectDate)
 	}
 
+	if filterObs {
+		obsStmt = statements.emptyObs
+	}
+	stmt.SQL = fmt.Sprintf(baseStmt, obsStmt)
+
 	if len(variables) > 0 {
 		stmt.Params["variables"] = variables
 		filters = append(filters, statements.selectVariableDcids)
@@ -272,7 +288,7 @@ func observationsToObservationResponse(req *pbv2.ObservationRequest, observation
 	// For now, V3 will match the behavior of V2 to preserve backward compatibility and allow datasource merging.
 	// TODO: Unify these responses more.
 	qo := selectFieldsToQueryOptions(req.Select)
-	if queryObs(&qo) {
+	if isObservationRequest(&qo) {
 		// Returns FacetObservations with PointStats.
 		return obsToObsResponse(req, observations)
 	} else if qo.facet {

internal/server/spanner/golden/query/get_observations_entity.json

@@ -3,16 +3,7 @@
     "VariableMeasured": "Count_HeatTemperatureEvent",
     "ObservationAbout": "wikidataId/Q341968",
     "Observations": {
-      "Observations": [
-        {
-          "Date": "2022-09",
-          "Value": "1"
-        },
-        {
-          "Date": "2022-10",
-          "Value": "2"
-        }
-      ]
+      "Observations": []
     },
     "Provenance": "dc/base/TemperatureEvents_Agg",
     "ObservationPeriod": "P1M",
@@ -26,12 +17,7 @@
     "VariableMeasured": "Count_HeatTemperatureEvent",
     "ObservationAbout": "wikidataId/Q341968",
     "Observations": {
-      "Observations": [
-        {
-          "Date": "2022",
-          "Value": "3"
-        }
-      ]
+      "Observations": []
     },
     "Provenance": "dc/base/TemperatureEvents_Agg",
     "ObservationPeriod": "P1Y",
@@ -45,12 +31,7 @@
     "VariableMeasured": "Count_Person",
     "ObservationAbout": "wikidataId/Q341968",
     "Observations": {
-      "Observations": [
-        {
-          "Date": "2016",
-          "Value": "653"
-        }
-      ]
+      "Observations": []
     },
     "Provenance": "dc/base/WikidataPopulation",
     "ObservationPeriod": "",
@@ -64,12 +45,7 @@
     "VariableMeasured": "Count_Person",
     "ObservationAbout": "wikidataId/Q341968",
     "Observations": {
-      "Observations": [
-        {
-          "Date": "2021",
-          "Value": "606"
-        }
-      ]
+      "Observations": []
     },
     "Provenance": "dc/base/WikipediaStatsData",
     "ObservationPeriod": "",
@@ -78,5 +54,19 @@
     "ScalingFactor": "",
     "ImportName": "WikipediaStatsData",
     "ProvenanceURL": "https://www.wikipedia.org"
+  },
+  {
+    "VariableMeasured": "Max_Temperature",
+    "ObservationAbout": "wikidataId/Q341968",
+    "Observations": {
+      "Observations": []
+    },
+    "Provenance": "dc/base/NOAA_EPA_Observed_Historical_Weather",
+    "ObservationPeriod": "P1M",
+    "MeasurementMethod": "dcAggregate/NASAGSOD_NASAGHCN_EPA",
+    "Unit": "Celsius",
+    "ScalingFactor": "",
+    "ImportName": "NOAA_EPA_Observed_Historical_Weather",
+    "ProvenanceURL": "https://www.noaa.gov/"
   }
 ]

internal/server/spanner/golden/query_test.go

@@ -348,6 +348,7 @@ func TestGetObservations(t *testing.T) {
 		variables  []string
 		entities   []string
 		date       string
+		filterObs  bool
 		goldenFile string
 	}{
 		{
@@ -357,6 +358,7 @@ func TestGetObservations(t *testing.T) {
 		},
 		{
 			entities:   []string{"wikidataId/Q341968"},
+			filterObs:  true,
 			goldenFile: "get_observations_entity.json",
 		},
 		{
@@ -372,7 +374,7 @@ func TestGetObservations(t *testing.T) {
 			goldenFile: "get_observations_date.json",
 		},
 	} {
-		actual, err := client.GetObservations(ctx, c.variables, c.entities, c.date)
+		actual, err := client.GetObservations(ctx, c.variables, c.entities, c.date, c.filterObs)
 
 		if err != nil {
 			t.Fatalf("GetObservations error (%v): %v", c.goldenFile, err)
@@ -417,6 +419,7 @@ func TestGetObservationsContainedInPlace(t *testing.T) {
 		variables        []string
 		containedInPlace *v2.ContainedInPlace
 		date             string
+		filterObs        bool
 		goldenFile       string
 	}{
 		{
@@ -437,7 +440,7 @@ func TestGetObservationsContainedInPlace(t *testing.T) {
 			goldenFile:       "get_observations_contained_in_date.json",
 		},
 	} {
-		actual, err := client.GetObservationsContainedInPlace(ctx, c.variables, c.containedInPlace, c.date)
+		actual, err := client.GetObservationsContainedInPlace(ctx, c.variables, c.containedInPlace, c.date, c.filterObs)
 
 		if err != nil {
 			t.Fatalf("GetObservations error (%v): %v", c.goldenFile, err)

internal/server/spanner/model.go

@@ -73,6 +73,7 @@ func (ts *TimeSeries) DecodeSpanner(val interface{}) (err error) {
 	if !ok {
 		return fmt.Errorf("failed to decode TimeSeries: (%v)", val)
 	}
+	ts.Observations = []*DateValue{}
 	for _, v := range listVal.Values {
 		var data map[string]string
 		err := json.Unmarshal([]byte(v.GetStringValue()), &data)

internal/server/spanner/query.go

@@ -167,15 +167,15 @@ func (sc *SpannerClient) GetNodeEdgesByID(ctx context.Context, ids []string, arc
 }
 
 // GetObservations retrieves observations from Spanner given a list of variables and entities.
-func (sc *SpannerClient) GetObservations(ctx context.Context, variables []string, entities []string, date string) ([]*Observation, error) {
+func (sc *SpannerClient) GetObservations(ctx context.Context, variables []string, entities []string, date string, filterObs bool) ([]*Observation, error) {
 	var observations []*Observation
 	if len(entities) == 0 {
 		return nil, fmt.Errorf("entity must be specified")
 	}
 
 	err := sc.queryAndCollect(
 		ctx,
-		buildBaseObsStatement(variables, entities, date),
+		buildBaseObsStatement(variables, entities, date, filterObs),
 		func() interface{} {
 			return &Observation{}
 		},
@@ -192,13 +192,13 @@ func (sc *SpannerClient) GetObservations(ctx context.Context, variables []string
 }
 
 // GetObservationsContainedInPlace retrieves observations from Spanner given a list of variables and an entity expression.
-func (sc *SpannerClient) GetObservationsContainedInPlace(ctx context.Context, variables []string, containedInPlace *v2.ContainedInPlace, date string) ([]*Observation, error) {
+func (sc *SpannerClient) GetObservationsContainedInPlace(ctx context.Context, variables []string, containedInPlace *v2.ContainedInPlace, date string, filterObs bool) ([]*Observation, error) {
 	var observations []*Observation
 	if len(variables) == 0 || containedInPlace == nil {
 		return observations, nil
 	}
 
-	stmt := buildBaseObsStatement(variables, []string{} /*entities*/, date)
+	stmt := buildBaseObsStatement(variables, []string{} /*entities*/, date, filterObs)
 	stmt.SQL = fmt.Sprintf(statements.getObsByVariableAndContainedInPlace, stmt.SQL)
 	stmt.Params["ancestor"] = containedInPlace.Ancestor
 	stmt.Params["childPlaceType"] = containedInPlace.ChildPlaceType

internal/server/spanner/statements.go

@@ -43,12 +43,18 @@ var statements = struct {
 	applyOffset string
 	// Subquery to apply page limit.
 	applyLimit string
-	// Fetch all Observations.
-	getAllObs string
-	// Fetch latest Observations.
-	getLatestObs string
+	// Fetch Observations.
+	getObs string
 	// Fetch Observations for a specific date.
 	getDateObs string
+	// Subquery to return all Observations.
+	allObs string
+	// Subquery to return the latest Observations.
+	latestObs string
+	// Subquery to return Observations for a specific date.
+	dateObs string
+	// Subquery to return empty Observations.
+	emptyObs string
 	// Filter by variable dcids.
 	selectVariableDcids string
 	// Filter by entity dcids.
@@ -245,26 +251,11 @@ var statements = struct {
 	applyLimit: fmt.Sprintf(`
 		LIMIT %d
 	`, PAGE_SIZE+1),
-	getAllObs: `
+	getObs: `
 		SELECT
 			variable_measured,
 			observation_about,
-			observations,
-			provenance,
-			COALESCE(observation_period, '') AS observation_period,
-			COALESCE(measurement_method, '') AS measurement_method,
-			COALESCE(unit, '') AS unit,
-			COALESCE(scaling_factor, '') AS scaling_factor,
-			import_name,
-			provenance_url
-		FROM 
-			Observation
-	`,
-	getLatestObs: `
-		SELECT
-			variable_measured,
-			observation_about,
-			observations[ARRAY_LENGTH(observations)-1] AS observations,
+			%s,
 			provenance,
 			COALESCE(observation_period, '') AS observation_period,
 			COALESCE(measurement_method, '') AS measurement_method,
@@ -279,7 +270,7 @@ var statements = struct {
 		SELECT
 			variable_measured,
 			observation_about,
-			obs AS observations,
+			%s,
 			provenance,
 			COALESCE(observation_period, '') AS observation_period,
 			COALESCE(measurement_method, '') AS measurement_method,
@@ -291,6 +282,18 @@ var statements = struct {
 			Observation,
 			UNNEST(observations) as obs
 	`,
+	allObs: `
+		observations
+	`,
+	latestObs: `
+		[observations[ARRAY_LENGTH(observations)-1]] AS observations
+	`,
+	dateObs: `
+		[obs] AS observations
+	`,
+	emptyObs: `
+		ARRAY<JSON>[] AS observations
+	`,
 	selectVariableDcids: `
 		variable_measured IN UNNEST(@variables)
 	`,

internal/server/v3/observation/golden/observation_test.go

@@ -222,7 +222,7 @@ func TestV3Observation(t *testing.T) {
 			{
 				req: &pbv2.ObservationRequest{
 					Entity: &pbv2.DcidOrExpression{
-						Dcids: []string{"wikidataId/Q341968", "wikidataId/Q1764983"},
+						Dcids: []string{"wikidataId/Q4671576", "wikidataId/Q1764983"},
 					},
 					Select: []string{"entity", "variable", "date", "value"},
 				},