Merge branch 'bnlj-only' into left-outer

Author: octaviansima

src/enclave/Enclave/BroadcastNestedLoopJoin.cpp

@@ -6,10 +6,9 @@
 #include "common.h"
 
 /** C++ implementation of a broadcast nested loop join.
- * Which row is broadcast and which row is not is handled in Scala.
+ * Assumes outer_rows is streamed and inner_rows is broadcast.
  * DOES NOT rely on rows to be tagged primary or secondary, and that
- * assumption will break the implementation. In other words,
- * outer is always left and inner is always right.
+ * assumption will break the implementation.
  */
 void broadcast_nested_loop_join(
   uint8_t *join_expr, size_t join_expr_length,

src/enclave/Enclave/ExpressionEvaluation.h

@@ -1756,10 +1756,9 @@ class FlatbuffersJoinExprEvaluator {
     }
   }
 
-  /**
-   * Return true if the given row is from the primary table, indicated by its first field, which
-   * must be an IntegerField.
-   * Rows MUST have been tagged in Scala.
+   /** Return true if the given row is from the primary table, indicated by its first field, which
+    * must be an IntegerField.
+    * Rows MUST have been tagged in Scala.
    */
   bool is_primary(const tuix::Row *row) {
     return static_cast<const tuix::IntegerField *>(
@@ -1774,12 +1773,14 @@ class FlatbuffersJoinExprEvaluator {
     return is_primary(row1) ? row1 : row2;
   }
 
-  /** Return true if the two rows are from the same join group. */
+  /** Return true if the two rows satisfy the join condition. */
   bool eval_condition(const tuix::Row *row1, const tuix::Row *row2) {
     builder.Clear();
     bool row1_equals_row2;
 
-    /* Check equality for equi joins */
+    /** Check equality for equi joins. If it is a non-equi join, 
+     * the key evaluators will be empty, so the code never enters the for loop.
+    */
     auto &row1_evaluators = is_primary(row1) ? left_key_evaluators : right_key_evaluators;
     auto &row2_evaluators = is_primary(row2) ? left_key_evaluators : right_key_evaluators;
     for (uint32_t i = 0; i < row1_evaluators.size(); i++) {

src/flatbuffers/operators.fbs

@@ -58,8 +58,7 @@ table JoinExpr {
     // pairs of rows  where each expression from the left is equal to the matching expression from the right.
     left_keys:[Expr];
     right_keys:[Expr];
-    // In the case of non-equi joins, we pass in a condition
-    // as an expression and evaluate that on each pair of rows.
+    // In the case of non-equi joins, we pass in a condition as an expression and evaluate that on each pair of rows.
     // TODO: have equi joins use this condition rather than an additional filter operation.
     condition:Expr;
 }

src/main/scala/edu/berkeley/cs/rise/opaque/execution/SGXEnclave.scala

@@ -43,7 +43,7 @@ class SGXEnclave extends java.io.Serializable {
     eid: Long, joinExpr: Array[Byte], input: Array[Byte]): Array[Byte]
 
   @native def BroadcastNestedLoopJoin(
-    eid: Long, joinExpr: Array[Byte], innerBlock: Array[Byte], outerBlock: Array[Byte]): Array[Byte]
+    eid: Long, joinExpr: Array[Byte], outerBlock: Array[Byte], innerBlock: Array[Byte]): Array[Byte]
 
   @native def NonObliviousAggregate(
     eid: Long, aggOp: Array[Byte], inputRows: Array[Byte], isPartial: Boolean): (Array[Byte])

src/main/scala/edu/berkeley/cs/rise/opaque/execution/operators.scala

@@ -350,7 +350,7 @@ case class EncryptedBroadcastNestedLoopJoinExec(
     val rightRDD = right.asInstanceOf[OpaqueOperatorExec].executeBlocked()
 
     joinType match {
-      case LeftAnti | LeftOuter => {
+      case LeftExistence(_) | LeftOuter => {
         join(leftRDD, rightRDD, joinExprSer)
       }
       case _ =>
@@ -361,7 +361,8 @@ case class EncryptedBroadcastNestedLoopJoinExec(
   def join(leftRDD: RDD[Block], rightRDD: RDD[Block],
       joinExprSer: Array[Byte]): RDD[Block] = {
     // We pick which side to broadcast/stream according to buildSide.
-    // NOTE: outer_rows and inner_rows in C++ ALWAYS correspond to left and right side.
+    // BuildRight means the right relation <=> the broadcast relation.
+    // NOTE: outer_rows and inner_rows in C++ correspond to stream and broadcast side respectively.
     var (streamRDD, broadcastRDD) = buildSide match {
       case BuildRight =>
         (leftRDD, rightRDD)

src/main/scala/edu/berkeley/cs/rise/opaque/strategies.scala

@@ -114,8 +114,11 @@ object OpaqueOperators extends Strategy {
 
     // Used to match non-equi joins
     case Join(left, right, joinType, condition, hint) if isEncrypted(left) && isEncrypted(right) =>
-      // Pick which side to broadcast: if left join, broadcast right. If right join, broadcast left.
-      // Otherwise, pick the smallest side to broadcast.
+      // How to pick broadcast side: if left join, broadcast right. If right join, broadcast left.
+      // This is the simplest and most performant method, but may be worth revisting if one side is
+      // significantly smaller than the other. Otherwise, pick the smallest side to broadcast.
+      // NOTE: the current implementation of BNLJ only works under the assumption that
+      // left join <==> broadcast right AND right join <==> broadcast left.
       val desiredBuildSide = if (joinType.isInstanceOf[InnerLike] || joinType == FullOuter)
           getSmallerSide(left, right) else
           getBroadcastSideBNLJ(joinType)