TAJO-1352: Improve the join order algorithm to consider missed cases of associative join operators

tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/SchemaUtil.java

@@ -161,4 +161,64 @@ public static String toDisplayString(Schema schema) {
     DDLBuilder.buildSchema(sb, schema);
     return sb.toString();
   }
+
+  /**
+   * Calculate the row size from the given schema.
+   *
+   * @param schema input schema
+   * @return estimated row size in bytes
+   */
+  public static int estimateRowByteSizeWithSchema(Schema schema) {
+    int size = 0;
+    for (Column column : schema.fields) {
+      size += getColByteSize(column);
+    }
+    return size;
+  }
+
+  /**
+   * Return the size of the given column. For the variable-length columns, it returns a prefixed value.
+   *
+   * @param col input column
+   * @return column length in bytes
+   */
+  public static int getColByteSize(Column col) {
+    if (col.getDataType().getLength() > 0) {
+      return col.getDataType().getLength();
+    }
+    switch (col.getDataType().getType()) {
+      case BOOLEAN:
+        return 1;
+      case CHAR:
+        return 1;
+      case BIT:
+        return 1;
+      case INT2:
+        return 2;
+      case INT4:
+        return 4;
+      case INT8:
+        return 8;
+      case FLOAT4:
+        return 4;
+      case FLOAT8:
+        return 8;
+      case INET4:
+        return 4;
+      case INET6:
+        return 32;
+      case TEXT:
+        return 256;
+      case BLOB:
+        return 256;
+      case DATE:
+        return 4;
+      case TIME:
+        return 8;
+      case TIMESTAMP:
+        return 8;
+      default:
+        return 0;
+    }
+  }
 }

tajo-common/src/main/java/org/apache/tajo/util/Pair.java

@@ -29,6 +29,11 @@ public Pair(T1 first, T2 second) {
     this.second = second;
   }
 
+  public void set(T1 first, T2 second) {
+    this.first = first;
+    this.second = second;
+  }
+
   public T1 getFirst() {
     return first;
   }

tajo-core/pom.xml

@@ -78,7 +78,6 @@
             <exclude>src/test/resources/dataset/**</exclude>
             <exclude>src/test/resources/queries/**</exclude>
             <exclude>src/test/resources/results/**</exclude>
-            <exclude>src/test/resources/results/**</exclude>
             <exclude>src/main/resources/META-INF/services/*</exclude>
             <exclude>src/main/resources/webapps/static/js/*</exclude>
           </excludes>

tajo-core/src/main/java/org/apache/tajo/master/exec/DDLExecutor.java

@@ -39,9 +39,9 @@
 import org.apache.tajo.plan.LogicalPlan;
 import org.apache.tajo.plan.logical.*;
 import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.storage.TablespaceManager;
-import org.apache.tajo.storage.Tablespace;
 import org.apache.tajo.storage.StorageUtil;
+import org.apache.tajo.storage.Tablespace;
+import org.apache.tajo.storage.TablespaceManager;
 import org.apache.tajo.util.Pair;
 
 import java.io.IOException;
@@ -288,7 +288,6 @@ public TableDesc createTable(QueryContext queryContext,
    * @param purge     Remove all data if purge is true.
    */
   public boolean dropTable(QueryContext queryContext, String tableName, boolean ifExists, boolean purge) {
-    CatalogService catalog = context.getCatalog();
 
     String databaseName;
     String simpleTableName;

tajo-core/src/main/java/org/apache/tajo/master/exec/ExplainPlanPreprocessorForTest.java

@@ -27,11 +27,11 @@
 import org.apache.tajo.plan.expr.AlgebraicUtil;
 import org.apache.tajo.plan.expr.EvalNode;
 import org.apache.tajo.plan.logical.*;
-import org.apache.tajo.plan.util.PlannerUtil;
 import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor;
-import org.apache.tajo.util.TUtil;
 
-import java.util.*;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Stack;
 
 /**
  * Tajo's logical planner can generate different shapes of logical plans for the same query,
@@ -41,26 +41,11 @@
 public class ExplainPlanPreprocessorForTest {
   private static final PlanShapeFixerContext shapeFixerContext = new PlanShapeFixerContext();
   private static final PlanShapeFixer shapeFixer = new PlanShapeFixer();
-  private static final PidCollectorContext collectorContext = new PidCollectorContext();
-  private static final JoinPidCollector joinPidCollector = new JoinPidCollector();
-  private static final PidReseterContext resetContext = new PidReseterContext();
-  private static final JoinPidReseter joinPidReseter = new JoinPidReseter();
 
   public void prepareTest(LogicalPlan plan) throws PlanningException {
     // Plan shape fixer
     shapeFixerContext.reset();
     shapeFixer.visit(shapeFixerContext, plan, plan.getRootBlock());
-
-    /*
-     * During join order optimization, new join nodes are created based on the chosen join order.
-     * So, they have different pids for each query execution.
-     * JoinPidCollector and JoinPidReseter reset the pids of join nodes.
-     */
-    collectorContext.reset();
-    joinPidCollector.visit(collectorContext, plan, plan.getRootBlock());
-
-    resetContext.reset(collectorContext.joinPids);
-    joinPidReseter.visit(resetContext, plan, plan.getRootBlock());
   }
 
   private static class PlanShapeFixerContext {
@@ -150,18 +135,6 @@ public LogicalNode visitJoin(PlanShapeFixerContext context, LogicalPlan plan, Lo
       int rightChildNum = context.childNumbers.pop();
       int leftChildNum = context.childNumbers.pop();
 
-      if (PlannerUtil.isCommutativeJoin(node.getJoinType())) {
-
-        if (leftChildNum < rightChildNum) {
-          swapChildren(node);
-        } else if (leftChildNum == rightChildNum) {
-          if (node.getLeftChild().getOutSchema().toString().compareTo(node.getRightChild().getOutSchema().toString()) <
-              0) {
-            swapChildren(node);
-          }
-        }
-      }
-
       if (node.hasTargets()) {
         node.setTargets(sortTargets(node.getTargets()));
       }
@@ -200,12 +173,6 @@ private Target[] sortTargets(Target[] targets) {
       Arrays.sort(targets, targetComparator);
       return targets;
     }
-
-    private static void swapChildren(JoinNode node) {
-      LogicalNode tmpChild = node.getLeftChild();
-      node.setLeftChild(node.getRightChild());
-      node.setRightChild(tmpChild);
-    }
   }
 
   public static class ColumnComparator implements Comparator<Column> {
@@ -232,52 +199,4 @@ public int compare(Target o1, Target o2) {
     }
   }
 
-  private static class PidCollectorContext {
-    List<Integer> joinPids = TUtil.newList();
-    public void reset() {
-      joinPids.clear();
-    }
-  }
-
-  /**
-   * {@link JoinPidCollector} collects the pids of all join
-   * nodes.
-   */
-  private static class JoinPidCollector extends BasicLogicalPlanVisitor<PidCollectorContext, LogicalNode> {
-
-    @Override
-    public LogicalNode visitJoin(PidCollectorContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
-                                 JoinNode node, Stack<LogicalNode> stack) throws PlanningException {
-      context.joinPids.add(node.getPID());
-      super.visitJoin(context, plan, block, node, stack);
-
-      return null;
-    }
-  }
-
-  private static class PidReseterContext {
-    List<Integer> joinPids;
-
-    public void reset(List<Integer> joinPids) {
-      this.joinPids = joinPids;
-      Collections.sort(this.joinPids);
-    }
-  }
-
-  /**
-   * {@link JoinPidReseter} resets pids of join nodes with the pids collected by {@link JoinPidCollector} in ascending
-   * order while traversing the query plan.
-   */
-  private static class JoinPidReseter extends BasicLogicalPlanVisitor<PidReseterContext, LogicalNode> {
-
-    @Override
-    public LogicalNode visitJoin(PidReseterContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
-                                 JoinNode node, Stack<LogicalNode> stack) throws PlanningException {
-      super.visitJoin(context, plan, block, node, stack);
-      node.setPID(context.joinPids.remove(0));
-
-      return null;
-    }
-  }
-
 }

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

@@ -50,7 +50,7 @@ public static void classTearDown() throws ServiceException {
   }
 
   @Test
-  @Option(withExplain = true, withExplainGlobal = true, parameterized = true)
+  @Option(withExplain = true, withExplainGlobal = true, parameterized = true, sort = true)
   @SimpleTest(queries = {
       @QuerySpec("select n_name, r_name, n_regionkey, r_regionkey from nation, region order by n_name, r_name"),
       // testCrossJoinWithAsterisk
@@ -64,7 +64,7 @@ public final void testCrossJoin() throws Exception {
   }
 
   @Test
-  @Option(withExplain = true, withExplainGlobal = true, parameterized = true)
+  @Option(withExplain = true, withExplainGlobal = true, parameterized = true, sort = true)
   @SimpleTest()
   public final void testCrossJoinWithThetaJoinConditionInWhere() throws Exception {
     runSimpleTests();

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

@@ -348,7 +348,6 @@ public final void testLeftOuterWithEmptyTable() throws Exception {
     runSimpleTests();
   }
 
-  // TODO: this test is disabled due to a bug in broadcast join. It will be enabled after TAJO-1553
   @Test
   @Option(withExplain = true, withExplainGlobal = true, parameterized = true)
   @SimpleTest(queries = {
@@ -363,7 +362,6 @@ public final void testRightOuterJoinPredicationCaseByCase1() throws Exception {
     runSimpleTests();
   }
 
-  // TODO: this test is disabled due to a bug in broadcast join. It will be enabled after TAJO-1553
   @Test
   @Option(withExplain = true, withExplainGlobal = true, parameterized = true)
   @SimpleTest(queries = {
@@ -382,7 +380,6 @@ public final void testRightOuterJoinPredicationCaseByCase2() throws Exception {
     runSimpleTests();
   }
 
-  // TODO: this test is disabled due to a bug in broadcast join. It will be enabled after TAJO-1553
   @Test
   @Option(withExplain = true, withExplainGlobal = true, parameterized = true)
   @SimpleTest(queries = {
@@ -397,7 +394,6 @@ public final void testRightOuterJoinPredicationCaseByCase3() throws Exception {
     runSimpleTests();
   }
 
-  // TODO: this test is disabled due to a bug in broadcast join. It will be enabled after TAJO-1553
   @Test
   @Option(withExplain = true, withExplainGlobal = true, parameterized = true, sort = true)
   @SimpleTest(queries = {

tajo-core/src/test/java/org/apache/tajo/querymaster/TestTaskStatusUpdate.java

@@ -95,15 +95,15 @@ public final void case3() throws Exception {
       |-eb_1404143727281_0002_000005
          |-eb_1404143727281_0002_000004        (order by)
             |-eb_1404143727281_0002_000003     (join)
-               |-eb_1404143727281_0002_000002  (scan)
-               |-eb_1404143727281_0002_000001  (scan, filter)
+               |-eb_1404143727281_0002_000002  (scan, filter)
+               |-eb_1404143727281_0002_000001  (scan)
        */
       res = executeQuery();
 
       // in/out * stage(4)
-      long[] expectedNumRows = new long[]{2, 2, 5, 5, 7, 2, 2, 2};
-      long[] expectedNumBytes = new long[]{8, 34, 20, 75, 109, 34, 34, 18};
-      long[] expectedReadBytes = new long[]{8, 8, 20, 20, 109, 0, 34, 0};
+      long[] expectedNumRows = new long[]{5, 5, 2, 2, 7, 2, 2, 2};
+      long[] expectedNumBytes = new long[]{20, 75, 8, 34, 109, 34, 34, 18};
+      long[] expectedReadBytes = new long[]{20, 20, 8, 8, 109, 0, 34, 0};
 
       assertStatus(4, expectedNumRows, expectedNumBytes, expectedReadBytes);
     } finally {

tajo-core/src/test/resources/queries/TestInnerJoinWithSubQuery/testBroadcastSubquery2.sql

@@ -1,6 +1,6 @@
 select sum(b.l_quantity)
 from (
-      select a.l_orderkey, a.l_quantity
+      select a.l_orderkey, a.l_quantity, a.l_linenumber
         from lineitem a
         join part on a.l_partkey = p_partkey) b
 join orders c on c.o_orderkey = b.l_orderkey

tajo-core/src/test/resources/queries/TestOuterJoinQuery/testLeftOuterJoin2.sql

@@ -2,4 +2,4 @@ select a.l_orderkey, b.c_custkey, b.c_name, b.c_nationkey, d.n_name
 from lineitem a
 left outer join customer b on a.l_orderkey = b.c_custkey
 left outer join orders c on b.c_custkey = c.o_custkey
-left outer join nation d on a.l_orderkey = d.n_nationkey
+left outer join nation d on a.l_orderkey = d.n_nationkey

tajo-core/src/test/resources/results/TestInnerJoinQuery/testBroadcastTwoPartJoin.Hash.plan

@@ -34,14 +34,14 @@ SORT(6)
            => target list: default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.orders.o_custkey (INT4)
            => out schema: {(3) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.orders.o_custkey (INT4)}
            => in schema: {(4) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.orders.o_custkey (INT4), default.orders.o_orderkey (INT4)}
-            SCAN(0) on default.lineitem
-              => target list: default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)
-              => out schema: {(2) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)}
-              => in schema: {(16) default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)}
             SCAN(1) on default.orders
               => target list: default.orders.o_custkey (INT4), default.orders.o_orderkey (INT4)
               => out schema: {(2) default.orders.o_custkey (INT4), default.orders.o_orderkey (INT4)}
               => in schema: {(9) default.orders.o_clerk (TEXT), default.orders.o_comment (TEXT), default.orders.o_custkey (INT4), default.orders.o_orderdate (TEXT), default.orders.o_orderkey (INT4), default.orders.o_orderpriority (TEXT), default.orders.o_orderstatus (TEXT), default.orders.o_shippriority (INT4), default.orders.o_totalprice (FLOAT8)}
+            SCAN(0) on default.lineitem
+              => target list: default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)
+              => out schema: {(2) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)}
+              => in schema: {(16) default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)}
 explain
 -------------------------------
 -------------------------------------------------------------------------------
@@ -105,14 +105,14 @@ SORT(26)
            => target list: default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.orders.o_custkey (INT4)
            => out schema: {(3) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.orders.o_custkey (INT4)}
            => in schema: {(4) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.orders.o_custkey (INT4), default.orders.o_orderkey (INT4)}
-            SCAN(0) on default.lineitem
-              => target list: default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)
-              => out schema: {(2) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)}
-              => in schema: {(16) default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)}
             SCAN(1) on default.orders
               => target list: default.orders.o_custkey (INT4), default.orders.o_orderkey (INT4)
               => out schema: {(2) default.orders.o_custkey (INT4), default.orders.o_orderkey (INT4)}
               => in schema: {(9) default.orders.o_clerk (TEXT), default.orders.o_comment (TEXT), default.orders.o_custkey (INT4), default.orders.o_orderdate (TEXT), default.orders.o_orderkey (INT4), default.orders.o_orderpriority (TEXT), default.orders.o_orderstatus (TEXT), default.orders.o_shippriority (INT4), default.orders.o_totalprice (FLOAT8)}
+            SCAN(0) on default.lineitem
+              => target list: default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)
+              => out schema: {(2) default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4)}
+              => in schema: {(16) default.lineitem.l_comment (TEXT), default.lineitem.l_commitdate (TEXT), default.lineitem.l_discount (FLOAT8), default.lineitem.l_extendedprice (FLOAT8), default.lineitem.l_linenumber (INT4), default.lineitem.l_linestatus (TEXT), default.lineitem.l_orderkey (INT4), default.lineitem.l_partkey (INT4), default.lineitem.l_quantity (FLOAT8), default.lineitem.l_receiptdate (TEXT), default.lineitem.l_returnflag (TEXT), default.lineitem.l_shipdate (TEXT), default.lineitem.l_shipinstruct (TEXT), default.lineitem.l_shipmode (TEXT), default.lineitem.l_suppkey (INT4), default.lineitem.l_tax (FLOAT8)}
 
 =======================================================
 Block Id: eb_0000000000000_0000_000010 [ROOT]