TAJO-2054: Print physical operators in worker webUI.

tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalExec.java

@@ -90,4 +90,9 @@ protected Path getExecutorTmpDir() {
   public TableStats getInputStats() {
     return null;
   }
+
+  @Override
+  public String toString() {
+    return String.format("%s", getClass().getSimpleName());
+  }
 }

tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java

@@ -18,6 +18,7 @@
 
 package org.apache.tajo.engine.planner.physical;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -41,6 +42,7 @@
 import org.apache.tajo.storage.TupleComparator;
 import org.apache.tajo.storage.fragment.FileFragment;
 import org.apache.tajo.storage.fragment.FragmentConvertor;
+import org.apache.tajo.util.TUtil;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -253,4 +255,41 @@ public static void setNullCharIfNecessary(QueryContext context, PersistentStoreN
       }
     }
   }
+
+  /**
+   * A pretty print string of a physical operator and its descendant operators.
+   *
+   */
+  public static void printPhysicalPlan(PhysicalExec exec, StringBuilder builder, int depth) {
+
+    String pad = "  ";
+    if (exec instanceof UnaryPhysicalExec) {
+      UnaryPhysicalExec operator = TUtil.checkTypeAndGet(exec, UnaryPhysicalExec.class);
+
+      builder.append(StringUtils.repeat(pad, depth)).append(operator.toString());
+      builder.append("\n");
+
+      if (operator.getChild() != null) {
+        printPhysicalPlan(operator.getChild(), builder, ++depth);
+      }
+    } else if (exec instanceof BinaryPhysicalExec) {
+      BinaryPhysicalExec operator = TUtil.checkTypeAndGet(exec, BinaryPhysicalExec.class);
+
+      builder.append(StringUtils.repeat(pad, depth)).append(operator.getClass().getSimpleName());
+      builder.append("\n");
+
+      if (operator.getLeftChild() != null) {
+        printPhysicalPlan(operator.getLeftChild(), builder, depth + 1);
+      }
+
+      if (operator.getRightChild() != null) {
+        printPhysicalPlan(operator.getRightChild(), builder, depth + 1);
+      }
+    } else if (exec == null) {
+      return;
+    } else {
+      builder.append(StringUtils.repeat(pad, depth)).append(exec.toString());
+      builder.append("\n");
+    }
+  }
 }

tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java

@@ -363,9 +363,9 @@ public TableStats getInputStats() {
   @Override
   public String toString() {
     if (scanner != null) {
-      return "SeqScanExec:" + plan + "," + scanner.getClass().getName();
+      return String.format("%s:%s(%s)", getClass().getSimpleName(), scanner.getClass().getSimpleName(), plan);
     } else {
-      return "SeqScanExec:" + plan;
+      return String.format("%s(%s)", getClass().getSimpleName(), plan);
     }
   }
 }

tajo-core/src/main/java/org/apache/tajo/worker/TaskHistory.java

@@ -46,6 +46,7 @@ public class TaskHistory implements ProtoObject<TaskHistoryProto>, History {
   private CatalogProtos.TableStatsProto outputStats;
   private String outputPath;
   private String workingPath;
+  private String physicalPlan;
 
   private int finishedFetchCount;
   private int totalFetchCount;
@@ -90,6 +91,10 @@ public TaskHistory(TaskHistoryProto proto) {
       this.totalFetchCount = proto.getTotalFetchCount();
     }
 
+    if(proto.hasPhysicalPlan()) {
+      this.physicalPlan = proto.getPhysicalPlan();
+    }
+
     this.fetcherHistories = proto.getFetcherHistoriesList();
   }
 
@@ -134,6 +139,10 @@ public TaskHistoryProto getProto() {
       builder.setFinishedFetchCount(finishedFetchCount);
     }
 
+    if(physicalPlan != null) {
+      builder.setPhysicalPlan(physicalPlan);
+    }
+
     builder.addAllFetcherHistories(fetcherHistories);
     return builder.build();
   }
@@ -214,6 +223,14 @@ public void setOutputStats(CatalogProtos.TableStatsProto outputStats) {
     this.outputStats = outputStats;
   }
 
+  public String getPhysicalPlan() {
+    return physicalPlan;
+  }
+
+  public void setPhysicalPlan(String physicalPlan) {
+    this.physicalPlan = physicalPlan;
+  }
+
   @Override
   public HistoryType getHistoryType() {
     return HistoryType.TASK;

tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java

@@ -39,6 +39,7 @@
 import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.conf.TajoConf.ConfVars;
 import org.apache.tajo.engine.planner.physical.PhysicalExec;
+import org.apache.tajo.engine.planner.physical.PhysicalPlanUtil;
 import org.apache.tajo.engine.query.QueryContext;
 import org.apache.tajo.engine.query.TaskRequest;
 import org.apache.tajo.ipc.QueryMasterProtocol;
@@ -102,6 +103,7 @@ public class TaskImpl implements Task {
 
   private TupleComparator sortComp = null;
   private final int maxUrlLength;
+  private String physicalPlan;
 
   public TaskImpl(final TaskRequest request,
                   final ExecutionBlockContext executionBlockContext) throws IOException {
@@ -417,6 +419,10 @@ public void run() throws Exception {
         this.executor = executionBlockContext.getTQueryEngine().createPlan(context, plan);
         this.executor.init();
 
+        StringBuilder builder = new StringBuilder();
+        PhysicalPlanUtil.printPhysicalPlan(executor, builder, 0);
+        this.physicalPlan = builder.toString();
+
         while(!context.isStopped() && executor.next() != null) {
         }
       }
@@ -535,6 +541,10 @@ public TaskHistory createTaskHistory() {
         }
         taskHistory.setFinishedFetchCount(i);
       }
+
+      if(physicalPlan != null) {
+        taskHistory.setPhysicalPlan(physicalPlan);
+      }
     } catch (Exception e) {
       LOG.warn(e.getMessage(), e);
     }

tajo-core/src/main/proto/ResourceProtos.proto

@@ -309,4 +309,5 @@ message TaskHistoryProto {
   optional int32 finished_fetch_count = 10;
   optional int32 total_fetch_count = 11;
   repeated FetcherHistoryProto fetcher_histories = 12;
+  optional string physical_plan = 13;
 }

tajo-core/src/main/resources/webapps/worker/taskhistory.jsp

@@ -63,6 +63,7 @@
   <h2>Tajo Worker: <a href='index.jsp'><%=tajoWorker.getWorkerContext().getWorkerName()%></a></h2>
   <hr/>
   <h3>Task Detail: <%=request.getParameter("taskAttemptId")%></h3>
+  <pre style="white-space:pre-wrap;"><%= taskHistory.getPhysicalPlan() == null ? "" : taskHistory.getPhysicalPlan()%></pre>
   <table border="1" width="100%" class="border_table">
       <tr><td width="200" align="right">ID</td><td><%=request.getParameter("taskAttemptId")%></td></tr>
       <tr><td align="right">State</td><td><%=taskHistory.getState()%></td></tr>