TAJO-1554 Implement hash-theta join

Author: navis

tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/Schema.java

@@ -313,7 +313,7 @@ public boolean containsAll(Collection<Column> columns) {
 
   public boolean containsAny(Collection<Column> columns) {
     for (Column column : columns) {
-      if (fields.contains(column)) {
+      if (contains(column)) {
         return true;
       }
     }

tajo-common/src/main/java/org/apache/tajo/storage/VTuple.java

@@ -28,7 +28,7 @@
 import java.net.InetAddress;
 import java.util.Arrays;
 
-public class VTuple implements Tuple, Cloneable {
+public class VTuple implements Tuple, Cloneable, Comparable<Tuple> {
 	@Expose public Datum [] values;
 	@Expose private long offset;
 
@@ -237,4 +237,21 @@ public static String toDisplayString(Datum [] values) {
     str.append(')');
     return str.toString();
   }
+
+  @Override
+  public int compareTo(Tuple other) {
+    for (int i = 0; i < size(); i++) {
+      if (isNull(i) && other.isNull(i)) {
+        continue;
+      }
+      if (isNull(i) || other.isNull(i)) {
+        return isNull(i) ? 1 : -1;
+      }
+      int compare = get(i).compareTo(other.get(i));
+      if (compare != 0) {
+        return compare;
+      }
+    }
+    return 0;
+  }
 }

tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java

@@ -281,9 +281,24 @@ public boolean checkIfInMemoryInnerJoinIsPossible(TaskAttemptContext context, Lo
     return inMemoryInnerJoinFlag;
   }
 
-  public PhysicalExec createJoinPlan(TaskAttemptContext context, JoinNode joinNode, PhysicalExec leftExec,
-                                     PhysicalExec rightExec) throws IOException {
+  public CommonJoinExec createJoinPlan(TaskAttemptContext context, JoinNode joinNode, PhysicalExec leftExec,
+                                       PhysicalExec rightExec) throws IOException {
+    CommonJoinExec joinExec;
+    try {
+      joinExec = createJoinExec(context, joinNode, leftExec, rightExec);
+    } catch (Exception e) {
+      LOG.warn("Failed to make join exec, failing back to CROSS join", e);
+      return createCrossJoinPlan(context, joinNode, leftExec, rightExec);
+    }
+    if (joinExec.isThetaJoin() && !joinExec.isHashJoin()) {
+      LOG.warn("Generic theta join is not supported, yet");
+      return createCrossJoinPlan(context, joinNode, leftExec, rightExec);
+    }
+    return joinExec;
+  }
 
+  private CommonJoinExec createJoinExec(TaskAttemptContext context, JoinNode joinNode, PhysicalExec leftExec,
+                                        PhysicalExec rightExec) throws IOException {
     switch (joinNode.getJoinType()) {
       case CROSS:
         return createCrossJoinPlan(context, joinNode, leftExec, rightExec);
@@ -317,7 +332,7 @@ public PhysicalExec createJoinPlan(TaskAttemptContext context, JoinNode joinNode
     }
   }
 
-  private PhysicalExec createCrossJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createCrossJoinPlan(TaskAttemptContext context, JoinNode plan,
                                            PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -343,7 +358,7 @@ private PhysicalExec createCrossJoinPlan(TaskAttemptContext context, JoinNode pl
     }
   }
 
-  private PhysicalExec createInnerJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createInnerJoinPlan(TaskAttemptContext context, JoinNode plan,
                                            PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -416,7 +431,7 @@ private PhysicalExec createInnerJoinPlan(TaskAttemptContext context, JoinNode pl
     return new PhysicalExec [] {smaller, larger};
   }
 
-  private PhysicalExec createBestInnerJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createBestInnerJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     boolean inMemoryHashJoin = false;
     if (checkIfInMemoryInnerJoinIsPossible(context, plan.getLeftChild(), true)
@@ -434,7 +449,7 @@ private PhysicalExec createBestInnerJoinPlan(TaskAttemptContext context, JoinNod
     }
   }
 
-  private MergeJoinExec createMergeInnerJoin(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createMergeInnerJoin(TaskAttemptContext context, JoinNode plan,
                                              PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     SortSpec[][] sortSpecs = PlannerUtil.getSortKeysFromJoinQual(
         plan.getJoinQual(), leftExec.getSchema(), rightExec.getSchema());
@@ -455,7 +470,7 @@ private MergeJoinExec createMergeInnerJoin(TaskAttemptContext context, JoinNode
     return new MergeJoinExec(context, plan, outerSort, innerSort, sortSpecs[0], sortSpecs[1]);
   }
 
-  private PhysicalExec createLeftOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createLeftOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -479,7 +494,7 @@ private PhysicalExec createLeftOuterJoinPlan(TaskAttemptContext context, JoinNod
     }
   }
 
-  private PhysicalExec createBestLeftOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createBestLeftOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                    PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     String [] rightLineage = PlannerUtil.getRelationLineage(plan.getRightChild());
     long rightTableVolume = estimateSizeRecursive(context, rightLineage);
@@ -505,7 +520,7 @@ private PhysicalExec createBestLeftOuterJoinPlan(TaskAttemptContext context, Joi
     }
   }
 
-  private PhysicalExec createBestRightJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createBestRightJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     //if the left operand is small enough => implement it as a left outer hash join with exchanged operators (note:
     // blocking, but merge join is blocking as well)
@@ -529,7 +544,7 @@ private PhysicalExec createBestRightJoinPlan(TaskAttemptContext context, JoinNod
     }
   }
 
-  private PhysicalExec createRightOuterMergeJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createRightOuterMergeJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                      PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     //the left operand is too large, so opt for merge join implementation
     LOG.info("Right Outer Join (" + plan.getPID() +") chooses [Merge Join].");
@@ -551,7 +566,7 @@ private PhysicalExec createRightOuterMergeJoinPlan(TaskAttemptContext context, J
     return new RightOuterMergeJoinExec(context, plan, outerSort2, innerSort2, sortSpecs2[0], sortSpecs2[1]);
   }
 
-  private PhysicalExec createRightOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createRightOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                 PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -573,7 +588,7 @@ private PhysicalExec createRightOuterJoinPlan(TaskAttemptContext context, JoinNo
     }
   }
 
-  private PhysicalExec createFullOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createFullOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -596,7 +611,7 @@ private PhysicalExec createFullOuterJoinPlan(TaskAttemptContext context, JoinNod
     }
   }
 
-  private HashFullOuterJoinExec createFullOuterHashJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createFullOuterHashJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                             PhysicalExec leftExec, PhysicalExec rightExec)
       throws IOException {
     String [] leftLineage = PlannerUtil.getRelationLineage(plan.getLeftChild());
@@ -642,7 +657,7 @@ private MergeFullOuterJoinExec createFullOuterMergeJoinPlan(TaskAttemptContext c
     return new MergeFullOuterJoinExec(context, plan, outerSort3, innerSort3, sortSpecs3[0], sortSpecs3[1]);
   }
 
-  private PhysicalExec createBestFullOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createBestFullOuterJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                    PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     String [] leftLineage = PlannerUtil.getRelationLineage(plan.getLeftChild());
     String [] rightLineage = PlannerUtil.getRelationLineage(plan.getRightChild());
@@ -659,7 +674,7 @@ private PhysicalExec createBestFullOuterJoinPlan(TaskAttemptContext context, Joi
   /**
    *  Left semi join means that the left side is the IN side table, and the right side is the FROM side table.
    */
-  private PhysicalExec createLeftSemiJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createLeftSemiJoinPlan(TaskAttemptContext context, JoinNode plan,
                                               PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -684,7 +699,7 @@ private PhysicalExec createLeftSemiJoinPlan(TaskAttemptContext context, JoinNode
   /**
    *  Left semi join means that the left side is the FROM side table, and the right side is the IN side table.
    */
-  private PhysicalExec createRightSemiJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createRightSemiJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -709,7 +724,7 @@ private PhysicalExec createRightSemiJoinPlan(TaskAttemptContext context, JoinNod
   /**
    *  Left semi join means that the left side is the FROM side table, and the right side is the IN side table.
    */
-  private PhysicalExec createLeftAntiJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createLeftAntiJoinPlan(TaskAttemptContext context, JoinNode plan,
                                               PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);
@@ -734,7 +749,7 @@ private PhysicalExec createLeftAntiJoinPlan(TaskAttemptContext context, JoinNode
   /**
    *  Left semi join means that the left side is the FROM side table, and the right side is the IN side table.
    */
-  private PhysicalExec createRightAntiJoinPlan(TaskAttemptContext context, JoinNode plan,
+  private CommonJoinExec createRightAntiJoinPlan(TaskAttemptContext context, JoinNode plan,
                                                PhysicalExec leftExec, PhysicalExec rightExec) throws IOException {
     Enforcer enforcer = context.getEnforcer();
     EnforceProperty property = getAlgorithmEnforceProperty(enforcer, plan);