使用Go+Gremlingo在AWS Neptune执行Gremlin Upsert查询报错排查

错误原因分析 #

错误信息中的The provided traverser does not map to a value: v[...]->[SelectOneStep(last,e,null)]明确指向AddE步骤的逻辑问题：

• 你调用AddE("ATTACHED_TO").To("e")时，将字符串"e"直接传给了To()方法。在Gremlin语法中，字符串参数会被解析为顶点ID，而非之前通过As("e")标记的顶点别名。
• 系统尝试查找ID为"e"的顶点，自然无法匹配，因此抛出InvalidParameterException。

修复方案 #

将To("e")修改为To(anonT.Select("e"))，通过Select("e")遍历器引用之前标记的Entity顶点，而非把"e"当作顶点ID。

修改后的完整代码片段：

func (n NeptuneGremlinGraph) Put(assetID string, version string, records []les.DeltaEditRecord) error {
    g := gremlin.Traversal_().WithRemote(n.connection)
    anonT := gremlin.T__
    for _, r := range records {
        promise := g.V().HasLabel("Entity").
            Has("asset_id", assetID).
            Has("version", version).
            Has("entity_id", r.EntityID).
            Fold().
            Coalesce(anonT.Unfold(),
                anonT.AddV("Entity").
                    Property("asset_id", assetID).
                    Property("version", version).
                    Property("entity_id", r.EntityID)).
            As("e").
            V().HasLabel("Component").
            Has("asset_id", assetID).
            Has("version", version).
            Has("entity_id", r.EntityID).
            Has("component_id", r.ComponentID).
            Fold().
            Coalesce(anonT.Unfold().
                Property("value", r.Value),
                anonT.AddV("Component").
                    Property("asset_id", assetID).
                    Property("version", version).
                    Property("entity_id", r.EntityID).
                    Property("component_id", r.ComponentID).
                    Property("value", r.Value)).
            // 关键修改：用Select("e")引用别名顶点，而非直接传字符串"e"
            AddE("ATTACHED_TO").To(anonT.Select("e")).Iterate()
        err := <-promise
        if err != nil {
            return err
        }
    }

    return nil
}

可选优化（提升查询效率） #

原代码两次调用V()遍历全图，可通过SideEffect将Component的upsert逻辑嵌套在Entity的遍历流中，减少一次全图扫描：

func (n NeptuneGremlinGraph) Put(assetID string, version string, records []les.DeltaEditRecord) error {
    g := gremlin.Traversal_().WithRemote(n.connection)
    anonT := gremlin.T__
    for _, r := range records {
        promise := g.V().HasLabel("Entity").
            Has("asset_id", assetID).
            Has("version", version).
            Has("entity_id", r.EntityID).
            Fold().
            Coalesce(anonT.Unfold(),
                anonT.AddV("Entity").
                    Property("asset_id", assetID).
                    Property("version", version).
                    Property("entity_id", r.EntityID)).
            // 用SideEffect复用当前Entity顶点的上下文
            SideEffect(func(entityTraversal gremlin.Traversal) gremlin.Traversal {
                return entityTraversal.V().HasLabel("Component").
                    Has("asset_id", assetID).
                    Has("version", version).
                    Has("entity_id", r.EntityID).
                    Has("component_id", r.ComponentID).
                    Fold().
                    Coalesce(anonT.Unfold().Property("value", r.Value),
                        anonT.AddV("Component").
                            Property("asset_id", assetID).
                            Property("version", version).
                            Property("entity_id", r.EntityID).
                            Property("component_id", r.ComponentID).
                            Property("value", r.Value)).
                    AddE("ATTACHED_TO").To(entityTraversal)
            }).Iterate()
        err := <-promise
        if err != nil {
            return err
        }
    }

    return nil
}

内容的提问来源于stack exchange，提问作者Michael Burbidge