[WIP] Feature/server func

conformance/tests/ot_serverfunc.py

@@ -0,0 +1,24 @@
+
+attr_len_func = """
+function attr_len(x) {
+    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]
+}
+"""
+
+def test_flatmap(man):
+    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

engine/core/processors_func.go

@@ -0,0 +1,67 @@
+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
+}
+
+// Process LookupVerts
+func (fm *FlatMap) Process(ctx context.Context, man gdbi.Manager, in gdbi.InPipe, out gdbi.OutPipe) context.Context {
+
+	vm := goja.New()
+	_, 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)
+				fout, err := jfunc(goja.Null(), dataObj)
+				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

@@ -221,6 +221,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}, nil
+}
+
 func (sc *DefaultStmtCompiler) Custom(gs *gripql.GraphStatement, ps *gdbi.State) (gdbi.Processor, error) {
 
 	switch stmt := gs.GetStatement().(type) {

gdbi/pipeline.go

@@ -81,5 +81,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

@@ -216,6 +216,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())