DRILL-1957: Support nested loop join planning in order to enable NOT-IN, Inequality, Cartesian, uncorrelated EXISTS planning.

Add support for nested loop join planning where right input is scalar and is broadcast.

Add check for scalar subquery for NLJ. Add support for creating a Filter-NLJ plan.

Rebase on the branch with Jinfeng's Calcite rebasing work.

Conflicts:
	exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinUtils.java

Add unit tests for NLJoin.

Added test for inequality join.

Tests with BroadcastExchange, with HJ/MJ disabled.

Fix filter push down for NL joins by modifying row count computation for joins with always true conditions.  Rebase on master.  Refactor unit tests.

Improved checking of preconditions for NL join.

Handle the case where scalar aggregate is a child of Filter.

DRILL-1957: Support nested loop join planning in order to enable NOT-IN, Inequality, EXISTS planning.

Better checks for cartesian and inequality joins. Rebase on latest master.

Refactor costing for logical join.  Add tests.  Enable more TPC-H tests.

Remove the check for cartesian join from DrillJoinRel constructor.

Clear left and right keys before calling splitJoinCondition.

Address review comments: Remove redundant call to getJoinCategory.  Added comment in DrillRuleSet.

1 files changed, 50 insertions, 0 deletions

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinUtils.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinUtils.java
index 45b1093bc..5ed5d2709 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinUtils.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/JoinUtils.java
@@ -24,6 +24,9 @@ import org.apache.calcite.rex.RexNode;
 import org.apache.drill.common.logical.data.JoinCondition;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.volcano.RelSubset;
+import org.apache.drill.exec.planner.logical.DrillAggregateRel;
+import org.apache.drill.exec.planner.logical.DrillFilterRel;
 
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.expression.ErrorCollector;
@@ -38,6 +41,7 @@ import org.apache.drill.exec.resolver.TypeCastRules;
 
 import java.util.LinkedList;
 import java.util.List;
+import com.google.common.collect.Lists;
 
 public class JoinUtils {
   public static enum JoinComparator {
@@ -46,6 +50,12 @@ public class JoinUtils {
     IS_NOT_DISTINCT_FROM // 'IS NOT DISTINCT FROM' comparator
   }
 
+  public static enum JoinCategory {
+    EQUALITY,  // equality join
+    INEQUALITY,  // inequality join: <>, <, >
+    CARTESIAN   // no join condition
+  }
+
   // Check the comparator for the join condition. Note that a similar check is also
   // done in JoinPrel; however we have to repeat it here because a physical plan
   // may be submitted directly to Drill.
@@ -194,4 +204,44 @@ public class JoinUtils {
       }
     }
   }
+
+  public static boolean isScalarSubquery(RelNode childrel) {
+    DrillAggregateRel agg = null;
+    RelNode currentrel = childrel;
+    while (agg == null && currentrel != null) {
+      if (currentrel instanceof DrillAggregateRel) {
+        agg = (DrillAggregateRel)currentrel;
+      } else if (currentrel instanceof DrillFilterRel) {
+        currentrel = currentrel.getInput(0);
+      } else if (currentrel instanceof RelSubset) {
+        currentrel = ((RelSubset)currentrel).getBest() ;
+      } else {
+        break;
+      }
+    }
+
+    if (agg != null) {
+      if (agg.getGroupSet().isEmpty()) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  public static JoinCategory getJoinCategory(RelNode left, RelNode right, RexNode condition,
+      List<Integer> leftKeys, List<Integer> rightKeys) {
+    if (condition.isAlwaysTrue()) {
+      return JoinCategory.CARTESIAN;
+    }
+    leftKeys.clear();
+    rightKeys.clear();
+    RexNode remaining = RelOptUtil.splitJoinCondition(left, right, condition, leftKeys, rightKeys);
+
+    if (!remaining.isAlwaysTrue() || (leftKeys.size() == 0 || rightKeys.size() == 0) ) {
+      // for practical purposes these cases could be treated as inequality
+      return JoinCategory.INEQUALITY;
+    }
+    return JoinCategory.EQUALITY;
+  }
+
 }