Introduce 'preferred input' for joins above transitive joins #335

compiler/src/main/scala/ingraph/compiler/qplan2jplan/JPlanToFPlan.scala

@@ -75,8 +75,16 @@ object JPlanToFPlan {
 
         val rpTotal = req ++ reqOp
 
-        val rpLeft = propagate(rpTotal, j.left.output)
-        val rpRight = propagate(rpTotal, j.right.output).filter(x => !rpLeft.map(_.resolvedName).contains(x.resolvedName))
+        val (rpLeft, rpRight) = j.inputPreference match {
+          case Left() =>
+            val rpLeft  = propagate(rpTotal, j.left.output)
+            val rpRight = propagate(rpTotal, j.right.output).filter(x => !rpLeft.map(_.resolvedName).contains(x.resolvedName))
+            (rpLeft, rpRight)
+          case Right() =>
+            val rpRight = propagate(rpTotal, j.right.output)
+            val rpLeft  = propagate(rpTotal, j.left.output).filter(x => !rpRight.map(_.resolvedName).contains(x.resolvedName))
+            (rpLeft, rpRight)
+        }
 
         val left = transform(j.left, rpLeft)
         val right = transform(j.right, rpRight)

compiler/src/main/scala/ingraph/compiler/qplan2jplan/QPlanToJPlan.scala

@@ -23,9 +23,15 @@ object QPlanToJPlan {
           case _                    => transform(child)
         }
         val transitiveRight = expandToTransitiveEdges(src, trg, el, dir)
-        val labelJoin = jplan.GetVertices(trg)
-        // perform an additional join to ensure that the type of the target is correct
-        jplan.Join(jplan.TransitiveJoin(transitiveLeft, transitiveRight, el), labelJoin)
+        val sourceJoin = jplan.GetVertices(src)
+        val targetJoin = jplan.GetVertices(trg)
+        // perform an additional join to ensure that the type of the target nodes is correct
+        // note that this join is right-preferring, i.e. its 'requiredProperties' will be propagated to its right input
+        jplan.Join(
+          jplan.TransitiveJoin(transitiveLeft, transitiveRight, el),
+          targetJoin,
+          Right()
+        )
       case qplan.Top(skipExpr, limitExpr, qplan.Sort(order, child)) => jplan.SortAndTop(skipExpr, limitExpr, order, transform(child))
       // if Sort operator found w/o Top, then skip and limit defaults to None
       case qplan.Top(_, _, _) => throw new UnsupportedOperationException("Vanilla 'SKIP'/'LIMIT' is not supported, add an 'ORDER BY' clause. Please.")

compiler/src/main/scala/ingraph/model/jplan.scala

@@ -19,6 +19,12 @@ abstract class BinaryJNode extends GenericBinaryNode[JNode] with JNode {}
 
 trait JoinLike extends BinaryJNode {
   def common: Seq[ResolvableName] = left.output.filter(x => right.output.map(_.resolvedName).contains(x.resolvedName))
+
+  /**
+    * @return input side preference, used to determine which input node (left/right) should be preferred when
+    *         propagating the "required properties" by the schema inferencer of {@see JPlanToFPlan}
+    */
+  def inputPreference: Side = Left()
 }
 trait EquiJoinLike extends JoinLike {
   override def output: Seq[ResolvableName] = left.output ++ right.output.filter(x => !left.output.map(_.resolvedName).contains(x.resolvedName))
@@ -83,7 +89,7 @@ case class AntiJoin(left: JNode, right: JNode) extends BinaryJNode with JoinLike
   override def output: Seq[ResolvableName] = left.output
 }
 
-case class Join(left: JNode, right: JNode) extends BinaryJNode with EquiJoinLike {}
+case class Join(left: JNode, right: JNode, override val inputPreference: Side = Left()) extends BinaryJNode with EquiJoinLike {}
 
 case class TransitiveJoin(left: JNode, right: JNode, edgeList: EdgeListAttribute) extends BinaryJNode with EquiJoinLike {}
 

ire/src/main/scala/hu/bme/mit/ire/nodes/binary/TransitiveJoinNode.scala

@@ -167,8 +167,9 @@ class TransitiveJoinNode(override val next: (ReteMessage) => Unit,
     val sourceId = inputTuple(sourceVertexIndex).asInstanceOf[Number].longValue
     var pathsFromSourceVertexBuilder = new VectorBuilder[Tuple]
 
-    if (minHops == 0)
-      pathsFromSourceVertexBuilder += buildResultTuple(inputTuple, Path(), inputTuple)
+    if (minHops == 0) {
+      pathsFromSourceVertexBuilder += buildResultTuple(inputTuple, Path(), Vector(sourceId))
+    }
 
     val reachableFromSource = reachableVertices.getOrElse(sourceId, mutable.HashMap.empty)
     pathsFromSourceVertexBuilder ++= reachableFromSource.keysIterator.flatMap(

queries/ldbc-snb-bi/bi-14.cypher

@@ -5,7 +5,7 @@
     endDate: 20120630220000000
   }
 */
-MATCH (person:Person)<-[:HAS_CREATOR]-(post:Post)<-[:REPLY_OF*1..]-(reply:Message)
+MATCH (person:Person)<-[:HAS_CREATOR]-(post:Post)<-[:REPLY_OF*0..]-(reply:Message)
 WHERE  post.creationDate >= $startDate
   AND  post.creationDate <= $endDate
   AND reply.creationDate >= $startDate

tests/src/test/scala/tests/BiValidationTest.scala

@@ -26,17 +26,18 @@ class BiValidationTest extends FunSuite {
 //       |DETACH DELETE f
 //    """.stripMargin
 
-  val meta = Seq(1) //1
+  val meta = Seq(1)
   val listComprehensions = Seq(11, 13) //2
   val madness = Seq(25)
   val orderby = Seq(10)
   val hangs = Seq(19)
+  val startWithWith = Seq(20)
 
-  val transitives = Seq(14, 16, 18, 20)
+  val transitives = Seq(14, 16, 18) //3
   val buggy =   Seq(3, 5, 6, 8, 15, 17, 21) //7
   val working = Seq(2, 4, 7, 9, 12, 22, 23, 24) //8
 
-  val testCases: Seq[LdbcSnbTestCase] = (working++buggy++transitives).sorted map (i => new LdbcSnbTestCase("bi", i, f"${csvDir}/$i%02d/", csvPostfix, Seq()))
+  val testCases: Seq[LdbcSnbTestCase] = (transitives++buggy++working).sorted map (i => new LdbcSnbTestCase("bi", i, f"${csvDir}/$i%02d/", csvPostfix, Seq()))
 
 //  val ntr = new Neo4jTestRunner
 //  ntr.load(graphMLPath)