[NOID] Fixes #4157: apoc.dv.queryAndLink how to specify direction (#4222)

* Fixes #4157: apoc.dv.queryAndLink how to specify direction

* test fixes

* refactoring

Author: vga91

docs/asciidoc/modules/ROOT/pages/overview/apoc.dv/apoc.dv.queryAndLink.adoc

@@ -35,3 +35,13 @@ apoc.dv.queryAndLink(node :: NODE?, relName :: STRING?, name :: STRING?, params
 |path|PATH?
 |===
 
+== Configuration parameters
+
+The procedures support the following config parameters:
+
+.Config parameters
+[opts=header]
+|===
+| name | type | default | description
+| direction | String | "OUT" | The direction of the relationships, i.e. outgoing ("OUT") or incoming ("IN").
+|===

docs/asciidoc/modules/ROOT/pages/virtual-resource/index.adoc

@@ -113,6 +113,16 @@ phrase over the previous query.
 
 image::apoc.dv.jdbc-queryAndLink.png[scaledwidth="100%"]
 
+The default direction of the relationships is outgoing (i.e. `{direction: "OUT"}`), but it is possible to reverse it by the config parameters.
+Example:
+
+[source,cypher]
+----
+MATCH (hook:Hook) WITH hook
+CALL apoc.dv.queryAndLink(hook, $relType, $name, $queryParams, { direction: "IN" }) yield path
+RETURN path
+----
+
 === Listing the Virtualized Resource Catalog
 The apoc.dv.catalog.list procedure returns a list with all the existing Virtualized resources and their descriptions. It takes no parameters.
 

full/src/main/java/apoc/dv/DataVirtualizationCatalog.java

@@ -25,6 +25,7 @@
 import apoc.result.VirtualPath;
 import apoc.result.VirtualRelationship;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Stream;
 import org.neo4j.graphdb.GraphDatabaseService;
 import org.neo4j.graphdb.Node;
@@ -103,10 +104,30 @@ public Stream<PathResult> queryAndLink(
         VirtualizedResource vr = new DataVirtualizationCatalogHandler(db, apocConfig.getSystemDb(), null).get(name);
         final RelationshipType relationshipType = RelationshipType.withName(relName);
         final Pair<String, Map<String, Object>> procedureCallWithParams = vr.getProcedureCallWithParams(params, config);
+
+        String direction = (String) config.getOrDefault("direction", Direction.OUT.name());
+
         return tx.execute(procedureCallWithParams.first(), procedureCallWithParams.other()).stream()
                 .map(m -> (Node) m.get(("node")))
-                .map(n -> new VirtualRelationship(node, n, relationshipType))
-                .map(r -> new VirtualPath.Builder(r.getStartNode()).push(r).build())
+                .map(n -> getVirtualRelationship(node, n, direction, relationshipType))
+                .map(r -> {
+                    VirtualPath virtualPath = new VirtualPath(r.getStartNode());
+                    virtualPath.addRel(r);
+                    return virtualPath;
+                })
                 .map(PathResult::new);
     }
+
+    private VirtualRelationship getVirtualRelationship(
+            Node node, Node n, String direction, RelationshipType relationshipType) {
+        if (Objects.equals(direction.toUpperCase(), Direction.OUT.name())) {
+            return new VirtualRelationship(node, n, relationshipType);
+        }
+        return new VirtualRelationship(n, node, relationshipType);
+    }
+
+    enum Direction {
+        IN,
+        OUT;
+    }
 }