TAJO-1107: Broadcast join on non-leaf node scans only first data file.

tajo-core/src/main/java/org/apache/tajo/master/querymaster/Repartitioner.java

@@ -274,18 +274,21 @@ public static void scheduleFragmentsForJoinQuery(TaskSchedulerContext schedulerC
           intermediateFragments[index++] = fragments[eachIdx];
         }
         FileFragment[] broadcastFragments = new FileFragment[broadcastIndexList.size()];
+        ScanNode[] broadcastScans = new ScanNode[broadcastIndexList.size()];
         index = 0;
         for (Integer eachIdx : broadcastIndexList) {
           scans[eachIdx].setBroadcastTable(true);
-          broadcastFragments[index++] = fragments[eachIdx];
+          broadcastScans[index] = scans[eachIdx];
+          broadcastFragments[index] = fragments[eachIdx];
+          index++;
         }
         LOG.info(String.format("[Distributed Join Strategy] : Broadcast Join, join_node=%s", nonLeafScanNames));
         scheduleSymmetricRepartitionJoin(masterContext, schedulerContext, subQuery,
-            intermediateScans, intermediateScanStats, intermediateFragments, broadcastFragments);
+            intermediateScans, intermediateScanStats, intermediateFragments, broadcastScans, broadcastFragments);
       }
     } else {
       LOG.info("[Distributed Join Strategy] : Symmetric Repartition Join");
-      scheduleSymmetricRepartitionJoin(masterContext, schedulerContext, subQuery, scans, stats, fragments, null);
+      scheduleSymmetricRepartitionJoin(masterContext, schedulerContext, subQuery, scans, stats, fragments, null, null);
     }
   }
 
@@ -305,6 +308,7 @@ private static void scheduleSymmetricRepartitionJoin(QueryMasterTask.QueryMaster
                                                        ScanNode[] scans,
                                                        long[] stats,
                                                        FileFragment[] fragments,
+                                                       ScanNode[] broadcastScans,
                                                        FileFragment[] broadcastFragments) throws IOException {
     MasterPlan masterPlan = subQuery.getMasterPlan();
     ExecutionBlock execBlock = subQuery.getBlock();
@@ -388,12 +392,35 @@ private static void scheduleSymmetricRepartitionJoin(QueryMasterTask.QueryMaster
     int joinTaskNum = Math.min(maxTaskNum, hashEntries.size());
     LOG.info("The determined number of join tasks is " + joinTaskNum);
 
-    FileFragment[] rightFragments = new FileFragment[1 + (broadcastFragments == null ? 0 : broadcastFragments.length)];
-    rightFragments[0] = fragments[1];
+    List<FileFragment> rightFragments = new ArrayList<FileFragment>();
+    rightFragments.add(fragments[1]);
+
     if (broadcastFragments != null) {
-      System.arraycopy(broadcastFragments, 0, rightFragments, 1, broadcastFragments.length);
+      //In this phase a ScanNode has a single fragment.
+      //If there are more than one data files, that files should be added to fragments or partition path
+      AbstractStorageManager storageManager = subQuery.getStorageManager();
+      int index = 0;
+      for (FileFragment eachFragment: broadcastFragments) {
+        Path[] partitionScanPaths = null;
+        ScanNode scan = broadcastScans[index];
+        TableDesc tableDesc = masterContext.getTableDescMap().get(scan.getCanonicalName());
+        if (scan.getType() == NodeType.PARTITIONS_SCAN) {
+          PartitionedTableScanNode partitionScan = (PartitionedTableScanNode)scan;
+          partitionScanPaths = partitionScan.getInputPaths();
+          // set null to inputPaths in getFragmentsFromPartitionedTable()
+          getFragmentsFromPartitionedTable(subQuery.getStorageManager(), scan, tableDesc);
+          partitionScan.setInputPaths(partitionScanPaths);
+        } else {
+          Collection<FileFragment> scanFragments = subQuery.getStorageManager().getSplits(scan.getCanonicalName(),
+              tableDesc.getMeta(), tableDesc.getSchema(), tableDesc.getPath());
+          if (scanFragments != null) {
+            rightFragments.addAll(scanFragments);
+          }
+        }
+        index++;
+      }
     }
-    SubQuery.scheduleFragment(subQuery, fragments[0], Arrays.asList(rightFragments));
+    SubQuery.scheduleFragment(subQuery, fragments[0], rightFragments);
 
     // Assign partitions to tasks in a round robin manner.
     for (Entry<Integer, Map<ExecutionBlockId, List<IntermediateEntry>>> entry

tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java

@@ -18,6 +18,7 @@
 
 package org.apache.tajo.engine.query;
 
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.*;
@@ -42,6 +43,8 @@
 import java.io.File;
 import java.io.OutputStream;
 import java.sql.ResultSet;
+import java.util.ArrayList;
+import java.util.List;
 
 import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
 import static org.junit.Assert.*;
@@ -711,6 +714,8 @@ and a.o_orderstatus in ('F')
   }
   @Test
   public void testMultipleBroadcastDataFileWithZeroLength() throws Exception {
+    // According to node type(leaf or non-leaf) Broadcast join is determined differently by Repartitioner.
+    // testMultipleBroadcastDataFileWithZeroLength testcase is for the leaf node
     createMultiFile("nation", 2, new TupleCreator() {
       public Tuple createTuple(String[] columnDatas) {
         return new VTuple(new Datum[]{
@@ -721,7 +726,7 @@ public Tuple createTuple(String[] columnDatas) {
         });
       }
     });
-    addEmptyDataFile("nation");
+    addEmptyDataFile("nation_multifile", false);
 
     ResultSet res = executeQuery();
 
@@ -731,13 +736,110 @@ public Tuple createTuple(String[] columnDatas) {
     executeString("DROP TABLE nation_multifile PURGE");
   }
 
-  private void addEmptyDataFile(String tableName) throws Exception {
-    String multiTableName = tableName + "_multifile";
-    TableDesc table = client.getTableDesc(multiTableName);
+  @Test
+  public void testMultipleBroadcastDataFileWithZeroLength2() throws Exception {
+    // According to node type(leaf or non-leaf) Broadcast join is determined differently by Repartitioner.
+    // testMultipleBroadcastDataFileWithZeroLength2 testcase is for the non-leaf node
+    createMultiFile("nation", 2, new TupleCreator() {
+      public Tuple createTuple(String[] columnDatas) {
+        return new VTuple(new Datum[]{
+            new Int4Datum(Integer.parseInt(columnDatas[0])),
+            new TextDatum(columnDatas[1]),
+            new Int4Datum(Integer.parseInt(columnDatas[2])),
+            new TextDatum(columnDatas[3])
+        });
+      }
+    });
+    addEmptyDataFile("nation_multifile", false);
+
+    ResultSet res = executeQuery();
+
+    assertResultSet(res);
+    cleanupQuery(res);
+
+    executeString("DROP TABLE nation_multifile PURGE");
+  }
+
+  @Test
+  public void testMultiplePartitionedBroadcastDataFileWithZeroLength() throws Exception {
+    String tableName = CatalogUtil.normalizeIdentifier("nation_partitioned");
+    ResultSet res = testBase.execute(
+        "create table " + tableName + " (n_name text) partition by column(n_nationkey int4, n_regionkey int4) ");
+    res.close();
+    TajoTestingCluster cluster = testBase.getTestingCluster();
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    res = executeString("insert overwrite into " + tableName
+        + " select n_name, n_nationkey, n_regionkey from nation");
+    res.close();
+
+    addEmptyDataFile("nation_partitioned", true);
+
+    res = executeQuery();
+
+    assertResultSet(res);
+    cleanupQuery(res);
+
+    executeString("DROP TABLE nation_partitioned PURGE");
+  }
+
+  @Test
+  public void testMultiplePartitionedBroadcastDataFileWithZeroLength2() throws Exception {
+    String tableName = CatalogUtil.normalizeIdentifier("nation_partitioned");
+    ResultSet res = testBase.execute(
+        "create table " + tableName + " (n_name text) partition by column(n_nationkey int4, n_regionkey int4) ");
+    res.close();
+    TajoTestingCluster cluster = testBase.getTestingCluster();
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    res = executeString("insert overwrite into " + tableName
+        + " select n_name, n_nationkey, n_regionkey from nation");
+    res.close();
+
+    addEmptyDataFile("nation_partitioned", true);
+
+    res = executeQuery();
+
+    assertResultSet(res);
+    cleanupQuery(res);
+
+    executeString("DROP TABLE nation_partitioned PURGE");
+  }
+
+  private void addEmptyDataFile(String tableName, boolean isPartitioned) throws Exception {
+    TableDesc table = client.getTableDesc(tableName);
+
+    FileSystem fs = table.getPath().getFileSystem(conf);
+    if (isPartitioned) {
+      List<Path> partitionPathList = getPartitionPathList(fs, table.getPath());
+      for (Path eachPath: partitionPathList) {
+        Path dataPath = new Path(eachPath, 0 + "_empty.csv");
+        OutputStream out = fs.create(dataPath);
+        out.close();
+      }
+    } else {
+      Path dataPath = new Path(table.getPath(), 0 + "_empty.csv");
+      OutputStream out = fs.create(dataPath);
+      out.close();
+    }
+  }
+
+  private List<Path> getPartitionPathList(FileSystem fs, Path path) throws Exception {
+    FileStatus[] files = fs.listStatus(path);
+    List<Path> paths = new ArrayList<Path>();
+    if (files != null) {
+      for (FileStatus eachFile: files) {
+        if (eachFile.isFile()) {
+          paths.add(path);
+          return paths;
+        } else {
+          paths.addAll(getPartitionPathList(fs, eachFile.getPath()));
+        }
+      }
+    }
 
-    Path dataPath = new Path(table.getPath(), 999999 + "_empty.csv");
-    FileSystem fs = dataPath.getFileSystem(conf);
-    OutputStream out = fs.create(dataPath);
-    out.close();
+    return paths;
   }
 }

tajo-core/src/test/resources/queries/TestJoinBroadcast/testMultipleBroadcastDataFileWithZeroLength2.sql

@@ -0,0 +1,5 @@
+select b.o_orderkey, b.o_orderdate, b.o_custkey, a.c_custkey, a.c_name, c.n_nationkey, c.n_name
+from customer_large a
+ left outer join orders_large b on a.c_custkey = b.o_custkey
+ left outer join nation_multifile c on a.c_nationkey = c.n_nationkey
+ where c.n_nationkey is not null

tajo-core/src/test/resources/queries/TestJoinBroadcast/testMultiplePartitionedBroadcastDataFileWithZeroLength.sql

@@ -0,0 +1,3 @@
+select * from customer_large a
+ left outer join nation_partitioned b on a.c_nationkey = b.n_nationkey
+ where b.n_nationkey is not null

tajo-core/src/test/resources/queries/TestJoinBroadcast/testMultiplePartitionedBroadcastDataFileWithZeroLength2.sql

@@ -0,0 +1,5 @@
+select b.o_orderkey, b.o_orderdate, b.o_custkey, a.c_custkey, a.c_name, c.n_nationkey, c.n_name
+from customer_large a
+ left outer join orders_large b on a.c_custkey = b.o_custkey
+ left outer join nation_partitioned c on a.c_nationkey = c.n_nationkey
+ where c.n_nationkey is not null