[WIP] Feature/server func

cmd/query/main.go

@@ -48,6 +48,7 @@ Example:
 		}
 
 		log.Debugf("Query: %s\n", query.String())
+
 		res, err := conn.Traversal(context.Background(), query)
 		if err != nil {
 			log.Errorf("Traversal error: %s", err)

conformance/tests/ot_serverfunc.py

@@ -0,0 +1,42 @@
+def test_flatmap(man):
+    attr_len_func = """
+    function attr_len(x, args) {
+        x["skin_colors_len"] = x["skin_colors"].length;
+        x["hair_colors_len"] = x["hair_colors"].length;
+        x["eye_colors_len"]  = x["eye_colors"].length;
+        return [x]
+    }
+    """
+
+    errors = []
+
+    G = man.setGraph("swapi")
+
+    count = 0
+    q = G.query().V().hasLabel("Species").flatMap("attr_len", attr_len_func, {})
+    for row in q:
+        count += 1
+        if row["data"]["skin_colors_len"] != len(row["data"]["skin_colors"]):
+            errors.append("count function not correct")
+    if count != 5:
+        errors.append("Incorrect row count returned: %d != 5" % (count))
+    return errors
+
+
+def test_command_line_args(man):
+    func = """
+    function AddObj(x, args) {
+        for (var k in args){
+            x[k]= args[k]
+        }
+        return [x]
+    }
+    """
+
+    errors = []
+    G = man.setGraph("swapi")
+    q = G.query().V().hasLabel("Species").flatMap("AddObj", func, {"OtherSpecies":["Sullustan", "RathStar",  "Bith"]})
+    for row in q:
+        if "OtherSpecies" not in row['data'] or not all(species in row['data']["OtherSpecies"] for species in ["Sullustan", "RathStar", "Bith"]):
+            errors.append("Appended items '{'OtherSpecies':['Sullustan', 'RathStar',  'Bith']}' Not Present in row")
+    return errors

engine/core/processors_func.go

@@ -0,0 +1,85 @@
+package core
+
+import (
+	"context"
+
+	"github.com/bmeg/grip/gdbi"
+	"github.com/bmeg/grip/log"
+	"github.com/dop251/goja"
+)
+
+// LookupVerts starts query by looking on vertices
+type FlatMap struct {
+	Source    string
+	Func_name string
+	Args      map[string]any
+}
+
+// Process LookupVerts
+func (fm *FlatMap) Process(ctx context.Context, man gdbi.Manager, in gdbi.InPipe, out gdbi.OutPipe) context.Context {
+
+	vm := goja.New()
+
+	// Add support for console.log for debugging purposes
+	console := vm.NewObject()
+	_ = console.Set("log", func(call goja.FunctionCall) goja.Value {
+		args := make([]any, len(call.Arguments))
+		for i, arg := range call.Arguments {
+			args[i] = arg.Export()
+		}
+		log.Infof("JS log: %v", args...)
+		return goja.Undefined()
+	})
+	_ = vm.Set("console", console)
+
+	_, err := vm.RunString(fm.Source)
+	if err != nil {
+		log.Errorf("User function compile error: %s", err)
+	}
+
+	jobj := vm.Get(fm.Func_name)
+	if jobj == nil {
+		log.Errorf("User Function not found: %s", fm.Func_name)
+	}
+
+	jfunc, ok := goja.AssertFunction(jobj)
+	if !ok {
+		log.Errorf("Defined object not function: %#v", jobj)
+	}
+	go func() {
+		defer close(out)
+		for t := range in {
+			if t.IsSignal() {
+				out <- t
+				continue
+			}
+			if jfunc != nil {
+				src := t.GetCurrent().Get()
+				data := src.ToDict()
+				dataObj := vm.ToValue(data)
+
+				argsObj := vm.ToValue(fm.Args)
+
+				fout, err := jfunc(goja.Null(), dataObj, argsObj)
+
+				if err == nil {
+					o := fout.Export()
+					if oList, ok := o.([]any); ok {
+						for _, od := range oList {
+							if om, ok := od.(map[string]any); ok {
+								d := gdbi.DataElement{}
+								d.FromDict(om)
+								d.ID = src.ID
+								d.Label = src.Label
+								out <- t.AddCurrent(&d)
+							}
+						}
+					}
+				} else {
+					log.Errorf("Function error: %s", err)
+				}
+			}
+		}
+	}()
+	return ctx
+}

engine/core/statement_compiler.go

@@ -225,6 +225,10 @@ func (sc *DefaultStmtCompiler) Aggregate(stmt *gripql.GraphStatement_Aggregate,
 	return &aggregate{stmt.Aggregate.Aggregations}, nil
 }
 
+func (sc *DefaultStmtCompiler) FlatMap(stmt *gripql.GraphStatement_FlatMap, ps *gdbi.State) (gdbi.Processor, error) {
+	return &FlatMap{Source: stmt.FlatMap.Source, Func_name: stmt.FlatMap.Function, Args: stmt.FlatMap.Args.AsMap()}, nil
+}
+
 func (sc *DefaultStmtCompiler) Custom(gs *gripql.GraphStatement, ps *gdbi.State) (gdbi.Processor, error) {
 
 	switch stmt := gs.GetStatement().(type) {

gdbi/pipeline.go

@@ -83,5 +83,7 @@ type StatementCompiler interface {
 	Fields(gs *gripql.GraphStatement_Fields, ps *State) (Processor, error)
 	Aggregate(gs *gripql.GraphStatement_Aggregate, ps *State) (Processor, error)
 
+	FlatMap(gs *gripql.GraphStatement_FlatMap, ps *State) (Processor, error)
+
 	Custom(gs *gripql.GraphStatement, ps *State) (Processor, error)
 }

gdbi/statement_processor.go

@@ -224,6 +224,9 @@ func StatementProcessor(
 	case *gripql.GraphStatement_Unwind:
 		return sc.Unwind(stmt, ps)
 
+	case *gripql.GraphStatement_FlatMap:
+		return sc.FlatMap(stmt, ps)
+
 	case *gripql.GraphStatement_Fields:
 		if ps.LastType != VertexData && ps.LastType != EdgeData {
 			return nil, fmt.Errorf(`"fields" statement is only valid for edge or vertex types not: %s`, ps.LastType.String())