DRILL-1335: Fix merge join operator when compare null against null value.

2 files changed, 218 insertions, 1 deletions

diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoinMulCondition.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoinMulCondition.java
index d9cfa5c99..bf402d127 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoinMulCondition.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestMergeJoinMulCondition.java
@@ -37,7 +37,6 @@ import org.junit.rules.TestRule;
 import com.google.common.base.Charsets;
 import com.google.common.io.Files;
 
-//@Ignore("Currently returns wrong result.  Stopped working when incoming became more than one result set.")
 public class TestMergeJoinMulCondition extends PopUnitTestBase {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestMergeJoinMulCondition.class);
 
@@ -72,4 +71,51 @@ public class TestMergeJoinMulCondition extends PopUnitTestBase {
     }
   }
 
+  @Test
+  // The physical plan is obtained through sql:
+  // alter session set `planner.enable_hashjoin`=false;
+  // select * from cp.`region.json` t1, cp.`region.json` t2 where t1.non_exist = t2.non_exist2 ;
+  public void testMergeJoinInnerNullKey() throws Exception {
+    RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+
+    try(Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+        DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+
+      bit1.run();
+      client.connect();
+      List<QueryResultBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+          Files.toString(FileUtils.getResourceAsFile("/join/merge_join_nullkey.json"), Charsets.UTF_8).replace("${JOIN_TYPE}", "INNER"));
+      int count = 0;
+      for(QueryResultBatch b : results) {
+        if (b.getHeader().getRowCount() != 0)
+          count += b.getHeader().getRowCount();
+        b.release();
+      }
+      assertEquals(0, count);
+    }
+  }
+
+  @Test
+  // The physical plan is obtained through sql:
+  // alter session set `planner.enable_hashjoin`=false;
+  // select * from cp.`region.json` t1 left outer join cp.`region.json` t2 on  t1.non_exist = t2.non_exist2 ;
+  public void testMergeJoinLeftOuterNullKey() throws Exception {
+    RemoteServiceSet serviceSet = RemoteServiceSet.getLocalServiceSet();
+
+    try(Drillbit bit1 = new Drillbit(CONFIG, serviceSet);
+        DrillClient client = new DrillClient(CONFIG, serviceSet.getCoordinator());) {
+
+      bit1.run();
+      client.connect();
+      List<QueryResultBatch> results = client.runQuery(org.apache.drill.exec.proto.UserBitShared.QueryType.PHYSICAL,
+          Files.toString(FileUtils.getResourceAsFile("/join/merge_join_nullkey.json"), Charsets.UTF_8).replace("${JOIN_TYPE}", "LEFT"));
+      int count = 0;
+      for(QueryResultBatch b : results) {
+        if (b.getHeader().getRowCount() != 0)
+          count += b.getHeader().getRowCount();
+        b.release();
+      }
+      assertEquals(110, count);
+    }
+  }
 }
diff --git a/exec/java-exec/src/test/resources/join/merge_join_nullkey.json b/exec/java-exec/src/test/resources/join/merge_join_nullkey.json
new file mode 100644
index 000000000..b283dda2e
--- /dev/null
+++ b/exec/java-exec/src/test/resources/join/merge_join_nullkey.json
@@ -0,0 +1,171 @@
+{
+  "head" : {
+    "version" : 1,
+    "generator" : {
+      "type" : "DefaultSqlHandler",
+      "info" : ""
+    },
+    "type" : "APACHE_DRILL_PHYSICAL",
+    "options" : [ ],
+    "queue" : 0,
+    "resultMode" : "EXEC"
+  },
+  "graph" : [ {
+    "pop" : "fs-scan",
+    "@id" : 11,
+    "files" : [ "/region.json" ],
+    "storage" : {
+      "type" : "file",
+      "enabled" : true,
+      "connection" : "classpath:///",
+      "workspaces" : null,
+      "formats" : {
+        "json" : {
+          "type" : "json"
+        },
+        "parquet" : {
+          "type" : "parquet"
+        }
+      }
+    },
+    "format" : {
+      "type" : "json"
+    },
+    "selectionRoot" : "/region.json",
+    "cost" : 18.0
+  }, {
+    "pop" : "project",
+    "@id" : 9,
+    "exprs" : [ {
+      "ref" : "`T2¦¦*`",
+      "expr" : "`*`"
+    } ],
+    "child" : 11,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "external-sort",
+    "@id" : 7,
+    "child" : 9,
+    "orderings" : [ {
+      "order" : "ASC",
+      "expr" : "`T2¦¦non_exist`",
+      "nullDirection" : "UNSPECIFIED"
+    } ],
+    "reverse" : false,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "selection-vector-remover",
+    "@id" : 5,
+    "child" : 7,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "fs-scan",
+    "@id" : 10,
+    "files" : [ "/region.json" ],
+    "storage" : {
+      "type" : "file",
+      "enabled" : true,
+      "connection" : "classpath:///",
+      "workspaces" : null,
+      "formats" : {
+        "json" : {
+          "type" : "json"
+        },
+        "parquet" : {
+          "type" : "parquet"
+        }
+      }
+    },
+    "format" : {
+      "type" : "json"
+    },
+    "selectionRoot" : "/region.json",
+    "cost" : 18.0
+  }, {
+    "pop" : "project",
+    "@id" : 8,
+    "exprs" : [ {
+      "ref" : "`T3¦¦*`",
+      "expr" : "`*`"
+    } ],
+    "child" : 10,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "external-sort",
+    "@id" : 6,
+    "child" : 8,
+    "orderings" : [ {
+      "order" : "ASC",
+      "expr" : "`T3¦¦non_exist2`",
+      "nullDirection" : "UNSPECIFIED"
+    } ],
+    "reverse" : false,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "selection-vector-remover",
+    "@id" : 4,
+    "child" : 6,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "merge-join",
+    "@id" : 3,
+    "left" : 5,
+    "right" : 4,
+    "conditions" : [ {
+      "relationship" : "==",
+      "left" : "`T2¦¦non_exist`",
+      "right" : "`T3¦¦non_exist2`"
+    } ],
+    "joinType" : "${JOIN_TYPE}",
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "project",
+    "@id" : 2,
+    "exprs" : [ {
+      "ref" : "`T2¦¦*`",
+      "expr" : "`T2¦¦*`"
+    }, {
+      "ref" : "`T3¦¦*`",
+      "expr" : "`T3¦¦*`"
+    } ],
+    "child" : 3,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "project",
+    "@id" : 1,
+    "exprs" : [ {
+      "ref" : "`*`",
+      "expr" : "`T2¦¦*`"
+    }, {
+      "ref" : "`*0`",
+      "expr" : "`T3¦¦*`"
+    } ],
+    "child" : 2,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  }, {
+    "pop" : "screen",
+    "@id" : 0,
+    "child" : 1,
+    "initialAllocation" : 1000000,
+    "maxAllocation" : 10000000000,
+    "cost" : 18.0
+  } ]
+}