tajo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jihoon...@apache.org
Subject [2/6] tajo git commit: TAJO-1026: Implement Query history persistency manager.
Date Tue, 11 Nov 2014 14:24:47 GMT
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/QueryHistory.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/QueryHistory.java b/tajo-core/src/main/java/org/apache/tajo/util/history/QueryHistory.java
new file mode 100644
index 0000000..7a81b4b
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/QueryHistory.java
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.util.history;
+
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.engine.json.CoreGsonHelper;
+import org.apache.tajo.ipc.ClientProtos.QueryHistoryProto;
+import org.apache.tajo.ipc.ClientProtos.SubQueryHistoryProto;
+import org.apache.tajo.json.GsonObject;
+import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.KeyValueProto;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class QueryHistory implements GsonObject, History {
+  @Expose
+  private String queryId;
+  @Expose
+  private String queryMaster;
+  @Expose
+  private int httpPort;
+  @Expose
+  private List<String[]> sessionVariables;
+  @Expose
+  private String logicalPlan;
+  @Expose
+  private String distributedPlan;
+  @Expose
+  private List<SubQueryHistory> subQueryHistories;
+
+  public String getQueryId() {
+    return queryId;
+  }
+
+  public void setQueryId(String queryId) {
+    this.queryId = queryId;
+  }
+
+  public void setQueryMaster(String queryMaster) {
+    this.queryMaster = queryMaster;
+  }
+
+  public void setSubQueryHistories(List<SubQueryHistory> subQueryHistories) {
+    this.subQueryHistories = subQueryHistories;
+  }
+
+  public String getQueryMaster() {
+    return queryMaster;
+  }
+
+  public int getHttpPort() {
+    return httpPort;
+  }
+
+  public void setHttpPort(int httpPort) {
+    this.httpPort = httpPort;
+  }
+
+  public List<SubQueryHistory> getSubQueryHistories() {
+    return subQueryHistories;
+  }
+
+  public List<String[]> getSessionVariables() {
+    return sessionVariables;
+  }
+
+  public String getLogicalPlan() {
+    return logicalPlan;
+  }
+
+  public String getDistributedPlan() {
+    return distributedPlan;
+  }
+
+  public void setSessionVariables(List<String[]> sessionVariables) {
+    this.sessionVariables = sessionVariables;
+  }
+
+  public void setLogicalPlan(String logicalPlan) {
+    this.logicalPlan = logicalPlan;
+  }
+
+  public void setDistributedPlan(String distributedPlan) {
+    this.distributedPlan = distributedPlan;
+  }
+
+  @Override
+  public String toJson() {
+    return CoreGsonHelper.toJson(this, QueryHistory.class);
+  }
+
+  @Override
+  public HistoryType getHistoryType() {
+    return HistoryType.QUERY;
+  }
+
+  public static QueryHistory fromJson(String json) {
+    return CoreGsonHelper.fromJson(json, QueryHistory.class);
+  }
+
+  public QueryHistoryProto getProto() {
+    QueryHistoryProto.Builder builder = QueryHistoryProto.newBuilder();
+
+    builder.setQueryId(queryId)
+      .setQueryMaster(queryMaster)
+      .setHttpPort(httpPort)
+      .setLogicalPlan(logicalPlan)
+      .setDistributedPlan(distributedPlan);
+
+    List<KeyValueProto> sessionProtos = new ArrayList<KeyValueProto>();
+
+    if (sessionVariables != null) {
+      KeyValueProto.Builder keyValueBuilder = KeyValueProto.newBuilder();
+      for (String[] eachSessionVal: sessionVariables) {
+        keyValueBuilder.clear();
+        keyValueBuilder.setKey(eachSessionVal[0]);
+        keyValueBuilder.setValue(eachSessionVal[1]);
+
+        sessionProtos.add(keyValueBuilder.build());
+      }
+    }
+    builder.addAllSessionVariables(sessionProtos);
+
+
+    List<SubQueryHistoryProto> subQueryHistoryProtos = new ArrayList<SubQueryHistoryProto>();
+    if (subQueryHistories != null) {
+      for (SubQueryHistory eachSubQuery: subQueryHistories) {
+        subQueryHistoryProtos.add((eachSubQuery.getProto()));
+      }
+    }
+    builder.addAllSubQueryHistories(subQueryHistoryProtos);
+
+    return builder.build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/QueryUnitHistory.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/QueryUnitHistory.java b/tajo-core/src/main/java/org/apache/tajo/util/history/QueryUnitHistory.java
new file mode 100644
index 0000000..556a971
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/QueryUnitHistory.java
@@ -0,0 +1,167 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.util.history;
+
+import com.google.gson.annotations.Expose;
+import org.apache.tajo.engine.json.CoreGsonHelper;
+import org.apache.tajo.json.GsonObject;
+
+public class QueryUnitHistory implements GsonObject {
+  @Expose private String id;
+  @Expose private String hostAndPort;
+  @Expose private int httpPort;
+  @Expose private String state;
+  @Expose private float progress;
+  @Expose private long launchTime;
+  @Expose private long finishTime;
+  @Expose private int retryCount;
+
+  @Expose private int numShuffles;
+  @Expose private String shuffleKey;
+  @Expose private String shuffleFileName;
+
+  @Expose private String[] dataLocations;
+  @Expose private String[] fragments;
+  @Expose private String[][] fetchs;
+
+  public String getId() {
+    return id;
+  }
+
+  public void setId(String id) {
+    this.id = id;
+  }
+
+  public String getHostAndPort() {
+    return hostAndPort;
+  }
+
+  public void setHostAndPort(String hostAndPort) {
+    this.hostAndPort = hostAndPort;
+  }
+
+  public int getHttpPort() {
+    return httpPort;
+  }
+
+  public void setHttpPort(int httpPort) {
+    this.httpPort = httpPort;
+  }
+
+  public int getRetryCount() {
+    return retryCount;
+  }
+
+  public void setRetryCount(int retryCount) {
+    this.retryCount = retryCount;
+  }
+
+  public String getState() {
+    return state;
+  }
+
+  public void setState(String state) {
+    this.state = state;
+  }
+
+  public float getProgress() {
+    return progress;
+  }
+
+  public void setProgress(float progress) {
+    this.progress = progress;
+  }
+
+  public long getLaunchTime() {
+    return launchTime;
+  }
+
+  public void setLaunchTime(long launchTime) {
+    this.launchTime = launchTime;
+  }
+
+  public long getFinishTime() {
+    return finishTime;
+  }
+
+  public void setFinishTime(long finishTime) {
+    this.finishTime = finishTime;
+  }
+
+  public int getNumShuffles() {
+    return numShuffles;
+  }
+
+  public void setNumShuffles(int numShuffles) {
+    this.numShuffles = numShuffles;
+  }
+
+  public String getShuffleKey() {
+    return shuffleKey;
+  }
+
+  public void setShuffleKey(String shuffleKey) {
+    this.shuffleKey = shuffleKey;
+  }
+
+  public String getShuffleFileName() {
+    return shuffleFileName;
+  }
+
+  public void setShuffleFileName(String shuffleFileName) {
+    this.shuffleFileName = shuffleFileName;
+  }
+
+  public String[] getDataLocations() {
+    return dataLocations;
+  }
+
+  public void setDataLocations(String[] dataLocations) {
+    this.dataLocations = dataLocations;
+  }
+
+  public String[] getFragments() {
+    return fragments;
+  }
+
+  public void setFragments(String[] fragments) {
+    this.fragments = fragments;
+  }
+
+  public String[][] getFetchs() {
+    return fetchs;
+  }
+
+  public void setFetchs(String[][] fetchs) {
+    this.fetchs = fetchs;
+  }
+
+  @Override
+  public String toJson() {
+    return CoreGsonHelper.toJson(this, QueryUnitHistory.class);
+  }
+
+  public long getRunningTime() {
+    if(finishTime > 0) {
+      return finishTime - launchTime;
+    } else {
+      return 0;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/SubQueryHistory.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/SubQueryHistory.java b/tajo-core/src/main/java/org/apache/tajo/util/history/SubQueryHistory.java
new file mode 100644
index 0000000..b3ac4d2
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/SubQueryHistory.java
@@ -0,0 +1,270 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.util.history;
+
+import com.google.gson.annotations.Expose;
+import com.google.gson.reflect.TypeToken;
+import org.apache.tajo.engine.json.CoreGsonHelper;
+import org.apache.tajo.ipc.ClientProtos.SubQueryHistoryProto;
+import org.apache.tajo.json.GsonObject;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SubQueryHistory implements GsonObject {
+  @Expose
+  private String executionBlockId;
+  @Expose
+  private String state;
+  @Expose
+  private long startTime;
+  @Expose
+  private long finishTime;
+  @Expose
+  private int succeededObjectCount;
+  @Expose
+  private int failedObjectCount;
+  @Expose
+  private int killedObjectCount;
+  @Expose
+  private int totalScheduledObjectsCount;
+
+  @Expose
+  private long totalInputBytes;
+  @Expose
+  private long totalReadBytes;
+  @Expose
+  private long totalReadRows;
+  @Expose
+  private long totalWriteBytes;
+  @Expose
+  private long totalWriteRows;
+  @Expose
+  private int numShuffles;
+  @Expose
+  private float progress;
+
+  @Expose
+  private String plan;
+  @Expose
+  private int hostLocalAssigned;
+  @Expose
+  private int rackLocalAssigned;
+
+  private List<QueryUnitHistory> queryUnits;
+
+  public String getExecutionBlockId() {
+    return executionBlockId;
+  }
+
+  public void setExecutionBlockId(String executionBlockId) {
+    this.executionBlockId = executionBlockId;
+  }
+
+  public String getState() {
+    return state;
+  }
+
+  public void setState(String state) {
+    this.state = state;
+  }
+
+  public long getStartTime() {
+    return startTime;
+  }
+
+  public void setStartTime(long startTime) {
+    this.startTime = startTime;
+  }
+
+  public long getFinishTime() {
+    return finishTime;
+  }
+
+  public void setFinishTime(long finishTime) {
+    this.finishTime = finishTime;
+  }
+
+  public int getSucceededObjectCount() {
+    return succeededObjectCount;
+  }
+
+  public void setSucceededObjectCount(int succeededObjectCount) {
+    this.succeededObjectCount = succeededObjectCount;
+  }
+
+  public int getTotalScheduledObjectsCount() {
+    return totalScheduledObjectsCount;
+  }
+
+  public void setTotalScheduledObjectsCount(int totalScheduledObjectsCount) {
+    this.totalScheduledObjectsCount = totalScheduledObjectsCount;
+  }
+
+  public long getTotalInputBytes() {
+    return totalInputBytes;
+  }
+
+  public void setTotalInputBytes(long totalInputBytes) {
+    this.totalInputBytes = totalInputBytes;
+  }
+
+  public long getTotalReadBytes() {
+    return totalReadBytes;
+  }
+
+  public void setTotalReadBytes(long totalReadBytes) {
+    this.totalReadBytes = totalReadBytes;
+  }
+
+  public long getTotalReadRows() {
+    return totalReadRows;
+  }
+
+  public void setTotalReadRows(long totalReadRows) {
+    this.totalReadRows = totalReadRows;
+  }
+
+  public long getTotalWriteBytes() {
+    return totalWriteBytes;
+  }
+
+  public void setTotalWriteBytes(long totalWriteBytes) {
+    this.totalWriteBytes = totalWriteBytes;
+  }
+
+  public long getTotalWriteRows() {
+    return totalWriteRows;
+  }
+
+  public void setTotalWriteRows(long totalWriteRows) {
+    this.totalWriteRows = totalWriteRows;
+  }
+
+  public int getNumShuffles() {
+    return numShuffles;
+  }
+
+  public void setNumShuffles(int numShuffles) {
+    this.numShuffles = numShuffles;
+  }
+
+  public float getProgress() {
+    return progress;
+  }
+
+  public void setProgress(float progress) {
+    this.progress = progress;
+  }
+
+  public String getPlan() {
+    return plan;
+  }
+
+  public void setPlan(String plan) {
+    this.plan = plan;
+  }
+
+  public int getHostLocalAssigned() {
+    return hostLocalAssigned;
+  }
+
+  public void setHostLocalAssigned(int hostLocalAssigned) {
+    this.hostLocalAssigned = hostLocalAssigned;
+  }
+
+  public int getRackLocalAssigned() {
+    return rackLocalAssigned;
+  }
+
+  public void setRackLocalAssigned(int rackLocalAssigned) {
+    this.rackLocalAssigned = rackLocalAssigned;
+  }
+
+  public int getFailedObjectCount() {
+    return failedObjectCount;
+  }
+
+  public void setFailedObjectCount(int failedObjectCount) {
+    this.failedObjectCount = failedObjectCount;
+  }
+
+  public int getKilledObjectCount() {
+    return killedObjectCount;
+  }
+
+  public void setKilledObjectCount(int killedObjectCount) {
+    this.killedObjectCount = killedObjectCount;
+  }
+
+  public List<QueryUnitHistory> getQueryUnits() {
+    return queryUnits;
+  }
+
+  public void setQueryUnits(List<QueryUnitHistory> queryUnits) {
+    this.queryUnits = queryUnits;
+  }
+
+  @Override
+  public String toJson() {
+    return CoreGsonHelper.toJson(this, SubQueryHistory.class);
+  }
+
+  public String toQueryUnitsJson() {
+    if (queryUnits == null) {
+      return "";
+    }
+    return CoreGsonHelper.getInstance().toJson(queryUnits, new TypeToken<List<QueryUnitHistory>>() {
+    }.getType());
+  }
+
+  public static List<QueryUnitHistory> fromJsonQueryUnits(String json) {
+    if (json == null || json.trim().isEmpty()) {
+      return new ArrayList<QueryUnitHistory>();
+    }
+    return CoreGsonHelper.getInstance().fromJson(json, new TypeToken<List<QueryUnitHistory>>() {
+    }.getType());
+  }
+
+  public SubQueryHistoryProto getProto() {
+    SubQueryHistoryProto.Builder builder = SubQueryHistoryProto.newBuilder();
+    builder.setExecutionBlockId(executionBlockId)
+      .setState(state)
+      .setStartTime(startTime)
+      .setFinishTime(finishTime)
+      .setSucceededObjectCount(succeededObjectCount)
+      .setFailedObjectCount(failedObjectCount)
+      .setKilledObjectCount(killedObjectCount)
+      .setTotalScheduledObjectsCount(totalScheduledObjectsCount)
+
+      .setTotalInputBytes(totalInputBytes)
+      .setTotalReadBytes(totalReadBytes)
+      .setTotalReadRows(totalReadRows)
+      .setTotalWriteBytes(totalWriteBytes)
+      .setTotalWriteRows(totalWriteRows)
+      .setNumShuffles(numShuffles)
+      .setProgress(progress)
+
+      .setPlan(plan)
+      .setHostLocalAssigned(hostLocalAssigned)
+      .setRackLocalAssigned(rackLocalAssigned);
+
+    return builder.build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
index b3dae8b..ddd7e13 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -45,6 +45,8 @@ import org.apache.tajo.rpc.RpcConnectionPool;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
 import org.apache.tajo.storage.HashShuffleAppenderManager;
 import org.apache.tajo.util.*;
+import org.apache.tajo.util.history.HistoryReader;
+import org.apache.tajo.util.history.HistoryWriter;
 import org.apache.tajo.util.metrics.TajoSystemMetrics;
 import org.apache.tajo.webapp.StaticHttpServer;
 
@@ -129,6 +131,10 @@ public class TajoWorker extends CompositeService {
 
   private JvmPauseMonitor pauseMonitor;
 
+  private HistoryWriter taskHistoryWriter;
+
+  private HistoryReader historyReader;
+
   public TajoWorker() throws Exception {
     super(TajoWorker.class.getName());
   }
@@ -253,6 +259,12 @@ public class TajoWorker extends CompositeService {
       LOG.fatal(e.getMessage(), e);
       System.exit(-1);
     }
+
+    taskHistoryWriter = new HistoryWriter(workerContext.getWorkerName(), false);
+    addIfService(taskHistoryWriter);
+    taskHistoryWriter.init(conf);
+
+    historyReader = new HistoryReader(workerContext.getWorkerName(), this.systemConf);
   }
 
   private void initWorkerMetrics() {
@@ -541,6 +553,14 @@ public class TajoWorker extends CompositeService {
     public HashShuffleAppenderManager getHashShuffleAppenderManager() {
       return hashShuffleAppenderManager;
     }
+
+    public HistoryWriter getTaskHistoryWriter() {
+      return taskHistoryWriter;
+    }
+
+    public HistoryReader getHistoryReader() {
+      return historyReader;
+    }
   }
 
   private int getStandAlonePullServerPort() {

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
index fb4f861..a41ffce 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
@@ -33,13 +33,19 @@ import org.apache.tajo.TajoIdProtos;
 import org.apache.tajo.TajoProtos;
 import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.ClientProtos.GetQueryHistoryResponse;
+import org.apache.tajo.ipc.ClientProtos.QueryIdRequest;
+import org.apache.tajo.ipc.ClientProtos.ResultCode;
 import org.apache.tajo.ipc.QueryMasterClientProtocol;
 import org.apache.tajo.ipc.TajoWorkerProtocol;
 import org.apache.tajo.master.querymaster.Query;
+import org.apache.tajo.master.querymaster.QueryInProgress;
+import org.apache.tajo.master.querymaster.QueryJobManager;
 import org.apache.tajo.master.querymaster.QueryMasterTask;
 import org.apache.tajo.rpc.BlockingRpcServer;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
 import org.apache.tajo.util.NetUtils;
+import org.apache.tajo.util.history.QueryHistory;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -229,5 +235,33 @@ public class TajoWorkerClientService extends AbstractService {
       LOG.info("Stop Query:" + queryId);
       return BOOL_TRUE;
     }
+
+    @Override
+    public GetQueryHistoryResponse getQueryHistory(RpcController controller, QueryIdRequest request) throws ServiceException {
+      GetQueryHistoryResponse.Builder builder = GetQueryHistoryResponse.newBuilder();
+
+      try {
+        QueryId queryId = new QueryId(request.getQueryId());
+
+        QueryMasterTask queryMasterTask = workerContext.getQueryMaster().getQueryMasterTask(queryId);
+        QueryHistory queryHistory = null;
+        if (queryMasterTask == null) {
+          queryHistory = workerContext.getHistoryReader().getQueryHistory(queryId.toString());
+        } else {
+          queryHistory = queryMasterTask.getQuery().getQueryHistory();
+        }
+
+        if (queryHistory != null) {
+          builder.setQueryHistory(queryHistory.getProto());
+        }
+        builder.setResultCode(ResultCode.OK);
+      } catch (Throwable t) {
+        LOG.warn(t.getMessage(), t);
+        builder.setResultCode(ResultCode.ERROR);
+        builder.setErrorMessage(org.apache.hadoop.util.StringUtils.stringifyException(t));
+      }
+
+      return builder.build();
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
index a7b52c7..00edc79 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/Task.java
@@ -508,7 +508,8 @@ public class Task {
   }
 
   public void cleanupTask() {
-    executionBlockContext.addTaskHistory(taskRunnerId, getId(), createTaskHistory());
+    TaskHistory taskHistory = createTaskHistory();
+    executionBlockContext.addTaskHistory(taskRunnerId, getId(), taskHistory);
     executionBlockContext.getTasks().remove(getId());
 
     fetcherRunners.clear();
@@ -521,6 +522,8 @@ public class Task {
     } catch (IOException e) {
       LOG.fatal(e.getMessage(), e);
     }
+
+    executionBlockContext.getWorkerContext().getTaskHistoryWriter().appendHistory(taskHistory);
   }
 
   public TaskHistory createTaskHistory() {

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/worker/TaskHistory.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskHistory.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskHistory.java
index dab6ba3..9b6fd0d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskHistory.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskHistory.java
@@ -23,6 +23,7 @@ import com.google.common.collect.Lists;
 import org.apache.tajo.QueryUnitAttemptId;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.common.ProtoObject;
+import org.apache.tajo.util.history.History;
 
 import java.util.Collections;
 import java.util.List;
@@ -34,7 +35,7 @@ import static org.apache.tajo.ipc.TajoWorkerProtocol.TaskHistoryProto;
 /**
  * The history class for Task processing.
  */
-public class TaskHistory implements ProtoObject<TaskHistoryProto> {
+public class TaskHistory implements ProtoObject<TaskHistoryProto>, History {
 
   private QueryUnitAttemptId queryUnitAttemptId;
   private TaskAttemptState state;
@@ -212,4 +213,9 @@ public class TaskHistory implements ProtoObject<TaskHistoryProto> {
   public void setOutputStats(CatalogProtos.TableStatsProto outputStats) {
     this.outputStats = outputStats;
   }
+
+  @Override
+  public HistoryType getHistoryType() {
+    return HistoryType.TASK;
+  }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/admin/query.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/admin/query.jsp b/tajo-core/src/main/resources/webapps/admin/query.jsp
index 9ddc90c..5afb3b2 100644
--- a/tajo-core/src/main/resources/webapps/admin/query.jsp
+++ b/tajo-core/src/main/resources/webapps/admin/query.jsp
@@ -20,7 +20,6 @@
 <%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
 
 <%@ page import="org.apache.tajo.master.TajoMaster" %>
-<%@ page import="org.apache.tajo.master.ha.HAService" %>
 <%@ page import="org.apache.tajo.master.querymaster.QueryInProgress" %>
 <%@ page import="org.apache.tajo.master.rm.Worker" %>
 <%@ page import="org.apache.tajo.util.JSPUtil" %>
@@ -28,6 +27,8 @@
 <%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
 <%@ page import="java.text.SimpleDateFormat" %>
 <%@ page import="java.util.*" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
+<%@ page import="org.apache.tajo.master.querymaster.QueryInfo" %>
 
 <%
   TajoMaster master = (TajoMaster) StaticHttpServer.getInstance().getAttribute("tajo.info.server.object");
@@ -38,8 +39,28 @@
   runningQueries.addAll(master.getContext().getQueryJobManager().getRunningQueries());
           JSPUtil.sortQueryInProgress(runningQueries, true);
 
-  List<QueryInProgress> finishedQueries =
-          JSPUtil.sortQueryInProgress(master.getContext().getQueryJobManager().getFinishedQueries(), true);
+  int currentPage = 1;
+  if (request.getParameter("page") != null && !request.getParameter("page").isEmpty()) {
+    currentPage = Integer.parseInt(request.getParameter("page"));
+  }
+  int pageSize = HistoryReader.DEFAULT_PAGE_SIZE;
+  if (request.getParameter("pageSize") != null && !request.getParameter("pageSize").isEmpty()) {
+    try {
+      pageSize = Integer.parseInt(request.getParameter("pageSize"));
+    } catch (NumberFormatException e) {
+      pageSize = HistoryReader.DEFAULT_PAGE_SIZE;
+    }
+  }
+
+  String keyword = request.getParameter("keyword");
+  HistoryReader historyReader = master.getContext().getHistoryReader();
+  List<QueryInfo> allFinishedQueries = historyReader.getQueries(keyword);
+
+  int numOfFinishedQueries = allFinishedQueries.size();
+  int totalPage = numOfFinishedQueries % pageSize == 0 ?
+      numOfFinishedQueries / pageSize : numOfFinishedQueries / pageSize + 1;
+
+  List<QueryInfo> finishedQueries = JSPUtil.getPageNavigationList(allFinishedQueries, currentPage, pageSize);
 
   SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
@@ -56,16 +77,6 @@
       portMap.put(queryMaster.getConnectionInfo().getHost(), queryMaster.getConnectionInfo().getHttpInfoPort());
     }
   }
-
-  HAService haService = master.getContext().getHAService();
-  String activeLabel = "";
-  if (haService != null) {
-    if (haService.isActiveStatus()) {
-      activeLabel = "<font color='#1e90ff'>(active)</font>";
-    } else {
-      activeLabel = "<font color='#1e90ff'>(backup)</font>";
-    }
-  }
 %>
 
 <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
@@ -100,7 +111,7 @@
 <body>
 <%@ include file="header.jsp"%>
 <div class='contents'>
-  <h2>Tajo Master: <%=master.getMasterName()%> <%=activeLabel%></h2>
+  <h2>Tajo Master: <%=master.getMasterName()%> <%=JSPUtil.getMasterActiveLabel(master.getContext())%></h2>
   <hr/>
   <h3>Running Queries</h3>
 <%
@@ -114,7 +125,7 @@
       for(QueryInProgress eachQuery: runningQueries) {
         long time = System.currentTimeMillis() - eachQuery.getQueryInfo().getStartTime();
         String detailView = "http://" + eachQuery.getQueryInfo().getQueryMasterHost() + ":" + portMap.get(eachQuery.getQueryInfo().getQueryMasterHost()) +
-                "/querydetail.jsp?queryId=" + eachQuery.getQueryId();
+                "/querydetail.jsp?queryId=" + eachQuery.getQueryId() + "&startTime=" + eachQuery.getQueryInfo().getStartTime();
     %>
     <tr>
       <td><a href='<%=detailView%>'><%=eachQuery.getQueryId()%></a></td>
@@ -141,28 +152,37 @@
       out.write("No finished queries");
     } else {
   %>
+  <div align="right">
+    <form action='query.jsp' method='GET'>
+      Page Size: <input type="text" name="pageSize" value="<%=pageSize%>" size="5"/>
+      &nbsp;<input type="submit" value="Submit">
+    </form>
+  </div>
   <table width="100%" border="1" class='border_table'>
     <tr></tr><th>QueryId</th><th>Query Master</th><th>Started</th><th>Finished</th><th>Time</th><th>Status</th><th>sql</th></tr>
     <%
-      for(QueryInProgress eachQuery: finishedQueries) {
-        long runTime = eachQuery.getQueryInfo().getFinishTime() > 0 ?
-                eachQuery.getQueryInfo().getFinishTime() - eachQuery.getQueryInfo().getStartTime() : -1;
-        String detailView = "http://" + eachQuery.getQueryInfo().getQueryMasterHost() + ":" + portMap.get(eachQuery.getQueryInfo().getQueryMasterHost())  +
-                "/querydetail.jsp?queryId=" + eachQuery.getQueryId();
+      for(QueryInfo eachQuery: finishedQueries) {
+        long runTime = eachQuery.getFinishTime() > 0 ?
+                eachQuery.getFinishTime() - eachQuery.getStartTime() : -1;
+        String detailView = "querydetail.jsp?queryId=" + eachQuery.getQueryIdStr() + "&startTime=" + eachQuery.getStartTime();
     %>
     <tr>
-      <td><a href='<%=detailView%>'><%=eachQuery.getQueryId()%></a></td>
-      <td><%=eachQuery.getQueryInfo().getQueryMasterHost()%></td>
-      <td><%=df.format(eachQuery.getQueryInfo().getStartTime())%></td>
-      <td><%=eachQuery.getQueryInfo().getFinishTime() > 0 ? df.format(eachQuery.getQueryInfo().getFinishTime()) : "-"%></td>
+      <td><a href='<%=detailView%>'><%=eachQuery.getQueryIdStr()%></a></td>
+      <td><%=eachQuery.getQueryMasterHost()%></td>
+      <td><%=df.format(eachQuery.getStartTime())%></td>
+      <td><%=eachQuery.getFinishTime() > 0 ? df.format(eachQuery.getFinishTime()) : "-"%></td>
       <td><%=runTime == -1 ? "-" : StringUtils.formatTime(runTime) %></td>
-      <td><%=eachQuery.getQueryInfo().getQueryState()%></td>
-      <td><%=eachQuery.getQueryInfo().getSql()%></td>
+      <td><%=eachQuery.getQueryState()%></td>
+      <td><%=eachQuery.getSql()%></td>
     </tr>
     <%
       }
     %>
   </table>
+  <div align="center">
+    <%=JSPUtil.getPageNavigation(currentPage, totalPage, "query.jsp?pageSize=" + pageSize)%>
+  </div>
+  <p/>
 <%
   }
 %>

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/admin/querydetail.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/admin/querydetail.jsp b/tajo-core/src/main/resources/webapps/admin/querydetail.jsp
new file mode 100644
index 0000000..41b0e8f
--- /dev/null
+++ b/tajo-core/src/main/resources/webapps/admin/querydetail.jsp
@@ -0,0 +1,116 @@
+<%
+  /*
+  * Licensed to the Apache Software Foundation (ASF) under one
+  * or more contributor license agreements. See the NOTICE file
+  * distributed with this work for additional information
+  * regarding copyright ownership. The ASF licenses this file
+  * to you under the Apache License, Version 2.0 (the
+  * "License"); you may not use this file except in compliance
+  * with the License. You may obtain a copy of the License at
+  *
+  * http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+%>
+<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
+
+<%@ page import="org.apache.tajo.util.JSPUtil" %>
+<%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
+<%@ page import="org.apache.tajo.master.TajoMaster" %>
+<%@ page import="java.text.SimpleDateFormat" %>
+<%@ page import="java.util.List" %>
+<%@ page import="org.apache.tajo.util.history.QueryHistory" %>
+<%@ page import="org.apache.tajo.util.history.SubQueryHistory" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
+
+<%
+  TajoMaster master = (TajoMaster) StaticHttpServer.getInstance().getAttribute("tajo.info.server.object");
+  HistoryReader reader = master.getContext().getHistoryReader();
+
+  String queryId = request.getParameter("queryId");
+  String startTime = request.getParameter("startTime");
+  QueryHistory queryHistory = reader.getQueryHistory(queryId, Long.parseLong(startTime));
+
+  List<SubQueryHistory> subQueryHistories =
+      queryHistory != null ? JSPUtil.sortSubQueryHistory(queryHistory.getSubQueryHistories()) : null;
+
+  SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+%>
+
+<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
+<html>
+<head>
+  <link rel="stylesheet" type="text/css" href="/static/style.css"/>
+  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
+  <title>Query Detail Info</title>
+</head>
+<body>
+<%@ include file="header.jsp"%>
+<div class='contents'>
+  <h2>Tajo Master: <%=master.getMasterName()%> <%=JSPUtil.getMasterActiveLabel(master.getContext())%></h2>
+  <hr/>
+  <h3><%=queryId%></h3>
+<%
+if (queryHistory == null) {
+%>
+  <div>No Query history data.</div>
+<%
+} else {
+  if (subQueryHistories == null) {
+%>
+    <div>No SubQuery history data.</div>
+<%
+  } else {
+%>
+  <table width="100%" border="1" class="border_table">
+    <tr><th>ID</th><th>State</th><th>Started</th><th>Finished</th><th>Running time</th><th>Progress</th><th>Succeeded/Total</th><th>Failed/Killed</th></tr>
+<%
+    for(SubQueryHistory eachSubQuery: subQueryHistories) {
+        String detailLink = "querytasks.jsp?queryId=" + queryId + "&ebid=" + eachSubQuery.getExecutionBlockId() + "&startTime=" + startTime;
+%>
+  <tr>
+    <td><a href='<%=detailLink%>'><%=eachSubQuery.getExecutionBlockId()%></a></td>
+    <td><%=eachSubQuery.getState()%></td>
+    <td><%=df.format(eachSubQuery.getStartTime())%></td>
+    <td><%=eachSubQuery.getFinishTime() == 0 ? "-" : df.format(eachSubQuery.getFinishTime())%></td>
+    <td><%=JSPUtil.getElapsedTime(eachSubQuery.getStartTime(), eachSubQuery.getFinishTime())%></td>
+    <td align='center'><%=JSPUtil.percentFormat(eachSubQuery.getProgress())%>%</td>
+    <td align='center'><%=eachSubQuery.getSucceededObjectCount()%> / <%=eachSubQuery.getTotalScheduledObjectsCount()%></td>
+    <td align='center'><%=eachSubQuery.getFailedObjectCount()%> / <%=eachSubQuery.getKilledObjectCount()%></td>
+  </tr>
+  <%
+    }  //end of for
+  %>
+  </table>
+<%
+  }   //end of else [if (subQueryHistories == null)]
+%>
+  <p/>
+  <h3>Applied Session Variables</h3>
+  <table width="100%" border="1" class="border_table">
+<%
+  for(String[] sessionVariable: queryHistory.getSessionVariables()) {
+%>
+    <tr><td width="200"><%=sessionVariable[0]%></td><td><%=sessionVariable[1]%></td>
+<%
+  }
+%>
+  </table>
+  <hr/>
+  <h3>Logical Plan</h3>
+  <pre style="white-space:pre-wrap;"><%=queryHistory.getLogicalPlan()%></pre>
+  <hr/>
+  <h3>Distributed Query Plan</h3>
+  <pre style="white-space:pre-wrap;"><%=queryHistory.getDistributedPlan()%></pre>
+  <hr/>
+<%
+}   //end of else [if (query == null)]
+%>
+</div>
+</body>
+</html>
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/admin/querytasks.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/admin/querytasks.jsp b/tajo-core/src/main/resources/webapps/admin/querytasks.jsp
new file mode 100644
index 0000000..7a23157
--- /dev/null
+++ b/tajo-core/src/main/resources/webapps/admin/querytasks.jsp
@@ -0,0 +1,249 @@
+<%
+  /*
+  * Licensed to the Apache Software Foundation (ASF) under one
+  * or more contributor license agreements. See the NOTICE file
+  * distributed with this work for additional information
+  * regarding copyright ownership. The ASF licenses this file
+  * to you under the Apache License, Version 2.0 (the
+  * "License"); you may not use this file except in compliance
+  * with the License. You may obtain a copy of the License at
+  *
+  * http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+%>
+<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
+
+<%@ page import="org.apache.tajo.util.FileUtil" %>
+<%@ page import="org.apache.tajo.util.JSPUtil" %>
+<%@ page import="org.apache.tajo.util.TajoIdUtils" %>
+<%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
+<%@ page import="java.text.NumberFormat" %>
+<%@ page import="java.text.SimpleDateFormat" %>
+<%@ page import="org.apache.tajo.master.TajoMaster" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
+<%@ page import="org.apache.tajo.util.history.QueryHistory" %>
+<%@ page import="org.apache.tajo.util.history.SubQueryHistory" %>
+<%@ page import="org.apache.tajo.master.rm.Worker" %>
+<%@ page import="java.util.*" %>
+<%@ page import="org.apache.tajo.util.history.QueryUnitHistory" %>
+
+<%
+  TajoMaster master = (TajoMaster) StaticHttpServer.getInstance().getAttribute("tajo.info.server.object");
+  HistoryReader reader = master.getContext().getHistoryReader();
+
+  String queryId = request.getParameter("queryId");
+  String startTime = request.getParameter("startTime");
+  String ebId = request.getParameter("ebid");
+
+  QueryHistory queryHistory = reader.getQueryHistory(queryId);
+
+  List<SubQueryHistory> subQueryHistories =
+      queryHistory != null ? JSPUtil.sortSubQueryHistory(queryHistory.getSubQueryHistories()) : null;
+
+  SubQueryHistory subQuery = null;
+  if (subQueryHistories != null) {
+    for (SubQueryHistory eachSubQuery: subQueryHistories) {
+      if (eachSubQuery.getExecutionBlockId().equals(ebId)) {
+        subQuery = eachSubQuery;
+        break;
+      }
+    }
+  }
+
+  String sort = request.getParameter("sort");
+  if(sort == null) {
+    sort = "id";
+  }
+  String sortOrder = request.getParameter("sortOrder");
+  if(sortOrder == null) {
+    sortOrder = "asc";
+  }
+
+  String nextSortOrder = "asc";
+  if("asc".equals(sortOrder)) {
+    nextSortOrder = "desc";
+  }
+
+  String status = request.getParameter("status");
+  if(status == null || status.isEmpty() || "null".equals(status)) {
+    status = "ALL";
+  }
+
+  Collection<Worker> allWorkers = master.getContext().getResourceManager().getWorkers().values();
+
+  Map<String, Worker> workerMap = new HashMap<String, Worker>();
+  if(allWorkers != null) {
+    for(Worker eachWorker: allWorkers) {
+      workerMap.put(eachWorker.getConnectionInfo().getHostAndPeerRpcPort(), eachWorker);
+    }
+  }
+
+  SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+  long totalInputBytes = 0;
+  long totalReadBytes = 0;
+  long totalReadRows = 0;
+  long totalWriteBytes = 0;
+  long totalWriteRows = 0;
+
+  if (subQuery != null) {
+    totalInputBytes = subQuery.getTotalInputBytes();
+    totalReadBytes = subQuery.getTotalReadBytes();
+    totalReadRows = subQuery.getTotalReadRows();
+    totalWriteBytes = subQuery.getTotalWriteBytes();
+    totalWriteRows = subQuery.getTotalWriteRows();
+  }
+
+  List<QueryUnitHistory> allQueryUnits = reader.getQueryUnitHistory(queryId, ebId);
+  int numTasks = allQueryUnits.size();
+  int numShuffles = 0;
+  float totalProgress = 0.0f;
+
+  if (allQueryUnits != null) {
+    for(QueryUnitHistory eachQueryUnit: allQueryUnits) {
+      totalProgress += eachQueryUnit.getProgress();
+      numShuffles = eachQueryUnit.getNumShuffles();
+    }
+  }
+
+  int currentPage = 1;
+  if (request.getParameter("page") != null && !request.getParameter("page").isEmpty()) {
+    currentPage = Integer.parseInt(request.getParameter("page"));
+  }
+  int pageSize = HistoryReader.DEFAULT_TASK_PAGE_SIZE;
+  if (request.getParameter("pageSize") != null && !request.getParameter("pageSize").isEmpty()) {
+    try {
+      pageSize = Integer.parseInt(request.getParameter("pageSize"));
+    } catch (NumberFormatException e) {
+      pageSize = HistoryReader.DEFAULT_TASK_PAGE_SIZE;
+    }
+  }
+
+  String url = "querytasks.jsp?queryId=" + queryId + "&ebid=" + ebId + "&startTime=" + startTime +
+      "&page=" + currentPage + "&pageSize=" + pageSize +
+      "&status=" + status + "&sortOrder=" + nextSortOrder + "&sort=";
+
+  String pageUrl = "querytasks.jsp?queryId=" + queryId + "&ebid=" + ebId + "&startTime=" + startTime +
+      "&status=" + status + "&sortOrder=" + nextSortOrder + "&sort=";
+
+  NumberFormat nf = NumberFormat.getInstance(Locale.US);
+%>
+
+<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
+<html>
+<head>
+  <link rel="stylesheet" type="text/css" href="/static/style.css"/>
+  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
+  <title>Query Detail Info</title>
+</head>
+<body>
+<%@ include file="header.jsp"%>
+<div class='contents'>
+  <h2>Tajo Master: <%=master.getMasterName()%> <%=JSPUtil.getMasterActiveLabel(master.getContext())%></h2>
+  <hr/>
+  <h3><a href='querydetail.jsp?queryId=<%=queryId%>&startTime=<%=startTime%>'><%=ebId.toString()%></a></h3>
+  <hr/>
+  <p/>
+  <pre style="white-space:pre-wrap;"><%=subQuery.getPlan()%></pre>
+  <p/>
+  <table border="1" width="100%" class="border_table">
+    <tr><td align='right' width='180px'>Status:</td><td><%=subQuery.getState()%></td></tr>
+    <tr><td align='right'>Started:</td><td><%=df.format(subQuery.getStartTime())%> ~ <%=subQuery.getFinishTime() == 0 ? "-" : df.format(subQuery.getFinishTime())%></td></tr>
+    <tr><td align='right'># Tasks:</td><td><%=numTasks%> (Local Tasks: <%=subQuery.getHostLocalAssigned()%>, Rack Local Tasks: <%=subQuery.getRackLocalAssigned()%>)</td></tr>
+    <tr><td align='right'>Progress:</td><td><%=JSPUtil.percentFormat((float) (totalProgress / numTasks))%>%</td></tr>
+    <tr><td align='right'># Shuffles:</td><td><%=numShuffles%></td></tr>
+    <tr><td align='right'>Input Bytes:</td><td><%=FileUtil.humanReadableByteCount(totalInputBytes, false) + " (" + nf.format(totalInputBytes) + " B)"%></td></tr>
+    <tr><td align='right'>Actual Processed Bytes:</td><td><%=totalReadBytes == 0 ? "-" : FileUtil.humanReadableByteCount(totalReadBytes, false) + " (" + nf.format(totalReadBytes) + " B)"%></td></tr>
+    <tr><td align='right'>Input Rows:</td><td><%=nf.format(totalReadRows)%></td></tr>
+    <tr><td align='right'>Output Bytes:</td><td><%=FileUtil.humanReadableByteCount(totalWriteBytes, false) + " (" + nf.format(totalWriteBytes) + " B)"%></td></tr>
+    <tr><td align='right'>Output Rows:</td><td><%=nf.format(totalWriteRows)%></td></tr>
+  </table>
+  <hr/>
+
+  <form action='querytasks.jsp' method='GET'>
+  Status:
+    <select name="status" onchange="this.form.submit()">
+        <option value="ALL" <%="ALL".equals(status) ? "selected" : ""%>>ALL</option>
+        <option value="TA_ASSIGNED" <%="TA_ASSIGNED".equals(status) ? "selected" : ""%>>TA_ASSIGNED</option>
+        <option value="TA_PENDING" <%="TA_PENDING".equals(status) ? "selected" : ""%>>TA_PENDING</option>
+        <option value="TA_RUNNING" <%="TA_RUNNING".equals(status) ? "selected" : ""%>>TA_RUNNING</option>
+        <option value="TA_SUCCEEDED" <%="TA_SUCCEEDED".equals(status) ? "selected" : ""%>>TA_SUCCEEDED</option>
+        <option value="TA_FAILED" <%="TA_FAILED".equals(status) ? "selected" : ""%>>TA_FAILED</option>
+    </select>
+    &nbsp;&nbsp;
+    Page Size: <input type="text" name="pageSize" value="<%=pageSize%>" size="5"/>
+    &nbsp;&nbsp;
+    <input type="submit" value="Filter">
+    <input type="hidden" name="queryId" value="<%=queryId%>"/>
+    <input type="hidden" name="ebid" value="<%=ebId%>"/>
+    <input type="hidden" name="sort" value="<%=sort%>"/>
+    <input type="hidden" name="sortOrder" value="<%=sortOrder%>"/>
+    <input type="hidden" name="startTime" value="<%=startTime%>"/>
+  </form>
+<%
+  List<QueryUnitHistory> filteredQueryUnit = new ArrayList<QueryUnitHistory>();
+  for(QueryUnitHistory eachQueryUnit: allQueryUnits) {
+    if (!"ALL".equals(status)) {
+      if (!status.equals(eachQueryUnit.getState().toString())) {
+        continue;
+      }
+    }
+    filteredQueryUnit.add(eachQueryUnit);
+  }
+  JSPUtil.sortQueryUnitHistory(filteredQueryUnit, sort, sortOrder);
+  List<QueryUnitHistory> queryUnits = JSPUtil.getPageNavigationList(filteredQueryUnit, currentPage, pageSize);
+
+  int numOfQueryUnits = filteredQueryUnit.size();
+  int totalPage = numOfQueryUnits % pageSize == 0 ?
+      numOfQueryUnits / pageSize : numOfQueryUnits / pageSize + 1;
+%>
+  <div align="right"># Tasks: <%=numOfQueryUnits%> / # Pages: <%=totalPage%></div>
+  <table border="1" width="100%" class="border_table">
+    <tr><th>No</th><th><a href='<%=url%>id'>Id</a></th><th>Status</th><th>Progress</th><th><a href='<%=url%>startTime'>Started</a></th><th><a href='<%=url%>runTime'>Running Time</a></th><th><a href='<%=url%>host'>Host</a></th></tr>
+<%
+  int rowNo = (currentPage - 1) * pageSize + 1;
+  for (QueryUnitHistory eachQueryUnit: queryUnits) {
+    String queryUnitDetailUrl = "";
+    if (eachQueryUnit.getId() != null) {
+      queryUnitDetailUrl = "queryunit.jsp?queryId=" + queryId + "&ebid=" + ebId + "&startTime=" + startTime +
+          "&queryUnitAttemptId=" + eachQueryUnit.getId() + "&sort=" + sort + "&sortOrder=" + sortOrder;
+    }
+    String queryUnitHost = eachQueryUnit.getHostAndPort() == null ? "-" : eachQueryUnit.getHostAndPort();
+    if (eachQueryUnit.getHostAndPort() != null) {
+      Worker worker = workerMap.get(eachQueryUnit.getHostAndPort());
+      if (worker != null) {
+        String[] hostTokens = eachQueryUnit.getHostAndPort().split(":");
+        queryUnitHost = "<a href='http://" + hostTokens[0] + ":" + worker.getConnectionInfo().getHttpInfoPort() +
+            "/taskhistory.jsp?queryUnitAttemptId=" + eachQueryUnit.getId() + "&startTime=" + eachQueryUnit.getLaunchTime() +
+            "'>" + eachQueryUnit.getHostAndPort() + "</a>";
+      }
+    }
+
+%>
+    <tr>
+      <td><%=rowNo%></td>
+      <td><a href="<%=queryUnitDetailUrl%>"><%=eachQueryUnit.getId()%></a></td>
+      <td><%=eachQueryUnit.getState()%></td>
+      <td><%=JSPUtil.percentFormat(eachQueryUnit.getProgress())%>%</td>
+      <td><%=eachQueryUnit.getLaunchTime() == 0 ? "-" : df.format(eachQueryUnit.getLaunchTime())%></td>
+      <td align='right'><%=eachQueryUnit.getLaunchTime() == 0 ? "-" : eachQueryUnit.getRunningTime() + " ms"%></td>
+      <td><%=queryUnitHost%></td>
+    </tr>
+    <%
+        rowNo++;
+      }
+    %>
+  </table>
+  <div align="center">
+    <%=JSPUtil.getPageNavigation(currentPage, totalPage, pageUrl + "&pageSize=" + pageSize)%>
+  </div>
+  <p/>
+</div>
+</body>
+</html>
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/admin/queryunit.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/admin/queryunit.jsp b/tajo-core/src/main/resources/webapps/admin/queryunit.jsp
new file mode 100644
index 0000000..697469f
--- /dev/null
+++ b/tajo-core/src/main/resources/webapps/admin/queryunit.jsp
@@ -0,0 +1,134 @@
+<%
+    /*
+    * Licensed to the Apache Software Foundation (ASF) under one
+    * or more contributor license agreements. See the NOTICE file
+    * distributed with this work for additional information
+    * regarding copyright ownership. The ASF licenses this file
+    * to you under the Apache License, Version 2.0 (the
+    * "License"); you may not use this file except in compliance
+    * with the License. You may obtain a copy of the License at
+    *
+    * http://www.apache.org/licenses/LICENSE-2.0
+    *
+    * Unless required by applicable law or agreed to in writing, software
+    * distributed under the License is distributed on an "AS IS" BASIS,
+    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    * See the License for the specific language governing permissions and
+    * limitations under the License.
+    */
+%>
+<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
+
+<%@ page import="org.apache.tajo.util.JSPUtil" %>
+<%@ page import="org.apache.tajo.util.TajoIdUtils" %>
+<%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
+<%@ page import="java.text.SimpleDateFormat" %>
+<%@ page import="org.apache.tajo.master.TajoMaster" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
+<%@ page import="org.apache.tajo.util.history.QueryUnitHistory" %>
+<%@ page import="java.util.List" %>
+
+<%
+  TajoMaster master = (TajoMaster) StaticHttpServer.getInstance().getAttribute("tajo.info.server.object");
+  HistoryReader reader = master.getContext().getHistoryReader();
+
+  String queryId = request.getParameter("queryId");
+  String ebId = request.getParameter("ebid");
+
+  String status = request.getParameter("status");
+  if(status == null || status.isEmpty() || "null".equals(status)) {
+      status = "ALL";
+  }
+
+  String queryUnitAttemptId = request.getParameter("queryUnitAttemptId");
+
+  List<QueryUnitHistory> allQueryUnits = reader.getQueryUnitHistory(queryId, ebId);
+
+  QueryUnitHistory queryUnit = null;
+  for(QueryUnitHistory eachQueryUnit: allQueryUnits) {
+    if (eachQueryUnit.getId().equals(queryUnitAttemptId)) {
+      queryUnit = eachQueryUnit;
+      break;
+    }
+  }
+
+  SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+  String backUrl = request.getHeader("referer");
+%>
+<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
+<html>
+<head>
+  <link rel="stylesheet" type="text/css" href="/static/style.css"/>
+  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
+  <title>Query Unit Detail</title>
+</head>
+<body>
+<%
+  if (queryUnit == null) {
+%>
+    <div>No QueryUnit history.</div>
+    <div><a href="<%=backUrl%>">Back</a></div>
+<%
+    return;
+  }
+
+  String fragmentInfo = "";
+  String delim = "";
+
+  for (String eachFragment : queryUnit.getFragments()) {
+      fragmentInfo += delim + eachFragment;
+      delim = "<br/>";
+  }
+
+  String fetchInfo = "";
+  delim = "";
+  String previousKey = null;
+  for (String[] e : queryUnit.getFetchs()) {
+    if (previousKey == null || !previousKey.equals(e[0])) {
+      fetchInfo += delim + "<b>" + e[0] + "</b>";
+    }
+    delim = "<br/>";
+    fetchInfo += delim + e[1];
+
+    previousKey = e[0];
+  }
+
+  String dataLocationInfos = "";
+  delim = "";
+  for (String eachLocation: queryUnit.getDataLocations()) {
+    dataLocationInfos += delim + eachLocation.toString();
+    delim = "<br/>";
+  }
+
+  int numShuffles = queryUnit.getNumShuffles();
+  String shuffleKey = "-";
+  String shuffleFileName = "-";
+  if(numShuffles > 0) {
+    shuffleKey = queryUnit.getShuffleKey();
+    shuffleFileName = queryUnit.getShuffleFileName();
+  }
+%>
+
+
+<%@ include file="header.jsp"%>
+<div class='contents'>
+  <h2>Tajo Master: <%=master.getMasterName()%> <%=JSPUtil.getMasterActiveLabel(master.getContext())%></h2>
+  <hr/>
+  <h3><a href='<%=backUrl%>'><%=ebId%></a></h3>
+  <hr/>
+  <table border="1" width="100%" class="border_table">
+    <tr><td width="200" align="right">ID</td><td><%=queryUnit.getId()%></td></tr>
+    <tr><td align="right">Progress</td><td><%=JSPUtil.percentFormat(queryUnit.getProgress())%>%</td></tr>
+    <tr><td align="right">State</td><td><%=queryUnit.getState()%></td></tr>
+    <tr><td align="right">Launch Time</td><td><%=queryUnit.getLaunchTime() == 0 ? "-" : df.format(queryUnit.getLaunchTime())%></td></tr>
+    <tr><td align="right">Finish Time</td><td><%=queryUnit.getFinishTime() == 0 ? "-" : df.format(queryUnit.getFinishTime())%></td></tr>
+    <tr><td align="right">Running Time</td><td><%=queryUnit.getLaunchTime() == 0 ? "-" : queryUnit.getRunningTime() + " ms"%></td></tr>
+    <tr><td align="right">Host</td><td><%=queryUnit.getHostAndPort() == null ? "-" : queryUnit.getHostAndPort()%></td></tr>
+    <tr><td align="right">Shuffles</td><td># Shuffle Outputs: <%=numShuffles%>, Shuffle Key: <%=shuffleKey%>, Shuffle file: <%=shuffleFileName%></td></tr>
+    <tr><td align="right">Data Locations</td><td><%=dataLocationInfos%></td></tr>
+    <tr><td align="right">Fragment</td><td><%=fragmentInfo%></td></tr>
+    <tr><td align="right">Fetches</td><td><%=fetchInfo%></td></tr>
+  </table>
+</div>
+</body>
+</html>
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/worker/querydetail.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/worker/querydetail.jsp b/tajo-core/src/main/resources/webapps/worker/querydetail.jsp
index 3c40e99..ceb1c56 100644
--- a/tajo-core/src/main/resources/webapps/worker/querydetail.jsp
+++ b/tajo-core/src/main/resources/webapps/worker/querydetail.jsp
@@ -22,7 +22,6 @@
 <%@ page import="org.apache.tajo.QueryId" %>
 <%@ page import="org.apache.tajo.master.querymaster.Query" %>
 <%@ page import="org.apache.tajo.master.querymaster.QueryMasterTask" %>
-<%@ page import="org.apache.tajo.master.querymaster.SubQuery" %>
 <%@ page import="org.apache.tajo.util.JSPUtil" %>
 <%@ page import="org.apache.tajo.util.TajoIdUtils" %>
 <%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
@@ -31,6 +30,9 @@
 <%@ page import="java.util.List" %>
 <%@ page import="java.util.Map" %>
 <%@ page import="org.apache.tajo.SessionVars" %>
+<%@ page import="org.apache.tajo.util.history.QueryHistory" %>
+<%@ page import="org.apache.tajo.util.history.SubQueryHistory" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
 
 <%
   QueryId queryId = TajoIdUtils.parseQueryId(request.getParameter("queryId"));
@@ -39,16 +41,29 @@
   QueryMasterTask queryMasterTask = tajoWorker.getWorkerContext()
           .getQueryMasterManagerService().getQueryMaster().getQueryMasterTask(queryId, true);
 
-  if (queryMasterTask == null) {
-    out.write("<script type='text/javascript'>alert('no query'); history.back(0); </script>");
-    return;
+  boolean runningQuery = queryMasterTask != null;
+
+  QueryHistory queryHistory = null;
+
+  Query query = null;
+  if (queryMasterTask != null) {
+    query = queryMasterTask.getQuery();
+    if (query != null) {
+      queryHistory = query.getQueryHistory();
+    }
+  } else {
+    HistoryReader reader = tajoWorker.getWorkerContext().getHistoryReader();
+    queryHistory = reader.getQueryHistory(queryId.toString());
   }
-  Query query = queryMasterTask.getQuery();
-  List<SubQuery> subQueries = null;
-  if (query != null) {
-    subQueries = JSPUtil.sortSubQuery(query.getSubQueries());
+
+  if (!runningQuery && queryHistory == null) {
+    out.write("<script type='text/javascript'>alert('no query history'); history.back(0); </script>");
+    return;
   }
 
+  List<SubQueryHistory> subQueryHistories =
+      queryHistory != null ? JSPUtil.sortSubQueryHistory(queryHistory.getSubQueryHistories()) : null;
+
   SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 %>
 
@@ -65,24 +80,26 @@
   <h2>Tajo Worker: <a href='index.jsp'><%=tajoWorker.getWorkerContext().getWorkerName()%></a></h2>
   <hr/>
 <%
-if (query == null) {
-  String errorMessage = queryMasterTask.getErrorMessage();
+if (runningQuery && query == null) {
   out.write("Query Status: " + queryMasterTask.getState());
+  String errorMessage = queryMasterTask.getErrorMessage();
   if (errorMessage != null && !errorMessage.isEmpty()) {
     out.write("<p/>Message:<p/><pre>" + errorMessage + "</pre>");
   }
+} else if (subQueryHistories == null) {
+  out.write("<p/>Message:<p/><pre>No SubQueries</pre>");
 } else {
 %>
   <h3><%=queryId.toString()%> <a href='queryplan.jsp?queryId=<%=queryId%>'>[Query Plan]</a></h3>
   <table width="100%" border="1" class="border_table">
     <tr><th>ID</th><th>State</th><th>Started</th><th>Finished</th><th>Running time</th><th>Progress</th><th>Tasks</th></tr>
 <%
-for(SubQuery eachSubQuery: subQueries) {
+for(SubQueryHistory eachSubQuery: subQueryHistories) {
     eachSubQuery.getSucceededObjectCount();
-    String detailLink = "querytasks.jsp?queryId=" + queryId + "&ebid=" + eachSubQuery.getId();
+    String detailLink = "querytasks.jsp?queryId=" + queryId + "&ebid=" + eachSubQuery.getExecutionBlockId();
 %>
   <tr>
-    <td><a href='<%=detailLink%>'><%=eachSubQuery.getId()%></a></td>
+    <td><a href='<%=detailLink%>'><%=eachSubQuery.getExecutionBlockId()%></a></td>
     <td><%=eachSubQuery.getState()%></td>
     <td><%=df.format(eachSubQuery.getStartTime())%></td>
     <td><%=eachSubQuery.getFinishTime() == 0 ? "-" : df.format(eachSubQuery.getFinishTime())%></td>

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/worker/querytasks.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/worker/querytasks.jsp b/tajo-core/src/main/resources/webapps/worker/querytasks.jsp
index 23d3c46..f39f57c 100644
--- a/tajo-core/src/main/resources/webapps/worker/querytasks.jsp
+++ b/tajo-core/src/main/resources/webapps/worker/querytasks.jsp
@@ -26,17 +26,13 @@
 <%@ page import="org.apache.tajo.plan.util.PlannerUtil" %>
 <%@ page import="org.apache.tajo.ipc.TajoMasterProtocol" %>
 <%@ page import="org.apache.tajo.master.querymaster.*" %>
-<%@ page import="org.apache.tajo.util.FileUtil" %>
-<%@ page import="org.apache.tajo.util.JSPUtil" %>
-<%@ page import="org.apache.tajo.util.TajoIdUtils" %>
 <%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
 <%@ page import="org.apache.tajo.worker.TajoWorker" %>
 <%@ page import="java.text.NumberFormat" %>
 <%@ page import="java.text.SimpleDateFormat" %>
-<%@ page import="java.util.HashMap" %>
-<%@ page import="java.util.List" %>
-<%@ page import="java.util.Locale" %>
-<%@ page import="java.util.Map" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
+<%@ page import="org.apache.tajo.util.*" %>
+<%@ page import="java.util.*" %>
 
 <%
   String paramQueryId = request.getParameter("queryId");
@@ -101,22 +97,20 @@
 
   SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
-  String url = "querytasks.jsp?queryId=" + queryId + "&ebid=" + ebid + "&status=" + status + "&sortOrder=" + nextSortOrder + "&sort=";
-  QueryUnit[] queryUnits = subQuery.getQueryUnits();
-
+  QueryUnit[] allQueryUnits = subQuery.getQueryUnits();
 
   long totalInputBytes = 0;
   long totalReadBytes = 0;
   long totalReadRows = 0;
   long totalWriteBytes = 0;
   long totalWriteRows = 0;
-  int numTasks = queryUnits.length;
+  int numTasks = allQueryUnits.length;
 //  int numSucceededTasks = 0;
 //  int localReadTasks = subQuery.;
   int numShuffles = 0;
 
   float totalProgress = 0.0f;
-  for(QueryUnit eachQueryUnit: queryUnits) {
+  for(QueryUnit eachQueryUnit: allQueryUnits) {
     totalProgress += eachQueryUnit.getLastAttempt() != null ? eachQueryUnit.getLastAttempt().getProgress(): 0.0f;
     numShuffles = eachQueryUnit.getShuffleOutpuNum();
     if (eachQueryUnit.getLastAttempt() != null) {
@@ -134,7 +128,27 @@
     }
   }
 
-    NumberFormat nf = NumberFormat.getInstance(Locale.US);
+  int currentPage = 1;
+  if (request.getParameter("page") != null && !request.getParameter("page").isEmpty()) {
+    currentPage = Integer.parseInt(request.getParameter("page"));
+  }
+  int pageSize = HistoryReader.DEFAULT_TASK_PAGE_SIZE;
+  if (request.getParameter("pageSize") != null && !request.getParameter("pageSize").isEmpty()) {
+    try {
+      pageSize = Integer.parseInt(request.getParameter("pageSize"));
+    } catch (NumberFormatException e) {
+      pageSize = HistoryReader.DEFAULT_TASK_PAGE_SIZE;
+    }
+  }
+
+  String url = "querytasks.jsp?queryId=" + queryId + "&ebid=" + ebid +
+      "&page=" + currentPage + "&pageSize=" + pageSize +
+      "&status=" + status + "&sortOrder=" + nextSortOrder + "&sort=";
+
+  String pageUrl = "querytasks.jsp?queryId=" + paramQueryId + "&ebid=" + paramEbId +
+      "&status=" + status + "&sortOrder=" + nextSortOrder + "&sort=";
+
+  NumberFormat nf = NumberFormat.getInstance(Locale.US);
 %>
 
 <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
@@ -168,55 +182,71 @@
   </table>
   <hr/>
 
-
   <form action='querytasks.jsp' method='GET'>
   Status:
     <select name="status" onchange="this.form.submit()">
-        <option value="ALL" <%="ALL".equals(status) ? "selected" : ""%>>ALL</option>
-        <option value="SCHEDULED" <%="SCHEDULED".equals(status) ? "selected" : ""%>>SCHEDULED</option>
-        <option value="RUNNING" <%="RUNNING".equals(status) ? "selected" : ""%>>RUNNING</option>
-        <option value="SUCCEEDED" <%="SUCCEEDED".equals(status) ? "selected" : ""%>>SUCCEEDED</option>
+      <option value="ALL" <%="ALL".equals(status) ? "selected" : ""%>>ALL</option>
+      <option value="TA_ASSIGNED" <%="TA_ASSIGNED".equals(status) ? "selected" : ""%>>TA_ASSIGNED</option>
+      <option value="TA_PENDING" <%="TA_PENDING".equals(status) ? "selected" : ""%>>TA_PENDING</option>
+      <option value="TA_RUNNING" <%="TA_RUNNING".equals(status) ? "selected" : ""%>>TA_RUNNING</option>
+      <option value="TA_SUCCEEDED" <%="TA_SUCCEEDED".equals(status) ? "selected" : ""%>>TA_SUCCEEDED</option>
+      <option value="TA_FAILED" <%="TA_FAILED".equals(status) ? "selected" : ""%>>TA_FAILED</option>
     </select>
     &nbsp;&nbsp;
+    Page Size: <input type="text" name="pageSize" value="<%=pageSize%>" size="5"/>
+    &nbsp;&nbsp;
     <input type="submit" value="Filter">
     <input type="hidden" name="queryId" value="<%=paramQueryId%>"/>
     <input type="hidden" name="ebid" value="<%=paramEbId%>"/>
     <input type="hidden" name="sort" value="<%=sort%>"/>
     <input type="hidden" name="sortOrder" value="<%=sortOrder%>"/>
   </form>
+<%
+  List<QueryUnit> filteredQueryUnit = new ArrayList<QueryUnit>();
+  for(QueryUnit eachQueryUnit: allQueryUnits) {
+    if (!"ALL".equals(status)) {
+      if (!status.equals(eachQueryUnit.getLastAttemptStatus().toString())) {
+        continue;
+      }
+    }
+    filteredQueryUnit.add(eachQueryUnit);
+  }
+  JSPUtil.sortQueryUnit(filteredQueryUnit, sort, sortOrder);
+  List<QueryUnit> queryUnits = JSPUtil.getPageNavigationList(filteredQueryUnit, currentPage, pageSize);
+
+  int numOfQueryUnits = filteredQueryUnit.size();
+  int totalPage = numOfQueryUnits % pageSize == 0 ?
+      numOfQueryUnits / pageSize : numOfQueryUnits / pageSize + 1;
+
+  int rowNo = (currentPage - 1) * pageSize + 1;
+%>
+  <div align="right"># Tasks: <%=numOfQueryUnits%> / # Pages: <%=totalPage%></div>
   <table border="1" width="100%" class="border_table">
     <tr><th>No</th><th><a href='<%=url%>id'>Id</a></th><th>Status</th><th>Progress</th><th><a href='<%=url%>startTime'>Started</a></th><th><a href='<%=url%>runTime'>Running Time</a></th><th><a href='<%=url%>host'>Host</a></th></tr>
-    <%
-      JSPUtil.sortQueryUnit(queryUnits, sort, sortOrder);
-      int rowNo = 1;
-      for(QueryUnit eachQueryUnit: queryUnits) {
-          if(!"ALL".equals(status)) {
-            if(!status.equals(eachQueryUnit.getState().toString())) {
-              continue;
-            }
-          }
-          int queryUnitSeq = eachQueryUnit.getId().getId();
-          String queryUnitDetailUrl = "queryunit.jsp?queryId=" + paramQueryId + "&ebid=" + paramEbId +
-                  "&queryUnitSeq=" + queryUnitSeq + "&sort=" + sort + "&sortOrder=" + sortOrder;
-
-          String queryUnitHost = eachQueryUnit.getSucceededHost() == null ? "-" : eachQueryUnit.getSucceededHost();
-          if(eachQueryUnit.getSucceededHost() != null) {
-              TajoMasterProtocol.WorkerResourceProto worker =
-                      workerMap.get(eachQueryUnit.getLastAttempt().getWorkerConnectionInfo().getId());
-              if(worker != null) {
-                  QueryUnitAttempt lastAttempt = eachQueryUnit.getLastAttempt();
-                  if(lastAttempt != null) {
-                    QueryUnitAttemptId lastAttemptId = lastAttempt.getId();
-                    queryUnitHost = "<a href='http://" + eachQueryUnit.getSucceededHost() + ":" + worker.getConnectionInfo().getHttpInfoPort() + "/taskdetail.jsp?queryUnitAttemptId=" + lastAttemptId + "'>" + eachQueryUnit.getSucceededHost() + "</a>";
-                  }
-              }
-          }
+<%
+  for(QueryUnit eachQueryUnit: queryUnits) {
+    int queryUnitSeq = eachQueryUnit.getId().getId();
+    String queryUnitDetailUrl = "queryunit.jsp?queryId=" + paramQueryId + "&ebid=" + paramEbId +
+            "&page=" + currentPage + "&pageSize=" + pageSize +
+            "&queryUnitSeq=" + queryUnitSeq + "&sort=" + sort + "&sortOrder=" + sortOrder;
 
-    %>
+    String queryUnitHost = eachQueryUnit.getSucceededHost() == null ? "-" : eachQueryUnit.getSucceededHost();
+    if(eachQueryUnit.getSucceededHost() != null) {
+        TajoMasterProtocol.WorkerResourceProto worker =
+                workerMap.get(eachQueryUnit.getLastAttempt().getWorkerConnectionInfo().getId());
+        if(worker != null) {
+            QueryUnitAttempt lastAttempt = eachQueryUnit.getLastAttempt();
+            if(lastAttempt != null) {
+              QueryUnitAttemptId lastAttemptId = lastAttempt.getId();
+              queryUnitHost = "<a href='http://" + eachQueryUnit.getSucceededHost() + ":" + worker.getConnectionInfo().getHttpInfoPort() + "/taskdetail.jsp?queryUnitAttemptId=" + lastAttemptId + "'>" + eachQueryUnit.getSucceededHost() + "</a>";
+            }
+        }
+    }
+%>
     <tr>
       <td><%=rowNo%></td>
       <td><a href="<%=queryUnitDetailUrl%>"><%=eachQueryUnit.getId()%></a></td>
-      <td><%=eachQueryUnit.getState()%></td>
+      <td><%=eachQueryUnit.getLastAttemptStatus()%></td>
       <td><%=JSPUtil.percentFormat(eachQueryUnit.getLastAttempt().getProgress())%>%</td>
       <td><%=eachQueryUnit.getLaunchTime() == 0 ? "-" : df.format(eachQueryUnit.getLaunchTime())%></td>
       <td align='right'><%=eachQueryUnit.getLaunchTime() == 0 ? "-" : eachQueryUnit.getRunningTime() + " ms"%></td>
@@ -227,8 +257,10 @@
       }
     %>
   </table>
-  <%
-  %>
+  <div align="center">
+    <%=JSPUtil.getPageNavigation(currentPage, totalPage, pageUrl + "&pageSize=" + pageSize)%>
+  </div>
+  <p/>
 </div>
 </body>
 </html>
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/resources/webapps/worker/taskhistory.jsp
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/resources/webapps/worker/taskhistory.jsp b/tajo-core/src/main/resources/webapps/worker/taskhistory.jsp
new file mode 100644
index 0000000..b777d5f
--- /dev/null
+++ b/tajo-core/src/main/resources/webapps/worker/taskhistory.jsp
@@ -0,0 +1,123 @@
+<%
+    /*
+    * Licensed to the Apache Software Foundation (ASF) under one
+    * or more contributor license agreements. See the NOTICE file
+    * distributed with this work for additional information
+    * regarding copyright ownership. The ASF licenses this file
+    * to you under the Apache License, Version 2.0 (the
+    * "License"); you may not use this file except in compliance
+    * with the License. You may obtain a copy of the License at
+    *
+    * http://www.apache.org/licenses/LICENSE-2.0
+    *
+    * Unless required by applicable law or agreed to in writing, software
+    * distributed under the License is distributed on an "AS IS" BASIS,
+    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    * See the License for the specific language governing permissions and
+    * limitations under the License.
+    */
+%>
+<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
+
+<%@ page import="org.apache.tajo.ipc.TajoWorkerProtocol" %>
+<%@ page import="org.apache.tajo.util.JSPUtil" %>
+<%@ page import="org.apache.tajo.webapp.StaticHttpServer" %>
+<%@ page import="org.apache.tajo.worker.*" %>
+<%@ page import="java.text.SimpleDateFormat" %>
+<%@ page import="java.util.List" %>
+<%@ page import="org.apache.tajo.util.history.HistoryReader" %>
+
+<%
+  TajoWorker tajoWorker = (TajoWorker) StaticHttpServer.getInstance().getAttribute("tajo.info.server.object");
+  HistoryReader reader = new HistoryReader(tajoWorker.getWorkerContext().getWorkerName(), tajoWorker.getWorkerContext().getConf());
+
+  TaskHistory taskHistory = reader.getTaskHistory(request.getParameter("queryUnitAttemptId"),
+      Long.parseLong(request.getParameter("startTime")));
+
+  SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+  String referer = request.getHeader("referer");
+%>
+<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
+<html>
+<head>
+  <link rel="stylesheet" type="text/css" href="/static/style.css"/>
+  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
+  <title>tajo worker</title>
+</head>
+<body>
+<%@ include file="header.jsp"%>
+<%
+  if (taskHistory == null) {
+%>
+<div class='contents'>
+  <div>No Task history</div>
+  <div><a href="<%=referer%>">Back</a></div>
+</div>
+</body>
+</html>
+<%
+    return;
+  }   //end of if [taskHistory == null]
+  %>
+<div class='contents'>
+  <h2>Tajo Worker: <a href='index.jsp'><%=tajoWorker.getWorkerContext().getWorkerName()%></a></h2>
+  <hr/>
+  <h3>Task Detail: <%=request.getParameter("queryUnitAttemptId")%></h3>
+  <table border="1" width="100%" class="border_table">
+      <tr><td width="200" align="right">ID</td><td><%=request.getParameter("queryUnitAttemptId")%></td></tr>
+      <tr><td align="right">State</td><td><%=taskHistory.getState()%></td></tr>
+      <tr><td align="right">Start Time</td><td><%=taskHistory.getStartTime() == 0 ? "-" : df.format(taskHistory.getStartTime())%></td></tr>
+      <tr><td align="right">Finish Time</td><td><%=taskHistory.getFinishTime() == 0 ? "-" : df.format(taskHistory.getFinishTime())%></td></tr>
+      <tr><td align="right">Running Time</td><td><%=JSPUtil.getElapsedTime(taskHistory.getStartTime(), taskHistory.getFinishTime())%></td></tr>
+      <tr><td align="right">Progress</td><td><%=JSPUtil.percentFormat(taskHistory.getProgress())%>%</td></tr>
+      <tr><td align="right">Output Path</td><td><%=taskHistory.getOutputPath()%></td></tr>
+      <tr><td align="right">Working Path</td><td><%=taskHistory.getWorkingPath()%></td></tr>
+      <tr><td align="right">Input Statistics</td><td><%=JSPUtil.tableStatToString(taskHistory.getInputStats())%></td></tr>
+      <tr><td align="right">Output Statistics</td><td><%=JSPUtil.tableStatToString(taskHistory.getOutputStats())%></td></tr>
+  </table>
+  <hr/>
+<%
+  if (taskHistory.hasFetcherHistories()) {
+%>
+  <h3>Fetch Status &nbsp;
+      <span><%= taskHistory.getFinishedFetchCount() + "/" + taskHistory.getTotalFetchCount() %> (Finished/Total)</span>
+  </h3>
+<%
+    int index = 1;
+    int pageSize = 1000; //TODO pagination
+
+    List<TajoWorkerProtocol.FetcherHistoryProto> fetcherHistories = taskHistory.getFetcherHistories();
+    if (fetcherHistories.size() > 0) {
+%>
+  <table border="1" width="100%" class="border_table">
+  <tr><th>No</th><th>StartTime</th><th>FinishTime</th><th>RunTime</th><th>Status</th><th>File Length</th><th># Messages</th></tr>
+<%
+      for (TajoWorkerProtocol.FetcherHistoryProto eachFetcher : fetcherHistories) {
+%>
+    <tr>
+      <td><%=index%></td>
+      <td><%=df.format(eachFetcher.getStartTime())%></td>
+      <td><%=eachFetcher.getFinishTime() == 0 ? "-" : df.format(eachFetcher.getFinishTime())%></td>
+      <td><%=JSPUtil.getElapsedTime(eachFetcher.getStartTime(), eachFetcher.getFinishTime())%></td>
+      <td><%=eachFetcher.getState()%></td>
+      <td align="right"><%=eachFetcher.getFileLength()%></td>
+      <td align="right"><%=eachFetcher.getMessageReceivedCount()%></td>
+    </tr>
+<%
+        index++;
+        if (pageSize < index) {
+%>
+    <tr>
+      <td colspan="8">has more ...</td>
+    </tr>
+<%
+            break;
+        }
+      }   // end of for loop
+    }   // end of if [fetcherHistories.size() > 0]
+  }
+%>
+  </table>
+</div>
+</body>
+</html>
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java b/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
index faff5fb..30c94f3 100644
--- a/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
+++ b/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
@@ -28,16 +28,23 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.*;
+import org.apache.tajo.TajoProtos.QueryState;
 import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.catalog.FunctionDesc;
 import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.ClientProtos.QueryHistoryProto;
+import org.apache.tajo.ipc.ClientProtos.QueryInfoProto;
+import org.apache.tajo.ipc.ClientProtos.SubQueryHistoryProto;
 import org.apache.tajo.jdbc.TajoResultSet;
+import org.apache.tajo.master.querymaster.QueryInfo;
 import org.apache.tajo.storage.StorageConstants;
 import org.apache.tajo.storage.StorageUtil;
 import org.apache.tajo.util.CommonTestingUtil;
+import org.apache.tajo.util.history.QueryUnitHistory;
+import org.apache.tajo.util.history.SubQueryHistory;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -50,6 +57,7 @@ import java.sql.SQLException;
 import java.util.*;
 
 import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
 
 @Category(IntegrationTest.class)
 public class TestTajoClient {
@@ -722,4 +730,50 @@ public class TestTajoClient {
 
     assertEquals(expected, resultDatas);
   }
+
+  @Test
+  public void testGetQueryInfoAndHistory() throws Exception {
+    String sql = "select count(*) from lineitem";
+    ClientProtos.SubmitQueryResponse response = client.executeQuery(sql);
+
+    assertNotNull(response);
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    QueryInfoProto queryInfo = null;
+    long startTime = System.currentTimeMillis();
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+
+      if (queryInfo != null && queryInfo.getQueryState() == QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+
+      if (System.currentTimeMillis() - startTime > 30 * 1000) {
+        fail("Too long running query");
+      }
+    }
+    Thread.sleep(5 * 1000);
+
+    assertNotNull(queryInfo);
+    assertEquals(queryId.toString(), queryInfo.getQueryId());
+
+    QueryHistoryProto queryHistory = client.getQueryHistory(queryId);
+    assertNotNull(queryHistory);
+    assertEquals(queryId.toString(), queryHistory.getQueryId());
+    assertEquals(2, queryHistory.getSubQueryHistoriesCount());
+
+    List<SubQueryHistoryProto> queryUnitHistories =
+        new ArrayList<SubQueryHistoryProto>(queryHistory.getSubQueryHistoriesList());
+    Collections.sort(queryUnitHistories, new Comparator<SubQueryHistoryProto>() {
+      @Override
+      public int compare(SubQueryHistoryProto o1, SubQueryHistoryProto o2) {
+        return o1.getExecutionBlockId().compareTo(o2.getExecutionBlockId());
+      }
+    });
+    assertEquals(5, queryUnitHistories.get(0).getTotalReadRows());
+    assertEquals(1, queryUnitHistories.get(0).getTotalWriteRows());
+    assertEquals(1, queryUnitHistories.get(1).getTotalReadRows());
+    assertEquals(1, queryUnitHistories.get(1).getTotalWriteRows());
+  }
 }

http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/test/java/org/apache/tajo/util/TestJSPUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/util/TestJSPUtil.java b/tajo-core/src/test/java/org/apache/tajo/util/TestJSPUtil.java
index 74c2856..96aa5de 100644
--- a/tajo-core/src/test/java/org/apache/tajo/util/TestJSPUtil.java
+++ b/tajo-core/src/test/java/org/apache/tajo/util/TestJSPUtil.java
@@ -60,25 +60,54 @@ public class TestJSPUtil {
     Collections.shuffle(queryUnits);
 
     QueryUnit[] queryUnitArray = queryUnits.toArray(new QueryUnit[]{});
-    JSPUtil.sortQueryUnit(queryUnitArray, "id", "asc");
+    JSPUtil.sortQueryUnitArray(queryUnitArray, "id", "asc");
     for (int i = 0; i < 10; i++) {
       assertEquals(i, queryUnitArray[i].getId().getId());
     }
 
     queryUnitArray = queryUnits.toArray(new QueryUnit[]{});
-    JSPUtil.sortQueryUnit(queryUnitArray, "id", "desc");
+    JSPUtil.sortQueryUnitArray(queryUnitArray, "id", "desc");
     for (int i = 0; i < 10; i++) {
       assertEquals(9 - i, queryUnitArray[i].getId().getId());
     }
 
     queryUnitArray = queryUnits.toArray(new QueryUnit[]{});
-    JSPUtil.sortQueryUnit(queryUnitArray, "runTime", "asc");
+    JSPUtil.sortQueryUnitArray(queryUnitArray, "runTime", "asc");
     assertEquals(0, queryUnitArray[0].getId().getId());
     assertEquals(9, queryUnitArray[9].getId().getId());
 
     queryUnitArray = queryUnits.toArray(new QueryUnit[]{});
-    JSPUtil.sortQueryUnit(queryUnitArray, "runTime", "desc");
+    JSPUtil.sortQueryUnitArray(queryUnitArray, "runTime", "desc");
     assertEquals(8, queryUnitArray[0].getId().getId());
     assertEquals(9, queryUnitArray[9].getId().getId());
   }
+
+  @Test
+  public void testGetPageNavigationList() {
+    List<String> originList = new ArrayList<String>();
+
+    for (int i = 0; i < 35; i++) {
+      originList.add("Data" + (i + 1));
+    }
+
+    List<String> pageList = JSPUtil.getPageNavigationList(originList, 1, 10);
+    assertEquals(10, pageList.size());
+    assertEquals("Data1", pageList.get(0));
+    assertEquals("Data10", pageList.get(9));
+
+    pageList = JSPUtil.getPageNavigationList(originList, 2, 10);
+    assertEquals(10, pageList.size());
+    assertEquals("Data11", pageList.get(0));
+    assertEquals("Data20", pageList.get(9));
+
+    pageList = JSPUtil.getPageNavigationList(originList, 3, 10);
+    assertEquals(10, pageList.size());
+    assertEquals("Data21", pageList.get(0));
+    assertEquals("Data30", pageList.get(9));
+
+    pageList = JSPUtil.getPageNavigationList(originList, 4, 10);
+    assertEquals(5, pageList.size());
+    assertEquals("Data31", pageList.get(0));
+    assertEquals("Data35", pageList.get(4));
+  }
 }


Mime
View raw message