Switch V3 to use dc_graph_stable (#1568)

This switches Mixer to use dc_graph_stable, which has all the import
groups (except dcbranch)

This also updates the queries and goldens for the Node API (including
adding object_bytes) and Observation API

Note:
* The search index is still pending cherrypick release, so the
NodeSearch API will not work after this change
* This doesn't yet include the isDcAggregate field, so some of the facet
ids changed. As a follow up I can create a new build and then update
with this field
* This build includes the nodes that are missing name/types, so some of
them are dropped in the goldens. After they're properly merged/resolved
I can create a new build
* Since querying within maps is till unclear, for now I just removed
date filtering and am returning the full time series

Author: n-h-diaz

deploy/storage/spanner_graph_info.yaml

@@ -1,3 +1,3 @@
 project: datcom-store
 instance: dc-kg-test
-database: dc_graph_4
+database: dc_graph_stable

internal/proto/storage.pb.go

@@ -103,25 +103,24 @@ 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, filterObs)
+		observations, err = sds.client.GetObservationsContainedInPlace(ctx, variables, containedInPlace)
 		if err != nil {
 			return nil, fmt.Errorf("error getting observations (contained in): %v", err)
 		}
 	} else {
-		observations, err = sds.client.GetObservations(ctx, variables, entities, date, filterObs)
+		observations, err = sds.client.GetObservations(ctx, variables, entities)
 		if err != nil {
 			return nil, fmt.Errorf("error getting observations: %v", err)
 		}
 	}
 
-	observations = filterObservationsByFacet(observations, req.Filter)
+	observations = filterObservationsByDateAndFacet(observations, date, req.Filter)
 
 	return observationsToObservationResponse(req, observations), nil
 }

internal/server/spanner/datasource.go

@@ -165,15 +165,19 @@ func nodeEdgesToNodeResponse(nodes []string, edgesBySubjectID map[string][]*Edge
 			nodeResponse.NextToken = nextToken
 		}
 
-		nodeResponse.Data[subjectID] = nodeEdgesToLinkedGraph(edges)
+		linkedGraph, err := nodeEdgesToLinkedGraph(edges)
+		if err != nil {
+			return nil, err
+		}
+		nodeResponse.Data[subjectID] = linkedGraph
 	}
 
 	return nodeResponse, nil
 }
 
 // nodeEdgesToLinkedGraph converts an array of edges to a LinkedGraph proto.
 // This method assumes all edges are from the same entity.
-func nodeEdgesToLinkedGraph(edges []*Edge) *pbv2.LinkedGraph {
+func nodeEdgesToLinkedGraph(edges []*Edge) (*pbv2.LinkedGraph, error) {
 	linkedGraph := &pbv2.LinkedGraph{
 		Arcs: make(map[string]*pbv2.Nodes),
 	}
@@ -190,12 +194,22 @@ func nodeEdgesToLinkedGraph(edges []*Edge) *pbv2.LinkedGraph {
 			ProvenanceId: edge.Provenance,
 			Value:        edge.ObjectValue,
 		}
+
+		// Use object_bytes if set.
+		if edge.ObjectBytes != nil {
+			bytes, err := util.Unzip(edge.ObjectBytes)
+			if err != nil {
+				return nil, err
+			}
+			node.Value = string(bytes)
+		}
+
 		nodes.Nodes = append(nodes.Nodes, node)
 
 		linkedGraph.Arcs[edge.Predicate] = nodes
 	}
 
-	return linkedGraph
+	return linkedGraph, nil
 }
 
 func selectFieldsToQueryOptions(selectFields []string) queryOptions {
@@ -221,57 +235,50 @@ func isObservationRequest(qo *queryOptions) bool {
 	return qo.date && qo.value
 }
 
-// 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 {
+func buildBaseObsStatement(variables []string, entities []string) spanner.Statement {
 	stmt := spanner.Statement{
+		SQL:    statements.getObs,
 		Params: map[string]interface{}{},
 	}
-	filters := []string{}
-
-	var baseStmt string
-	var obsStmt string
-	switch date {
-	case "":
-		baseStmt = statements.getObs
-		obsStmt = statements.allObs
-	case shared.LATEST:
-		baseStmt = statements.getObs
-		obsStmt = statements.latestObs
-	default:
-		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)
 
+	filters := []string{}
 	if len(variables) > 0 {
 		stmt.Params["variables"] = variables
 		filters = append(filters, statements.selectVariableDcids)
 	}
-
 	if len(entities) > 0 {
 		stmt.Params["entities"] = entities
 		filters = append(filters, statements.selectEntityDcids)
 	}
-
 	stmt.SQL += WHERE + strings.Join(filters, AND)
 
 	return stmt
 }
 
-func filterObservationsByFacet(observations []*Observation, filter *pbv2.FacetFilter) []*Observation {
+func filterTimeSeriesByDate(ts *TimeSeries, date string) {
+	switch date {
+	case "":
+	case shared.LATEST:
+		if ts == nil || *ts == nil || len(*ts) == 0 {
+			*ts = TimeSeries{}
+		} else {
+			*ts = TimeSeries{(*ts)[len(*ts)-1]}
+		}
+	default:
+		for _, dv := range *ts {
+			if dv.Date == date {
+				*ts = TimeSeries{dv}
+				return
+			}
+		}
+		*ts = TimeSeries{}
+	}
+}
+
+func filterObservationsByDateAndFacet(observations []*Observation, date string, filter *pbv2.FacetFilter) []*Observation {
 	var filtered []*Observation
 	for _, observation := range observations {
+		filterTimeSeriesByDate(&observation.Observations, date)
 		facet := observationToFacet(observation)
 		if util.ShouldIncludeFacet(filter, facet) {
 			filtered = append(filtered, observation)
@@ -496,7 +503,6 @@ func observationToFacetObservation(observation *Observation, includeObs bool) (*
 	facet := observationToFacet(observation)
 
 	var observations []*pb.PointStat
-
 	for _, dateValue := range observation.Observations {
 		pointStat, err := dateValueToPointStat(dateValue)
 

internal/server/spanner/dsutil.go

@@ -1,10 +1,23 @@
 {
   "FireIncidentTypeEnum": [
+    {
+      "SubjectID": "FireIncidentTypeEnum",
+      "Predicate": "rangeIncludes",
+      "ObjectID": "classification",
+      "ObjectValue": "",
+      "ObjectBytes": null,
+      "Provenance": "dc/base/BaseSchema",
+      "Name": "classification",
+      "Types": [
+        "Property"
+      ]
+    },
     {
       "SubjectID": "FireIncidentTypeEnum",
       "Predicate": "typeOf",
       "ObjectID": "ComplexFire",
       "ObjectValue": "",
+      "ObjectBytes": null,
       "Provenance": "dc/base/BaseSchema",
       "Name": "ComplexFire",
       "Types": [
@@ -16,6 +29,7 @@
       "Predicate": "typeOf",
       "ObjectID": "PrescribedFire",
       "ObjectValue": "",
+      "ObjectBytes": null,
       "Provenance": "dc/base/BaseSchema",
       "Name": "PrescribedFire",
       "Types": [
@@ -30,6 +44,7 @@
       "Predicate": "typeOf",
       "ObjectID": "Wildfire",
       "ObjectValue": "",
+      "ObjectBytes": null,
       "Provenance": "dc/base/BaseSchema",
       "Name": "Wildfire",
       "Types": [
@@ -42,6 +57,7 @@
       "Predicate": "typeOf",
       "ObjectID": "WildlandFireUse",
       "ObjectValue": "",
+      "ObjectBytes": null,
       "Provenance": "dc/base/BaseSchema",
       "Name": "WildlandFireUse",
       "Types": [
@@ -50,11 +66,24 @@
     }
   ],
   "FoodTypeEnum": [
+    {
+      "SubjectID": "FoodTypeEnum",
+      "Predicate": "rangeIncludes",
+      "ObjectID": "foodType",
+      "ObjectValue": "",
+      "ObjectBytes": null,
+      "Provenance": "dc/base/BaseSchema",
+      "Name": "foodType",
+      "Types": [
+        "Property"
+      ]
+    },
     {
       "SubjectID": "FoodTypeEnum",
       "Predicate": "typeOf",
       "ObjectID": "Meal",
       "ObjectValue": "",
+      "ObjectBytes": null,
       "Provenance": "dc/base/BaseSchema",
       "Name": "Meal",
       "Types": [