DRILL-6321: Lateral Join and Unnest - rules, options, logical plan supports

Included changes:
* Add planner.enable_unnest_lateral option. Default value set to false.
* Enable FilterCorrectRule
* Add support to logical plan
* Fix rebase errors for DRILL-6321 commits

3 files changed, 38 insertions, 6 deletions

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillCorrelateRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillCorrelateRel.java
index e5eee4eef..7c49232ac 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillCorrelateRel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillCorrelateRel.java
@@ -24,9 +24,12 @@ import org.apache.calcite.rel.core.Correlate;
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.sql.SemiJoinType;
 import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.drill.common.logical.data.LateralJoin;
 import org.apache.drill.common.logical.data.LogicalOperator;
 import org.apache.drill.exec.planner.common.DrillCorrelateRelBase;
 
+import java.util.List;
+
 
 public class DrillCorrelateRel extends DrillCorrelateRelBase implements DrillRel {
 
@@ -45,7 +48,13 @@ public class DrillCorrelateRel extends DrillCorrelateRelBase implements DrillRel
 
   @Override
   public LogicalOperator implement(DrillImplementor implementor) {
-    //TODO: implementation for direct convert from RelNode to logical operator for explainPlan
-    return null;
+    final List<String> fields = getRowType().getFieldNames();
+    assert DrillJoinRel.isUnique(fields);
+    final int leftCount = left.getRowType().getFieldCount();
+
+    final LogicalOperator leftOp = DrillJoinRel.implementInput(implementor, 0, 0, left, this);
+    final LogicalOperator rightOp = DrillJoinRel.implementInput(implementor, 1, leftCount, right, this);
+
+    return new LateralJoin(leftOp, rightOp);
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillJoinRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillJoinRel.java
index b6b5c0310..b77fa6103 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillJoinRel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillJoinRel.java
@@ -103,9 +103,24 @@ public class DrillJoinRel extends DrillJoinRelBase implements DrillRel {
    * @return
    */
   private LogicalOperator implementInput(DrillImplementor implementor, int i, int offset, RelNode input) {
-    final LogicalOperator inputOp = implementor.visitChild(this, i, input);
+    return implementInput(implementor, i, offset, input, this);
+  }
+
+  /**
+   * Check to make sure that the fields of the inputs are the same as the output field names.
+   * If not, insert a project renaming them.
+   * @param implementor
+   * @param i
+   * @param offset
+   * @param input
+   * @param currentNode the node to be implemented
+   * @return
+   */
+  public static LogicalOperator implementInput(DrillImplementor implementor, int i, int offset,
+                                                RelNode input, DrillRel currentNode) {
+    final LogicalOperator inputOp = implementor.visitChild(currentNode, i, input);
     assert uniqueFieldNames(input.getRowType());
-    final List<String> fields = getRowType().getFieldNames();
+    final List<String> fields = currentNode.getRowType().getFieldNames();
     final List<String> inputFields = input.getRowType().getFieldNames();
     final List<String> outputFields = fields.subList(offset, offset + inputFields.size());
     if (!outputFields.equals(inputFields)) {
@@ -118,7 +133,8 @@ public class DrillJoinRel extends DrillJoinRelBase implements DrillRel {
     }
   }
 
-  private LogicalOperator rename(DrillImplementor implementor, LogicalOperator inputOp, List<String> inputFields, List<String> outputFields) {
+  private static LogicalOperator rename(DrillImplementor implementor, LogicalOperator inputOp,
+                                        List<String> inputFields, List<String> outputFields) {
     Project.Builder builder = Project.builder();
     builder.setInput(inputOp);
     for (Pair<String, String> pair : Pair.zip(inputFields, outputFields)) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillUnnestRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillUnnestRel.java
index ab827e577..0bf8c6844 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillUnnestRel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillUnnestRel.java
@@ -21,8 +21,11 @@ import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
 
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexFieldAccess;
 import org.apache.calcite.rex.RexNode;
+import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.logical.data.LogicalOperator;
+import org.apache.drill.common.logical.data.Unnest;
 import org.apache.drill.exec.planner.common.DrillUnnestRelBase;
 
 
@@ -37,7 +40,11 @@ public class DrillUnnestRel extends DrillUnnestRelBase implements DrillRel {
 
   @Override
   public LogicalOperator implement(DrillImplementor implementor) {
-    //TODO: implementation for direct convert from RelNode to logical operator for explainPlan
+    if(getRef() instanceof RexFieldAccess) {
+      final RexFieldAccess fldAccess = (RexFieldAccess)getRef();
+      return new Unnest(SchemaPath.getSimplePath(fldAccess.getField().getName()));
+    }
+
     return null;
   }