Merge branch 'main' into aman/telemetry

Author: Harshil Goel

dql/parser.go

@@ -2691,7 +2691,7 @@ func validKeyAtRoot(k string) bool {
 	switch k {
 	case "func", "orderasc", "orderdesc", "first", "offset", "after":
 		return true
-	case "from", "to", "numpaths", "minweight", "maxweight":
+	case "from", "to", "numpaths", "minweight", "maxweight", "maxfrontiersize":
 		// Specific to shortest path
 		return true
 	case "depth":

dql/parser_test.go

@@ -1302,7 +1302,7 @@ func TestParseQueryWithMultipleVar(t *testing.T) {
 func TestParseShortestPath(t *testing.T) {
 	query := `
 	{
-		shortest(from:0x0a, to:0x0b, numpaths: 3, minweight: 3, maxweight: 6) {
+		shortest(from:0x0a, to:0x0b, numpaths: 3, minweight: 3, maxweight: 6, maxfrontiersize: 1) {
 			friends
 			name
 		}
@@ -1317,6 +1317,7 @@ func TestParseShortestPath(t *testing.T) {
 	require.Equal(t, "3", res.Query[0].Args["numpaths"])
 	require.Equal(t, "3", res.Query[0].Args["minweight"])
 	require.Equal(t, "6", res.Query[0].Args["maxweight"])
+	require.Equal(t, "1", res.Query[0].Args["maxfrontiersize"])
 }
 
 func TestParseShortestPathWithUidVars(t *testing.T) {

query/query.go

@@ -165,6 +165,10 @@ type params struct {
 	MaxWeight float64
 	// MinWeight is the min weight allowed in a path returned by the shortest path algorithm.
 	MinWeight float64
+	// MaxFrontierSize limits the number of candidate paths stored in the priority queue.
+	// During shortest path computation. This prevents out-of-memory errors on large graphs
+	// but may affect solution optimality if set too low.
+	MaxFrontierSize int64
 
 	// ExploreDepth is used by recurse and shortest path queries to specify the maximum graph
 	// depth to explore.
@@ -714,6 +718,16 @@ func (args *params) fill(gq *dql.GraphQuery) error {
 			args.MinWeight = -math.MaxFloat64
 		}
 
+		if v, ok := gq.Args["maxfrontiersize"]; ok {
+			maxfrontiersize, err := strconv.ParseInt(v, 0, 64)
+			if err != nil {
+				return err
+			}
+			args.MaxFrontierSize = maxfrontiersize
+		} else if !ok {
+			args.MaxFrontierSize = math.MaxInt64
+		}
+
 		if gq.ShortestPathArgs.From == nil || gq.ShortestPathArgs.To == nil {
 			return errors.Errorf("from/to can't be nil for shortest path")
 		}
@@ -2640,7 +2654,7 @@ func (sg *SubGraph) sortAndPaginateUsingVar(ctx context.Context) error {
 func isValidArg(a string) bool {
 	switch a {
 	case "numpaths", "from", "to", "orderasc", "orderdesc", "first", "offset", "after", "depth",
-		"minweight", "maxweight":
+		"minweight", "maxweight", "maxfrontiersize":
 		return true
 	}
 	return false

query/shortest.go

@@ -405,6 +405,9 @@ func runKShortestPaths(ctx context.Context, sg *SubGraph) ([]*SubGraph, error) {
 				hop:  item.hop + 1,
 				path: route{route: curPath},
 			}
+			if int64(pq.Len()) > sg.Params.MaxFrontierSize {
+				pq.Pop()
+			}
 			heap.Push(&pq, node)
 		}
 		// Return the popped nodes path to pool.
@@ -558,6 +561,9 @@ func shortestPath(ctx context.Context, sg *SubGraph) ([]*SubGraph, error) {
 					cost: nodeCost,
 					hop:  item.hop + 1,
 				}
+				if int64(pq.Len()) > sg.Params.MaxFrontierSize {
+					pq.Pop()
+				}
 				heap.Push(&pq, node)
 			} else {
 				// We've already seen this node. So, just update the cost

systest/shortest-path/graph.rdf.gz

@@ -0,0 +1,61 @@
+//go:build integration2
+
+/*
+ * SPDX-FileCopyrightText: © Hypermode Inc. <[email protected]>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hypermodeinc/dgraph/v25/dgraphapi"
+	"github.com/hypermodeinc/dgraph/v25/dgraphtest"
+	"github.com/hypermodeinc/dgraph/v25/x"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestShortestPath(t *testing.T) {
+	conf := dgraphtest.NewClusterConfig().WithNumAlphas(1).WithNumZeros(1).WithReplicas(1).WithACL(time.Hour)
+	c, err := dgraphtest.NewLocalCluster(conf)
+	require.NoError(t, err)
+	defer func() { c.Cleanup(t.Failed()) }()
+	require.NoError(t, c.Start())
+
+	err = c.LiveLoad(dgraphtest.LiveOpts{
+		DataFiles:      []string{"graph.rdf.gz"},
+		SchemaFiles:    []string{"graph.schema.gz"},
+		GqlSchemaFiles: []string{},
+	})
+	require.NoError(t, err)
+
+	gc, cleanup, err := c.Client()
+	require.NoError(t, err)
+	defer cleanup()
+	require.NoError(t, gc.LoginIntoNamespace(context.Background(),
+		dgraphapi.DefaultUser, dgraphapi.DefaultPassword, x.GalaxyNamespace))
+
+	_, err = gc.Query(`
+	{
+		q(func: eq(guid, "85270d10-560e-4cc8-8703-4b4c563a2f4e")) {
+		 	a as uid
+  		}
+  		q1(func: eq(guid, "4a520068-80b6-42f2-9019-4e6ef8a02bb3")) {
+			b as uid
+  		}
+  
+ 		path as shortest(from: uid(a), to: uid(b), numpaths: 5, maxfrontiersize: 10000) {
+  			connected_to @facets(weight)
+ 		}
+
+ 		path(func: uid(path)) {
+   			uid
+ 		}
+	}
+	`)
+	require.NoError(t, err)
+}

systest/shortest-path/graph.schema.gz

@@ -1819,17 +1819,17 @@ func langForFunc(langs []string) string {
 	return langs[0]
 }
 
-func planForEqFilter(fc *functionContext, pred string, uidlist []uint64) {
-	checkUidEmpty := func(uids []uint64) bool {
-		for _, i := range uids {
-			if i == 0 {
-				return false
-			}
+func checkUidZero(uids []uint64) bool {
+	for _, uid := range uids {
+		if uid == 0 {
+			return true
 		}
-		return true
 	}
+	return false
+}
 
-	if !checkUidEmpty(uidlist) {
+func planForEqFilter(fc *functionContext, pred string, uidlist []uint64) {
+	if checkUidZero(uidlist) {
 		// We have a uid which has 0 in it. Mostly it would happen when there is only 0. But any one item
 		// being 0 could cause the query planner to fail. In case of 0 being present, we neeed to query the
 		// index itself.
@@ -1912,6 +1912,14 @@ func parseSrcFn(ctx context.Context, q *pb.Query) (*functionContext, error) {
 			}
 		}
 
+		generateIneqTokens := true
+		if fc.fname != eq && q.UidList != nil && uint64(len(q.UidList.Uids)) < Config.TypeFilterUidLimit {
+			if !checkUidZero(q.UidList.Uids) {
+				fc.n = len(q.UidList.Uids)
+				generateIneqTokens = false
+			}
+		}
+
 		var tokens []string
 		var ineqValues []types.Val
 		// eq can have multiple args.
@@ -1947,6 +1955,9 @@ func parseSrcFn(ctx context.Context, q *pb.Query) (*functionContext, error) {
 				lang = q.Langs[0]
 			}
 
+			if !generateIneqTokens {
+				continue
+			}
 			// Get tokens ge/le ineqValueToken.
 			if tokens, fc.ineqValueToken, err = getInequalityTokens(ctx, q.ReadTs, attr, f, lang,
 				ineqValues); err != nil {
@@ -1958,6 +1969,10 @@ func parseSrcFn(ctx context.Context, q *pb.Query) (*functionContext, error) {
 			fc.tokens = append(fc.tokens, tokens...)
 		}
 
+		if !generateIneqTokens {
+			return fc, nil
+		}
+
 		// In case of non-indexed predicate, there won't be any tokens. We will fetch value
 		// from data keys.
 		// If number of index keys is more than no. of uids to filter, so its better to fetch values