DRILL-6792: Find the right probe side fragment wrapper & fix DrillBuf reference count bugs & tune the execution flow & support left deep tree

closes #1504

3 files changed, 23 insertions, 58 deletions

diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPD.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPD.java
index 5eae12e6e..a5fc5ba49 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPD.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPD.java
@@ -40,9 +40,9 @@ public class TestHashJoinJPPD extends PhysicalOpUnitTestBase {
   public void testBroadcastHashJoin1Cond() {
     List<BloomFilterDef> bloomFilterDefs = new ArrayList<>();
     int numBytes = BloomFilter.optimalNumOfBytes(2600, 0.01);
-    BloomFilterDef bloomFilterDef = new BloomFilterDef(numBytes, true, "lft");
+    BloomFilterDef bloomFilterDef = new BloomFilterDef(numBytes, true, "lft", "rgt");
     bloomFilterDefs.add(bloomFilterDef);
-    RuntimeFilterDef runtimeFilterDef = new RuntimeFilterDef(true, false, bloomFilterDefs, false );
+    RuntimeFilterDef runtimeFilterDef = new RuntimeFilterDef(true, false, bloomFilterDefs, false, -1);
     HashJoinPOP joinConf = new HashJoinPOP(null, null,
       Lists.newArrayList(joinCond("lft", "EQUALS", "rgt")), JoinRelType.INNER, runtimeFilterDef);
     operatorFixture.getOptionManager().setLocalOption("exec.hashjoin.num_partitions", 4);
@@ -71,11 +71,11 @@ public class TestHashJoinJPPD extends PhysicalOpUnitTestBase {
   public void testBroadcastHashJoin2Cond() {
     List<BloomFilterDef> bloomFilterDefs = new ArrayList<>();
     int numBytes = BloomFilter.optimalNumOfBytes(2600, 0.01);
-    BloomFilterDef bloomFilterDef = new BloomFilterDef(numBytes, true, "lft");
-    BloomFilterDef bloomFilterDef1 = new BloomFilterDef(numBytes, true, "a");
+    BloomFilterDef bloomFilterDef = new BloomFilterDef(numBytes, true, "lft", "rgt");
+    BloomFilterDef bloomFilterDef1 = new BloomFilterDef(numBytes, true, "a", "b");
     bloomFilterDefs.add(bloomFilterDef);
     bloomFilterDefs.add(bloomFilterDef1);
-    RuntimeFilterDef runtimeFilterDef = new RuntimeFilterDef(true, false, bloomFilterDefs, false );
+    RuntimeFilterDef runtimeFilterDef = new RuntimeFilterDef(true, false, bloomFilterDefs, false, -1);
     HashJoinPOP joinConf = new HashJoinPOP(null, null,
       Lists.newArrayList(joinCond("lft", "EQUALS", "rgt"), joinCond("a", "EQUALS", "b")), JoinRelType.INNER, runtimeFilterDef);
     operatorFixture.getOptionManager().setLocalOption("exec.hashjoin.num_partitions", 4);
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPDPlan.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPDPlan.java
index 2370ffa06..ac174d1ce 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPDPlan.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/join/TestHashJoinJPPDPlan.java
@@ -34,38 +34,6 @@ public class TestHashJoinJPPDPlan extends JoinTestBase {
   }
 
   @Test
-  public void testInnnerHashJoin() throws Exception {
-    String sql = "SELECT nations.N_NAME, count(*)"
-      + "FROM\n"
-      + " dfs.`sample-data/nation.parquet` nations\n"
-      + "JOIN\n"
-      + "  dfs.`sample-data/region.parquet` regions\n"
-      + "  on nations.N_REGIONKEY = regions.R_REGIONKEY "
-      + "WHERE nations.N_NAME = 'A' "
-      + "group by nations.N_NAME";
-    String expectedColNames1 =  "\"runtimeFilterDef\"";
-    String expectedColNames2 =  "\"bloomFilterDefs\"";
-    String expectedColNames3 =  "\"runtime-filter\"";
-    testPhysicalPlan(sql, expectedColNames1, expectedColNames2, expectedColNames3);
-  }
-
-  @Test
-  public void testRightHashJoin() throws Exception {
-    String sql = "SELECT nations.N_NAME, count(*)"
-      + "FROM\n"
-      + " dfs.`sample-data/nation.parquet` nations\n"
-      + "RIGHT JOIN\n"
-      + "  dfs.`sample-data/region.parquet` regions\n"
-      + "  on nations.N_REGIONKEY = regions.R_REGIONKEY "
-      + "WHERE nations.N_NAME = 'A' "
-      + "group by nations.N_NAME";
-    String expectedColNames1 =  "\"runtimeFilterDef\"";
-    String expectedColNames2 =  "\"bloomFilterDefs\"";
-    String expectedColNames3 =  "\"runtime-filter\"";
-    testPhysicalPlan(sql, expectedColNames1, expectedColNames2, expectedColNames3);
-  }
-
-  @Test
   public void testLeftHashJoin() throws Exception {
     String sql = "SELECT nations.N_NAME, count(*)"
       + "FROM\n"
@@ -95,24 +63,4 @@ public class TestHashJoinJPPDPlan extends JoinTestBase {
     String excludedColNames3 =  "\"runtime-filter\"";
     testPlanWithAttributesMatchingPatterns(sql, null, new String[]{excludedColNames1, excludedColNames2, excludedColNames3});
   }
-
-  @Test
-  public void testInnnerHashJoinWithRightDeepTree() throws Exception {
-    String sql = "SELECT nations.N_NAME, count(*)"
-      + "FROM\n"
-      + " cp.`tpch/nation.parquet` nations\n"
-      + "JOIN\n"
-      + "  cp.`tpch/region.parquet` regions\n"
-      + "  on nations.N_REGIONKEY = regions.R_REGIONKEY "
-      + "JOIN cp.`tpch/customer.parquet` customers\n"
-      + " on nations.N_NATIONKEY = customers.C_NATIONKEY "
-      + "WHERE nations.N_NAME = 'A' "
-      + "group by nations.N_NAME";
-    String expectedColNames1 =  "\"runtimeFilterDef\"";
-    String expectedColNames2 =  "\"bloomFilterDefs\"";
-    String expectedColNames3 =  "\"runtime-filter\"";
-    testPhysicalPlan(sql, expectedColNames1, expectedColNames2, expectedColNames3);
-  }
-
-
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/work/filter/BloomFilterTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/work/filter/BloomFilterTest.java
index c05cdfdf8..c1d157600 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/work/filter/BloomFilterTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/work/filter/BloomFilterTest.java
@@ -135,7 +135,6 @@ public class BloomFilterTest {
 
   @Test
   public void testNotExist() throws Exception {
-
     Drillbit bit = new Drillbit(c, RemoteServiceSet.getLocalServiceSet(), ClassPathScanner.fromPrescan(c));
     bit.run();
     DrillbitContext bitContext = bit.getContext();
@@ -192,6 +191,12 @@ public class BloomFilterTest {
     long hashCode = probeHash64.hash64Code(0, 0, 0);
     boolean contain = bloomFilter.find(hashCode);
     Assert.assertFalse(contain);
+    bloomFilter.getContent().close();
+    vectorContainer.clear();
+    probeVectorContainer.clear();
+    context.close();
+    bitContext.close();
+    bit.close();
   }
 
 
@@ -254,6 +259,12 @@ public class BloomFilterTest {
     long hashCode = probeHash64.hash64Code(0, 0, 0);
     boolean contain = bloomFilter.find(hashCode);
     Assert.assertTrue(contain);
+    bloomFilter.getContent().close();
+    vectorContainer.clear();
+    probeVectorContainer.clear();
+    context.close();
+    bitContext.close();
+    bit.close();
   }
 
 
@@ -324,5 +335,11 @@ public class BloomFilterTest {
     long hashCode = probeHash64.hash64Code(0, 0, 0);
     boolean contain = bloomFilter.find(hashCode);
     Assert.assertTrue(contain);
+    bloomFilter.getContent().close();
+    vectorContainer.clear();
+    probeVectorContainer.clear();
+    context.close();
+    bitContext.close();
+    bit.close();
   }
 }