Improve subgraph performance with persistent backends

Do just one call to get all the edges for the given nodes, instead of
one call for each node.

This new call has been added as a new method to Graph and Backends.
GetNodesEdges, return the list with all edges for a list of nodes.

Batching is used to avoid hitting the max number of clauses set by ES
(is set to the default value of 512).

Author: adrianlzt

graffiti/graph/cachedbackend.go

@@ -110,6 +110,17 @@ func (c *CachedBackend) GetNodeEdges(n *Node, t Context, m ElementMatcher) (edge
 	return c.persistent.GetNodeEdges(n, t, m)
 }
 
+// GetNodesEdges return the list of all edges for a list of nodes within time slice, matching metadata
+func (c *CachedBackend) GetNodesEdges(n []*Node, t Context, m ElementMatcher) (edges []*Edge) {
+	mode := c.cacheMode.Load()
+
+	if t.TimeSlice == nil || mode == CacheOnlyMode || c.persistent == nil {
+		return c.memory.GetNodesEdges(n, t, m)
+	}
+
+	return c.persistent.GetNodesEdges(n, t, m)
+}
+
 // EdgeAdded add an edge in the cache
 func (c *CachedBackend) EdgeAdded(e *Edge) error {
 	mode := c.cacheMode.Load()

graffiti/graph/elasticsearch.go

@@ -87,6 +87,8 @@ const graphElementMapping = `
 const (
 	nodeType = "node"
 	edgeType = "edge"
+	// maxClauseCount limit the number of clauses in one query to ES
+	maxClauseCount = 512
 )
 
 // ElasticSearchBackend describes a persistent backend based on ElasticSearch
@@ -506,6 +508,53 @@ func (b *ElasticSearchBackend) GetNodeEdges(n *Node, t Context, m ElementMatcher
 	return
 }
 
+// GetNodesEdges return the list of all edges for a list of nodes within time slice
+func (b *ElasticSearchBackend) GetNodesEdges(nodeList []*Node, t Context, m ElementMatcher) (edges []*Edge) {
+	if len(nodeList) == 0 {
+		return []*Edge{}
+	}
+
+	// See comment at GetNodesFromIDs
+	// As we are adding two operations per item, make small batches
+	nodesBatch := batchNodes(nodeList, maxClauseCount/2)
+
+	for _, nList := range nodesBatch {
+		var filter *filters.Filter
+		if m != nil {
+			f, err := m.Filter()
+			if err != nil {
+				return []*Edge{}
+			}
+			filter = f
+		}
+
+		var searchQuery filters.SearchQuery
+		if !t.TimePoint {
+			searchQuery = filters.SearchQuery{Sort: true, SortBy: "UpdatedAt"}
+		}
+
+		nodesFilter := []*filters.Filter{}
+		for _, n := range nList {
+			nodesFilter = append(nodesFilter, filters.NewTermStringFilter("Parent", string(n.ID)))
+			nodesFilter = append(nodesFilter, filters.NewTermStringFilter("Child", string(n.ID)))
+		}
+		searchQuery.Filter = filters.NewOrFilter(nodesFilter...)
+
+		edges = append(edges, b.searchEdges(&TimedSearchQuery{
+			SearchQuery:   searchQuery,
+			TimeFilter:    getTimeFilter(t.TimeSlice),
+			ElementFilter: filter,
+		})...)
+
+	}
+
+	if len(edges) > 1 && t.TimePoint {
+		edges = dedupEdges(edges)
+	}
+
+	return
+}
+
 // IsHistorySupported returns that this backend does support history
 func (b *ElasticSearchBackend) IsHistorySupported() bool {
 	return true
@@ -647,3 +696,15 @@ func NewElasticSearchBackendFromConfig(cfg es.Config, extraDynamicTemplates map[
 
 	return newElasticSearchBackendFromClient(client, cfg.IndexPrefix, liveIndex, archiveIndex, logger), nil
 }
+
+func batchNodes(items []*Node, batchSize int) [][]*Node {
+	batches := make([][]*Node, 0, (len(items)+batchSize-1)/batchSize)
+
+	for batchSize < len(items) {
+		items, batches = items[batchSize:], append(batches, items[0:batchSize:batchSize])
+	}
+	batches = append(batches, items)
+
+	return batches
+
+}

graffiti/graph/graph.go

@@ -112,6 +112,7 @@ type Backend interface {
 	NodeDeleted(n *Node) error
 	GetNode(i Identifier, at Context) []*Node
 	GetNodeEdges(n *Node, at Context, m ElementMatcher) []*Edge
+	GetNodesEdges(n []*Node, at Context, m ElementMatcher) []*Edge
 
 	EdgeAdded(e *Edge) error
 	EdgeDeleted(e *Edge) error
@@ -1365,6 +1366,14 @@ func (g *Graph) GetNodeEdges(n *Node, m ElementMatcher) []*Edge {
 	return g.backend.GetNodeEdges(n, g.context, m)
 }
 
+// GetNodesEdges returns the list with all edges for a list of nodes
+func (g *Graph) GetNodesEdges(n []*Node, m ElementMatcher) []*Edge {
+	if len(n) == 0 {
+		return []*Edge{}
+	}
+	return g.backend.GetNodesEdges(n, g.context, m)
+}
+
 func (g *Graph) String() string {
 	j, _ := json.Marshal(g)
 	return string(j)

graffiti/graph/memory.go

@@ -153,6 +153,22 @@ func (m *MemoryBackend) GetNodeEdges(n *Node, t Context, meta ElementMatcher) []
 	return edges
 }
 
+// GetNodesEdges returns the list of edges for a list of nodes
+func (m *MemoryBackend) GetNodesEdges(nodeList []*Node, t Context, meta ElementMatcher) []*Edge {
+	edges := []*Edge{}
+	for _, n := range nodeList {
+		if n, ok := m.nodes[n.ID]; ok {
+			for _, e := range n.edges {
+				if e.MatchMetadata(meta) {
+					edges = append(edges, e.Edge)
+				}
+			}
+		}
+	}
+
+	return edges
+}
+
 // EdgeDeleted in the graph backend
 func (m *MemoryBackend) EdgeDeleted(e *Edge) error {
 	if _, ok := m.edges[e.ID]; !ok {

graffiti/graph/orientdb.go

@@ -232,6 +232,23 @@ func (o *OrientDBBackend) GetNodeEdges(n *Node, t Context, m ElementMatcher) (ed
 	return o.searchEdges(t, query)
 }
 
+// GetNodesEdges returns a list of a node edges within time slice
+func (o *OrientDBBackend) GetNodesEdges(nodeList []*Node, t Context, m ElementMatcher) (edges []*Edge) {
+	query := orientdb.FilterToExpression(getTimeFilter(t.TimeSlice), nil)
+	query += fmt.Sprintf(" AND (")
+	for i, n := range nodeList {
+		if i == len(nodeList)-1 {
+			query += fmt.Sprintf(" Parent = '%s' OR Child = '%s')", n.ID, n.ID)
+		} else {
+			query += fmt.Sprintf(" Parent = '%s' OR Child = '%s' OR", n.ID, n.ID)
+		}
+	}
+	if matcherQuery := matcherToOrientDBSelectString(m); matcherQuery != "" {
+		query += " AND " + matcherQuery
+	}
+	return o.searchEdges(t, query)
+}
+
 func (o *OrientDBBackend) createEdge(e *Edge) error {
 	fromQuery := fmt.Sprintf("SELECT FROM Node WHERE DeletedAt IS NULL AND ArchivedAt IS NULL AND ID = '%s'", e.Parent)
 	toQuery := fmt.Sprintf("SELECT FROM Node WHERE DeletedAt IS NULL AND ArchivedAt IS NULL AND ID = '%s'", e.Child)

graffiti/graph/traversal/traversal.go

@@ -1415,14 +1415,12 @@ func (tv *GraphTraversalV) SubGraph(ctx StepContext, s ...interface{}) *GraphTra
 
 	// then insert edges, ignore edge insert error since one of the linked node couldn't be part
 	// of the SubGraph
-	for _, n := range tv.nodes {
-		edges := tv.GraphTraversal.Graph.GetNodeEdges(n, nil)
-		for _, e := range edges {
-			switch err := memory.EdgeAdded(e); err {
-			case nil, graph.ErrParentNotFound, graph.ErrChildNotFound, graph.ErrEdgeConflict:
-			default:
-				return &GraphTraversal{error: fmt.Errorf("Error while adding edge to SubGraph: %s", err)}
-			}
+	edges := tv.GraphTraversal.Graph.GetNodesEdges(tv.nodes, nil)
+	for _, e := range edges {
+		switch err := memory.EdgeAdded(e); err {
+		case nil, graph.ErrParentNotFound, graph.ErrChildNotFound, graph.ErrEdgeConflict:
+		default:
+			return &GraphTraversal{error: fmt.Errorf("Error while adding edge to SubGraph: %s", err)}
 		}
 	}