You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2014/11/11 15:24:46 UTC
[1/6] tajo git commit: TAJO-1026: Implement Query history persistency
manager.
Repository: tajo
Updated Branches:
refs/heads/index_support a5298fb21 -> 4a364cf08
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/test/java/org/apache/tajo/util/history/TestHistoryWriterReader.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/util/history/TestHistoryWriterReader.java b/tajo-core/src/test/java/org/apache/tajo/util/history/TestHistoryWriterReader.java
new file mode 100644
index 0000000..d5b737f
--- /dev/null
+++ b/tajo-core/src/test/java/org/apache/tajo/util/history/TestHistoryWriterReader.java
@@ -0,0 +1,251 @@
+/**
+ * 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.common.io.Files;
+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.TajoProtos.TaskAttemptState;
+import org.apache.tajo.catalog.proto.CatalogProtos.TableStatsProto;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.master.querymaster.QueryInfo;
+import org.apache.tajo.util.TajoIdUtils;
+import org.apache.tajo.worker.TaskHistory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import static org.junit.Assert.*;
+import static org.junit.Assert.assertTrue;
+
+public class TestHistoryWriterReader extends QueryTestCaseBase {
+ public static final String HISTORY_DIR = "/tmp/tajo-test-history";
+ TajoConf tajoConf;
+ @Before
+ public void setUp() throws Exception {
+ tajoConf = new TajoConf(testingCluster.getConfiguration());
+ tajoConf.setVar(ConfVars.HISTORY_QUERY_DIR, HISTORY_DIR);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ Path path = TajoConf.getQueryHistoryDir(tajoConf);
+ FileSystem fs = path.getFileSystem(tajoConf);
+ fs.delete(path, true);
+ }
+
+ @Test
+ public void testQueryInfoReadAndWrite() throws Exception {
+ HistoryWriter writer = new HistoryWriter("127.0.0.1:28090", true);
+ try {
+ writer.init(tajoConf);
+ writer.start();
+
+ long startTime = System.currentTimeMillis();
+ QueryInfo queryInfo1 = new QueryInfo(QueryIdFactory.newQueryId(startTime, 1));
+ queryInfo1.setStartTime(startTime);
+ queryInfo1.setProgress(1.0f);
+ queryInfo1.setQueryState(QueryState.QUERY_SUCCEEDED);
+
+ QueryInfo queryInfo2 = new QueryInfo(QueryIdFactory.newQueryId(startTime, 2));
+ queryInfo2.setStartTime(startTime);
+ queryInfo2.setProgress(0.5f);
+ queryInfo2.setQueryState(QueryState.QUERY_FAILED);
+
+ writer.appendHistory(queryInfo1);
+ writer.appendHistory(queryInfo2);
+
+ // HistoryWriter writes asynchronous.
+ Thread.sleep(5 * 1000);
+
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
+ Path path = new Path(tajoConf.getVar(ConfVars.HISTORY_QUERY_DIR));
+
+ FileSystem fs = path.getFileSystem(tajoConf);
+ Path parentPath = new Path(path, df.format(startTime) + "/query-list");
+ FileStatus[] histFiles = fs.listStatus(parentPath);
+ assertNotNull(histFiles);
+ assertEquals(1, histFiles.length);
+ assertTrue(histFiles[0].isFile());
+ assertTrue(histFiles[0].getPath().getName().endsWith(".hist"));
+
+ HistoryReader reader = new HistoryReader("127.0.0.1:28090", tajoConf);
+ List<QueryInfo> queryInfos = reader.getQueries(null);
+ assertNotNull(queryInfos);
+ assertEquals(2, queryInfos.size());
+
+ QueryInfo foundQueryInfo = queryInfos.get(0);
+ assertEquals(queryInfo2.getQueryId(), foundQueryInfo.getQueryId());
+ assertEquals(queryInfo2.getQueryState(), foundQueryInfo.getQueryState());
+ assertEquals(queryInfo2.getProgress(), foundQueryInfo.getProgress(), 0);
+
+ foundQueryInfo = queryInfos.get(1);
+ assertEquals(queryInfo1.getQueryId(), foundQueryInfo.getQueryId());
+ assertEquals(queryInfo1.getQueryState(), foundQueryInfo.getQueryState());
+ assertEquals(queryInfo1.getProgress(), foundQueryInfo.getProgress(), 0);
+ } finally {
+ writer.stop();
+ }
+ }
+
+ @Test
+ public void testQueryHistoryReadAndWrite() throws Exception {
+ HistoryWriter writer = new HistoryWriter("127.0.0.1:28090", true);
+ writer.init(tajoConf);
+ writer.start();
+
+ try {
+ long startTime = System.currentTimeMillis();
+
+ QueryHistory queryHistory = new QueryHistory();
+ QueryId queryId = QueryIdFactory.newQueryId(startTime, 1);
+ queryHistory.setQueryId(queryId.toString());
+ queryHistory.setLogicalPlan("LogicalPlan");
+ List<SubQueryHistory> subQueries = new ArrayList<SubQueryHistory>();
+ for (int i = 0; i < 3; i++) {
+ ExecutionBlockId ebId = QueryIdFactory.newExecutionBlockId(queryId, i);
+ SubQueryHistory subQueryHistory = new SubQueryHistory();
+ subQueryHistory.setExecutionBlockId(ebId.toString());
+ subQueryHistory.setStartTime(startTime + i);
+
+ List<QueryUnitHistory> queryUnitHistories = new ArrayList<QueryUnitHistory>();
+ for (int j = 0; j < 5; j++) {
+ QueryUnitHistory queryUnitHistory = new QueryUnitHistory();
+ queryUnitHistory.setId(QueryIdFactory.newQueryUnitAttemptId(QueryIdFactory.newQueryUnitId(ebId), 1).toString());
+ queryUnitHistories.add(queryUnitHistory);
+ }
+ subQueryHistory.setQueryUnits(queryUnitHistories);
+ subQueries.add(subQueryHistory);
+ }
+ queryHistory.setSubQueryHistories(subQueries);
+
+ writer.appendHistory(queryHistory);
+
+ // HistoryWriter writes asynchronous.
+ Thread.sleep(5 * 1000);
+
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
+ Path path = new Path(tajoConf.getVar(ConfVars.HISTORY_QUERY_DIR));
+
+ FileSystem fs = path.getFileSystem(tajoConf);
+
+ assertTrue(fs.exists(new Path(path,
+ df.format(startTime) + "/query-detail/" + queryId.toString() + "/query.hist")));
+ for (int i = 0; i < 3; i++) {
+ String ebId = QueryIdFactory.newExecutionBlockId(queryId, i).toString();
+ assertTrue(fs.exists(new Path(path,
+ df.format(startTime) + "/query-detail/" + queryId.toString() + "/" + ebId + ".hist")));
+ }
+
+ HistoryReader reader = new HistoryReader("127.0.0.1:28090", tajoConf);
+ QueryHistory foundQueryHistory = reader.getQueryHistory(queryId.toString());
+ assertNotNull(foundQueryHistory);
+ assertEquals(queryId.toString(), foundQueryHistory.getQueryId());
+ assertEquals(3, foundQueryHistory.getSubQueryHistories().size());
+
+ for (int i = 0; i < 3; i++) {
+ String ebId = QueryIdFactory.newExecutionBlockId(queryId, i).toString();
+ SubQueryHistory subQueryHistory = foundQueryHistory.getSubQueryHistories().get(i);
+ assertEquals(ebId, subQueryHistory.getExecutionBlockId());
+ assertEquals(startTime + i, subQueryHistory.getStartTime());
+
+ // QueryUnitHistory is stored in the other file.
+ assertNull(subQueryHistory.getQueryUnits());
+
+ List<QueryUnitHistory> queryUnits = reader.getQueryUnitHistory(queryId.toString(), ebId);
+ assertNotNull(queryUnits);
+ assertEquals(5, queryUnits.size());
+
+ for (int j = 0; j < 5; j++) {
+ QueryUnitHistory queryUnitHistory = queryUnits.get(j);
+ assertEquals(subQueries.get(i).getQueryUnits().get(j).getId(), queryUnitHistory.getId());
+ }
+ }
+ } finally {
+ writer.stop();
+ }
+ }
+
+ @Test
+ public void testTaskHistoryReadAndWrite() throws Exception {
+ TajoConf tajoConf = new TajoConf();
+ File historyParentDir = Files.createTempDir();
+ historyParentDir.deleteOnExit();
+ tajoConf.setVar(ConfVars.HISTORY_TASK_DIR, "file://" + historyParentDir.getCanonicalPath());
+
+ HistoryWriter writer = new HistoryWriter("127.0.0.1:28090", false);
+ writer.init(tajoConf);
+ writer.start();
+
+ try {
+ // Write TaskHistory
+ TableStatsProto tableStats = TableStatsProto.newBuilder()
+ .setNumRows(10)
+ .setNumBytes(100)
+ .build();
+ long startTime = System.currentTimeMillis() - 2000;
+ QueryUnitAttemptId id1 = TajoIdUtils.parseQueryUnitAttemptId("ta_1412326813565_0001_000001_000001_00");
+ TaskHistory taskHistory1 = new TaskHistory(
+ id1, TaskAttemptState.TA_SUCCEEDED, 1.0f, startTime, System.currentTimeMillis(), tableStats);
+ writer.appendHistory(taskHistory1);
+
+ QueryUnitAttemptId id2 = TajoIdUtils.parseQueryUnitAttemptId("ta_1412326813565_0001_000001_000002_00");
+ TaskHistory taskHistory2 = new TaskHistory(
+ id2, TaskAttemptState.TA_SUCCEEDED, 1.0f, startTime, System.currentTimeMillis() - 500, tableStats);
+ writer.appendHistory(taskHistory2);
+
+ // HistoryWriter writes asynchronous.
+ Thread.sleep(5 * 1000);
+
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
+ String startDate = df.format(new Date(startTime));
+ Path taskParentPath = new Path(tajoConf.getVar(ConfVars.HISTORY_TASK_DIR),
+ startDate.substring(0, 8) + "/tasks/127.0.0.1_28090");
+
+ FileSystem fs = taskParentPath.getFileSystem(tajoConf);
+ assertTrue(fs.exists(taskParentPath));
+
+ HistoryReader reader = new HistoryReader("127.0.0.1:28090", tajoConf);
+ TaskHistory foundTaskHistory = reader.getTaskHistory(id1.toString(), startTime);
+ assertNotNull(foundTaskHistory);
+ assertEquals(id1, foundTaskHistory.getQueryUnitAttemptId());
+ assertEquals(taskHistory1, foundTaskHistory);
+
+ foundTaskHistory = reader.getTaskHistory(id2.toString(), startTime);
+ assertNotNull(foundTaskHistory);
+ assertEquals(id2, foundTaskHistory.getQueryUnitAttemptId());
+ assertEquals(taskHistory2, foundTaskHistory);
+
+ foundTaskHistory = reader.getTaskHistory("ta_1412326813565_0001_000001_000003_00", startTime);
+ assertNull(foundTaskHistory);
+ } finally {
+ writer.stop();
+ }
+ }
+}
[4/6] tajo git commit: TAJO-1162: to_char() returns "-00" second.
Posted by ji...@apache.org.
TAJO-1162: to_char() returns "-00" second.
Closes #234
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/f45e74f6
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/f45e74f6
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/f45e74f6
Branch: refs/heads/index_support
Commit: f45e74f6efd1175c4eb97dbbcd38c702e29031a0
Parents: e01b00a
Author: HyoungJun Kim <ba...@babokim-mbp.server.gruter.com>
Authored: Tue Nov 11 10:36:29 2014 +0900
Committer: HyoungJun Kim <ba...@babokim-mbp.server.gruter.com>
Committed: Tue Nov 11 10:36:29 2014 +0900
----------------------------------------------------------------------
CHANGES | 2 ++
.../java/org/apache/tajo/util/datetime/DateTimeFormat.java | 2 +-
.../org/apache/tajo/engine/function/TestDateTimeFunctions.java | 6 ++++++
3 files changed, 9 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/f45e74f6/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 6c53920..950a108 100644
--- a/CHANGES
+++ b/CHANGES
@@ -36,6 +36,8 @@ Release 0.9.1 - unreleased
(Jongyoung Park via hyunsik)
BUG FIXES
+
+ TAJO-1162: to_char() returns "-00" second. (Hyoungjun Kim)
TAJO-1154: TajoCli doesn't pause while running the non-forwarded
query. (Hyoungjun Kim via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/f45e74f6/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
index c0b9245..c47c359 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeFormat.java
@@ -1679,7 +1679,7 @@ public class DateTimeFormat {
*/
private static String formatInteger(int value, int size) {
char[] targetArray, tempArray;
- final boolean isPositive = value>0;
+ final boolean isPositive = value>=0;
final int tempValue = isPositive?value:-value;
int targetArraySize;
http://git-wip-us.apache.org/repos/asf/tajo/blob/f45e74f6/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
index ab8772b..619bc5b 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
@@ -104,6 +104,12 @@ public class TestDateTimeFunctions extends ExprTestBase {
// (expectedTimestamp / 1000) means the translation from millis seconds to unix timestamp
String q = String.format("select to_char(to_timestamp(%d), 'yyyy-MM');", (expectedTimestamp / 1000));
testSimpleEval(q, new String[]{expectedDateTime.toString(dateFormatStr)});
+
+ q = "select to_char(to_timestamp('1997-12-30 11:40:00', 'YYYY-MM-DD HH24:MI:SS'), 'YYYY-MM-DD HH24:MI:SS')";
+ testSimpleEval(q, new String[]{"1997-12-30 11:40:00"});
+
+ q = "select to_char(to_timestamp('1997-12-30 00:00:00', 'YYYY-MM-DD HH24:MI:SS'), 'YYYY-MM-DD HH24:MI:SS')";
+ testSimpleEval(q, new String[]{"1997-12-30 00:00:00"});
}
@Test
[3/6] tajo git commit: TAJO-1026: Implement Query history persistency
manager.
Posted by ji...@apache.org.
TAJO-1026: Implement Query history persistency manager.
Closes #179
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/e01b00a7
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/e01b00a7
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/e01b00a7
Branch: refs/heads/index_support
Commit: e01b00a7b05b98cc407ced04dad1c2f2144d0f88
Parents: a3e5bdd
Author: HyoungJun Kim <ba...@babokim-mbp.server.gruter.com>
Authored: Tue Nov 11 10:30:22 2014 +0900
Committer: HyoungJun Kim <ba...@babokim-mbp.server.gruter.com>
Committed: Tue Nov 11 10:30:22 2014 +0900
----------------------------------------------------------------------
CHANGES | 4 +-
.../org/apache/tajo/client/QueryClient.java | 6 +
.../org/apache/tajo/client/QueryClientImpl.java | 52 +++
.../org/apache/tajo/client/TajoClientImpl.java | 13 +-
tajo-client/src/main/proto/ClientProtos.proto | 60 +++
.../main/proto/QueryMasterClientProtocol.proto | 1 +
.../main/proto/TajoMasterClientProtocol.proto | 1 +
.../java/org/apache/tajo/conf/TajoConf.java | 26 ++
.../java/org/apache/tajo/master/TajoMaster.java | 21 +
.../tajo/master/TajoMasterClientService.java | 33 ++
.../apache/tajo/master/querymaster/Query.java | 39 ++
.../master/querymaster/QueryInProgress.java | 3 +
.../tajo/master/querymaster/QueryInfo.java | 79 +++-
.../tajo/master/querymaster/QueryMaster.java | 89 ++--
.../tajo/master/querymaster/QueryUnit.java | 79 ++++
.../tajo/master/querymaster/SubQuery.java | 76 ++++
.../main/java/org/apache/tajo/util/JSPUtil.java | 162 ++++++-
.../org/apache/tajo/util/history/History.java | 27 ++
.../tajo/util/history/HistoryCleaner.java | 136 ++++++
.../apache/tajo/util/history/HistoryReader.java | 308 +++++++++++++
.../apache/tajo/util/history/HistoryWriter.java | 450 +++++++++++++++++++
.../apache/tajo/util/history/QueryHistory.java | 151 +++++++
.../tajo/util/history/QueryUnitHistory.java | 167 +++++++
.../tajo/util/history/SubQueryHistory.java | 270 +++++++++++
.../java/org/apache/tajo/worker/TajoWorker.java | 20 +
.../tajo/worker/TajoWorkerClientService.java | 34 ++
.../main/java/org/apache/tajo/worker/Task.java | 5 +-
.../org/apache/tajo/worker/TaskHistory.java | 8 +-
.../src/main/resources/webapps/admin/query.jsp | 72 +--
.../resources/webapps/admin/querydetail.jsp | 116 +++++
.../main/resources/webapps/admin/querytasks.jsp | 249 ++++++++++
.../main/resources/webapps/admin/queryunit.jsp | 134 ++++++
.../resources/webapps/worker/querydetail.jsp | 43 +-
.../resources/webapps/worker/querytasks.jsp | 126 ++++--
.../resources/webapps/worker/taskhistory.jsp | 123 +++++
.../org/apache/tajo/client/TestTajoClient.java | 54 +++
.../java/org/apache/tajo/util/TestJSPUtil.java | 37 +-
.../util/history/TestHistoryWriterReader.java | 251 +++++++++++
38 files changed, 3384 insertions(+), 141 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 665313b..6c53920 100644
--- a/CHANGES
+++ b/CHANGES
@@ -5,12 +5,14 @@ Release 0.9.1 - unreleased
NEW FEATURES
+ TAJO-1026: Implement Query history persistency manager.(Hyoungjun Kim)
+
TAJO-233: Support PostgreSQL CatalogStore. (Jihun Kang via hyunsik)
TAJO-235: Support Oracle CatalogStore. (Jihun Kang via hyunsik)
IMPROVEMENT
-
+
TAJO-1133: Add 'bin/tajo version' command. (Jihun Kang via hyunsik)
TAJO-1145: Add 'bin/tajo --help' command. (Jihun Kang via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java b/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
index 59ef52b..9b24663 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/QueryClient.java
@@ -22,6 +22,8 @@ import com.google.protobuf.ServiceException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.tajo.QueryId;
import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.ipc.ClientProtos.QueryHistoryProto;
+import org.apache.tajo.ipc.ClientProtos.QueryInfoProto;
import org.apache.tajo.jdbc.TajoMemoryResultSet;
import java.io.Closeable;
@@ -111,4 +113,8 @@ public interface QueryClient extends Closeable {
public List<ClientProtos.WorkerResourceInfo> getClusterInfo() throws ServiceException;
public QueryStatus killQuery(final QueryId queryId) throws ServiceException, IOException;
+
+ public QueryInfoProto getQueryInfo(final QueryId queryId) throws ServiceException;
+
+ public QueryHistoryProto getQueryHistory(final QueryId queryId) throws ServiceException;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
index f92c9bf..5b78959 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/QueryClientImpl.java
@@ -618,4 +618,56 @@ public class QueryClientImpl implements QueryClient {
}
return status;
}
+
+ public QueryInfoProto getQueryInfo(final QueryId queryId) throws ServiceException {
+ return new ServerCallable<QueryInfoProto>(connection.connPool, connection.getTajoMasterAddr(),
+ TajoMasterClientProtocol.class, false, true) {
+ public QueryInfoProto call(NettyClientBase client) throws ServiceException {
+ connection.checkSessionAndGet(client);
+
+ QueryIdRequest.Builder builder = QueryIdRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQueryId(queryId.getProto());
+
+ TajoMasterClientProtocolService.BlockingInterface tajoMasterService = client.getStub();
+ GetQueryInfoResponse res = tajoMasterService.getQueryInfo(null,builder.build());
+ if (res.getResultCode() == ResultCode.OK) {
+ return res.getQueryInfo();
+ } else {
+ abort();
+ throw new ServiceException(res.getErrorMessage());
+ }
+ }
+ }.withRetries();
+ }
+
+ public QueryHistoryProto getQueryHistory(final QueryId queryId) throws ServiceException {
+ final QueryInfoProto queryInfo = getQueryInfo(queryId);
+
+ if (queryInfo.getHostNameOfQM() == null || queryInfo.getQueryMasterClientPort() == 0) {
+ return null;
+ }
+ InetSocketAddress qmAddress = new InetSocketAddress(
+ queryInfo.getHostNameOfQM(), queryInfo.getQueryMasterClientPort());
+
+ return new ServerCallable<QueryHistoryProto>(connection.connPool, qmAddress,
+ QueryMasterClientProtocol.class, false, true) {
+ public QueryHistoryProto call(NettyClientBase client) throws ServiceException {
+ connection.checkSessionAndGet(client);
+
+ QueryIdRequest.Builder builder = QueryIdRequest.newBuilder();
+ builder.setSessionId(connection.sessionId);
+ builder.setQueryId(queryId.getProto());
+
+ QueryMasterClientProtocolService.BlockingInterface queryMasterService = client.getStub();
+ GetQueryHistoryResponse res = queryMasterService.getQueryHistory(null,builder.build());
+ if (res.getResultCode() == ResultCode.OK) {
+ return res.getQueryHistory();
+ } else {
+ abort();
+ throw new ServiceException(res.getErrorMessage());
+ }
+ }
+ }.withRetries();
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
index 75de492..1d637ed 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/TajoClientImpl.java
@@ -32,10 +32,7 @@ import org.apache.tajo.catalog.partition.PartitionMethodDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
-import org.apache.tajo.ipc.ClientProtos.BriefQueryInfo;
-import org.apache.tajo.ipc.ClientProtos.GetQueryResultResponse;
-import org.apache.tajo.ipc.ClientProtos.SubmitQueryResponse;
-import org.apache.tajo.ipc.ClientProtos.WorkerResourceInfo;
+import org.apache.tajo.ipc.ClientProtos.*;
import org.apache.tajo.jdbc.TajoMemoryResultSet;
import org.apache.tajo.jdbc.TajoResultSet;
import org.apache.tajo.util.NetUtils;
@@ -158,6 +155,14 @@ public class TajoClientImpl extends SessionConnection implements TajoClient, Que
return queryClient.getClusterInfo();
}
+ public QueryInfoProto getQueryInfo(final QueryId queryId) throws ServiceException {
+ return queryClient.getQueryInfo(queryId);
+ }
+
+ public QueryHistoryProto getQueryHistory(final QueryId queryId) throws ServiceException {
+ return queryClient.getQueryHistory(queryId);
+ }
+
/*------------------------------------------------------------------------*/
// CatalogClient wrappers
/*------------------------------------------------------------------------*/
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-client/src/main/proto/ClientProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/proto/ClientProtos.proto b/tajo-client/src/main/proto/ClientProtos.proto
index 23ae6dd..4118458 100644
--- a/tajo-client/src/main/proto/ClientProtos.proto
+++ b/tajo-client/src/main/proto/ClientProtos.proto
@@ -234,3 +234,63 @@ message FunctionResponse {
repeated FunctionDescProto functions = 2;
optional string errorMessage = 3;
}
+
+message QueryInfoProto {
+ required string queryId = 1;
+ optional string sql = 2;
+ optional QueryState queryState = 3;
+ optional float progress = 4;
+ optional int64 startTime = 5;
+ optional int64 finishTime = 6;
+ optional string lastMessage = 7;
+ optional string hostNameOfQM = 8;
+ optional int32 queryMasterPort = 9;
+ optional int32 queryMasterClientPort = 10;
+ optional int32 queryMasterInfoPort = 11;
+}
+
+message SubQueryHistoryProto {
+ required string executionBlockId =1;
+ required string state = 2;
+ optional int64 startTime = 3;
+ optional int64 finishTime = 4;
+ optional int32 succeededObjectCount = 5;
+ optional int32 failedObjectCount = 6;
+ optional int32 killedObjectCount = 7;
+ optional int32 totalScheduledObjectsCount = 8;
+
+ optional int64 totalInputBytes = 9;
+ optional int64 totalReadBytes = 10;
+ optional int64 totalReadRows = 11;
+ optional int64 totalWriteBytes = 12;
+ optional int64 totalWriteRows = 13;
+ optional int32 numShuffles = 14;
+ optional float progress =15;
+
+ optional string plan = 16;
+ optional int32 hostLocalAssigned = 17;
+ optional int32 rackLocalAssigned = 18;
+}
+
+message QueryHistoryProto {
+ required string queryId = 1;
+ optional string queryMaster = 2;
+ optional int32 httpPort = 3;
+ optional string logicalPlan = 4;
+ optional string distributedPlan = 5;
+ repeated KeyValueProto sessionVariables = 6;
+ repeated SubQueryHistoryProto subQueryHistories = 7;
+}
+
+message GetQueryHistoryResponse {
+ required ResultCode resultCode = 1;
+ optional QueryHistoryProto queryHistory = 2;
+ optional string errorMessage = 3;
+}
+
+message GetQueryInfoResponse {
+ required ResultCode resultCode = 1;
+ optional QueryInfoProto queryInfo = 2;
+ optional string errorMessage = 3;
+}
+
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-client/src/main/proto/QueryMasterClientProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/proto/QueryMasterClientProtocol.proto b/tajo-client/src/main/proto/QueryMasterClientProtocol.proto
index 9d96505..3d8d70b 100644
--- a/tajo-client/src/main/proto/QueryMasterClientProtocol.proto
+++ b/tajo-client/src/main/proto/QueryMasterClientProtocol.proto
@@ -32,4 +32,5 @@ service QueryMasterClientProtocolService {
rpc getQueryResult(GetQueryResultRequest) returns (GetQueryResultResponse);
rpc getQueryStatus(GetQueryStatusRequest) returns (GetQueryStatusResponse);
rpc closeQuery(QueryIdProto) returns (BoolProto);
+ rpc getQueryHistory(QueryIdRequest) returns (GetQueryHistoryResponse);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/proto/TajoMasterClientProtocol.proto b/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
index 1afc069..bc59617 100644
--- a/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
+++ b/tajo-client/src/main/proto/TajoMasterClientProtocol.proto
@@ -51,6 +51,7 @@ service TajoMasterClientProtocolService {
rpc killQuery(QueryIdRequest) returns (BoolProto);
rpc getClusterInfo(GetClusterInfoRequest) returns (GetClusterInfoResponse);
rpc closeNonForwardQuery(QueryIdRequest) returns (BoolProto);
+ rpc getQueryInfo(QueryIdRequest) returns (GetQueryInfoResponse);
// Database Management APIs
rpc createDatabase(SessionedStringProto) returns (BoolProto);
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 786aed0..3966410 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -296,6 +296,11 @@ public class TajoConf extends Configuration {
// Metrics ----------------------------------------------------------------
METRICS_PROPERTY_FILENAME("tajo.metrics.property.file", "tajo-metrics.properties"),
+ // Query History ---------------------------------------------------------
+ HISTORY_QUERY_DIR("tajo.history.query.dir", STAGING_ROOT_DIR.defaultVal + "/history"),
+ HISTORY_TASK_DIR("tajo.history.task.dir", "file:///tmp/tajo-${user.name}/history"),
+ HISTORY_EXPIRY_TIME_DAY("tajo.history.expiry-time-day", 7),
+
// Misc -------------------------------------------------------------------
// Geo IP
@@ -678,6 +683,27 @@ public class TajoConf extends Configuration {
return new Path(stagingDirString);
}
+ public static Path getQueryHistoryDir(TajoConf conf) throws IOException {
+ String historyDirString = conf.getVar(ConfVars.HISTORY_QUERY_DIR);
+ if (!hasScheme(historyDirString)) {
+ Path stagingPath = getStagingDir(conf);
+ FileSystem fs = stagingPath.getFileSystem(conf);
+ Path path = new Path(fs.getUri().toString(), historyDirString);
+ conf.setVar(ConfVars.HISTORY_QUERY_DIR, path.toString());
+ return path;
+ }
+ return new Path(historyDirString);
+ }
+
+ public static Path getTaskHistoryDir(TajoConf conf) throws IOException {
+ String historyDirString = conf.getVar(ConfVars.HISTORY_TASK_DIR);
+ if (!hasScheme(historyDirString)) {
+ //Local dir
+ historyDirString = "file://" + historyDirString;
+ }
+ return new Path(historyDirString);
+ }
+
public static Path getSystemConfPath(TajoConf conf) {
String systemConfPathStr = conf.getVar(ConfVars.SYSTEM_CONF_PATH);
if (systemConfPathStr == null || systemConfPathStr.equals("")) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
index 25e1be5..17658ac 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -52,6 +52,8 @@ import org.apache.tajo.rpc.RpcChannelFactory;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.util.*;
import org.apache.tajo.util.metrics.TajoSystemMetrics;
+import org.apache.tajo.util.history.HistoryReader;
+import org.apache.tajo.util.history.HistoryWriter;
import org.apache.tajo.webapp.QueryExecutorServlet;
import org.apache.tajo.webapp.StaticHttpServer;
@@ -125,6 +127,10 @@ public class TajoMaster extends CompositeService {
private JvmPauseMonitor pauseMonitor;
+ private HistoryWriter historyWriter;
+
+ private HistoryReader historyReader;
+
public TajoMaster() throws Exception {
super(TajoMaster.class.getName());
}
@@ -309,6 +315,13 @@ public class TajoMaster extends CompositeService {
} catch (IOException e) {
LOG.error(e.getMessage(), e);
}
+
+ historyWriter = new HistoryWriter(getMasterName(), true);
+ historyWriter.init(getConfig());
+ addIfService(historyWriter);
+ historyWriter.start();
+
+ historyReader = new HistoryReader(getMasterName(), context.getConf());
}
private void writeSystemConf() throws IOException {
@@ -452,6 +465,14 @@ public class TajoMaster extends CompositeService {
public HAService getHAService() {
return haService;
}
+
+ public HistoryWriter getHistoryWriter() {
+ return historyWriter;
+ }
+
+ public HistoryReader getHistoryReader() {
+ return historyReader;
+ }
}
String getThreadTaskName(long id, String name) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
index a4688d9..2c81cd0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
@@ -57,6 +57,8 @@ import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.StringProto;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.util.NetUtils;
import org.apache.tajo.util.ProtoUtil;
+import org.apache.tajo.util.StringUtils;
+import org.apache.tajo.util.history.QueryHistory;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -537,6 +539,37 @@ public class TajoMasterClientService extends AbstractService {
}
}
+ @Override
+ public GetQueryInfoResponse getQueryInfo(RpcController controller, QueryIdRequest request) throws ServiceException {
+ GetQueryInfoResponse.Builder builder = GetQueryInfoResponse.newBuilder();
+
+ try {
+ context.getSessionManager().touch(request.getSessionId().getId());
+ QueryId queryId = new QueryId(request.getQueryId());
+
+ QueryJobManager queryJobManager = context.getQueryJobManager();
+ QueryInProgress queryInProgress = queryJobManager.getQueryInProgress(queryId);
+
+ QueryInfo queryInfo = null;
+ if (queryInProgress == null) {
+ queryInfo = context.getHistoryReader().getQueryInfo(queryId.toString());
+ } else {
+ queryInfo = queryInProgress.getQueryInfo();
+ }
+
+ if (queryInfo != null) {
+ builder.setQueryInfo(queryInfo.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();
+ }
+
/**
* It is invoked by TajoContainerProxy.
*/
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
index a16d36a..7114d39 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -52,6 +52,8 @@ import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.StorageConstants;
import org.apache.tajo.storage.StorageUtil;
import org.apache.tajo.util.TUtil;
+import org.apache.tajo.util.history.QueryHistory;
+import org.apache.tajo.util.history.SubQueryHistory;
import java.io.IOException;
import java.text.NumberFormat;
@@ -287,6 +289,42 @@ public class Query implements EventHandler<QueryEvent> {
finishTime = clock.getTime();
}
+ public QueryHistory getQueryHistory() {
+ QueryHistory queryHistory = makeQueryHistory();
+ queryHistory.setSubQueryHistories(makeSubQueryHistories());
+ return queryHistory;
+ }
+
+ private List<SubQueryHistory> makeSubQueryHistories() {
+ List<SubQueryHistory> subQueryHistories = new ArrayList<SubQueryHistory>();
+ for(SubQuery eachSubQuery: getSubQueries()) {
+ subQueryHistories.add(eachSubQuery.getSubQueryHistory());
+ }
+
+ return subQueryHistories;
+ }
+
+ private QueryHistory makeQueryHistory() {
+ QueryHistory queryHistory = new QueryHistory();
+
+ queryHistory.setQueryId(getId().toString());
+ queryHistory.setQueryMaster(context.getQueryMasterContext().getWorkerContext().getWorkerName());
+ queryHistory.setHttpPort(context.getQueryMasterContext().getWorkerContext().getConnectionInfo().getHttpInfoPort());
+ queryHistory.setLogicalPlan(plan.toString());
+ queryHistory.setLogicalPlan(plan.getLogicalPlan().toString());
+ queryHistory.setDistributedPlan(plan.toString());
+
+ List<String[]> sessionVariables = new ArrayList<String[]>();
+ for(Map.Entry<String,String> entry: plan.getContext().getAllKeyValus().entrySet()) {
+ if (SessionVars.exists(entry.getKey()) && SessionVars.isPublic(SessionVars.get(entry.getKey()))) {
+ sessionVariables.add(new String[]{entry.getKey(), entry.getValue()});
+ }
+ }
+ queryHistory.setSessionVariables(sessionVariables);
+
+ return queryHistory;
+ }
+
public List<String> getDiagnostics() {
readLock.lock();
try {
@@ -385,6 +423,7 @@ public class Query implements EventHandler<QueryEvent> {
}
query.eventHandler.handle(new QueryMasterQueryCompletedEvent(query.getId()));
query.setFinishTime();
+
return finalState;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
index 536778a..d949ca4 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
@@ -146,6 +146,8 @@ public class QueryInProgress extends CompositeService {
if(queryMasterRpc != null) {
RpcConnectionPool.getPool((TajoConf)getConfig()).closeConnection(queryMasterRpc);
}
+
+ masterContext.getHistoryWriter().appendHistory(queryInfo);
super.stop();
}
@@ -175,6 +177,7 @@ public class QueryInProgress extends CompositeService {
queryInfo.setQueryMaster(resource.getConnectionInfo().getHost());
queryInfo.setQueryMasterPort(resource.getConnectionInfo().getQueryMasterPort());
queryInfo.setQueryMasterclientPort(resource.getConnectionInfo().getClientPort());
+ queryInfo.setQueryMasterInfoPort(resource.getConnectionInfo().getHttpInfoPort());
getEventHandler().handle(new QueryJobEvent(QueryJobEvent.Type.QUERY_MASTER_START, queryInfo));
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInfo.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInfo.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInfo.java
index 955c5b3..00b95ac 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInfo.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryInfo.java
@@ -19,21 +19,41 @@
package org.apache.tajo.master.querymaster;
+import com.google.gson.annotations.Expose;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
+import org.apache.tajo.engine.json.CoreGsonHelper;
+import org.apache.tajo.ipc.ClientProtos.QueryInfoProto;
+import org.apache.tajo.json.GsonObject;
+import org.apache.tajo.util.TajoIdUtils;
+import org.apache.tajo.util.history.History;
-public class QueryInfo {
+public class QueryInfo implements GsonObject, History {
private QueryId queryId;
+ @Expose
private String sql;
- private String jsonExpr;
+ @Expose
private TajoProtos.QueryState queryState;
+ @Expose
private float progress;
+ @Expose
private long startTime;
+ @Expose
private long finishTime;
+ @Expose
private String lastMessage;
+ @Expose
private String hostNameOfQM;
+ @Expose
private int queryMasterPort;
+ @Expose
private int queryMasterClientPort;
+ @Expose
+ private int queryMasterInfoPort;
+ @Expose
+ private String queryIdStr;
+
+ private String jsonExpr;
public QueryInfo(QueryId queryId) {
this(queryId, null, null);
@@ -41,6 +61,7 @@ public class QueryInfo {
public QueryInfo(QueryId queryId, String sql, String jsonExpr) {
this.queryId = queryId;
+ this.queryIdStr = queryId.toString();
this.sql = sql;
this.jsonExpr = jsonExpr;
this.queryState = TajoProtos.QueryState.QUERY_MASTER_INIT;
@@ -60,7 +81,14 @@ public class QueryInfo {
public void setQueryMaster(String hostName) {
this.hostNameOfQM = hostName;
+ }
+
+ public int getQueryMasterInfoPort() {
+ return queryMasterInfoPort;
+ }
+ public void setQueryMasterInfoPort(int queryMasterInfoPort) {
+ this.queryMasterInfoPort = queryMasterInfoPort;
}
public void setQueryMasterPort(int port) {
@@ -128,4 +156,51 @@ public class QueryInfo {
public String getJsonExpr() {
return jsonExpr;
}
+
+ @Override
+ public String toJson() {
+ return CoreGsonHelper.toJson(this, QueryInfo.class);
+ }
+
+ @Override
+ public HistoryType getHistoryType() {
+ return HistoryType.QUERY_SUMMARY;
+ }
+
+ public static QueryInfo fromJson(String json) {
+ QueryInfo queryInfo = CoreGsonHelper.fromJson(json, QueryInfo.class);
+ queryInfo.queryId = TajoIdUtils.parseQueryId(queryInfo.queryIdStr);
+ return queryInfo;
+ }
+
+ public String getQueryIdStr() {
+ return queryIdStr;
+ }
+
+ public QueryInfoProto getProto() {
+ QueryInfoProto.Builder builder = QueryInfoProto.newBuilder();
+
+ builder.setQueryId(queryId.toString())
+ .setQueryState(queryState)
+ .setProgress(progress)
+ .setStartTime(startTime)
+ .setFinishTime(finishTime)
+ .setQueryMasterPort(queryMasterPort)
+ .setQueryMasterClientPort(queryMasterClientPort)
+ .setQueryMasterInfoPort(queryMasterInfoPort);
+
+ if (sql != null) {
+ builder.setSql(sql);
+ }
+
+ if (lastMessage != null) {
+ builder.setLastMessage(lastMessage);
+ }
+
+ if (hostNameOfQM != null) {
+ builder.setHostNameOfQM(hostNameOfQM);
+ }
+
+ return builder.build();
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
index 7c3d799..b3b4dbb 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
@@ -44,6 +44,7 @@ import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.util.HAServiceUtil;
import org.apache.tajo.util.NetUtils;
+import org.apache.tajo.util.history.QueryHistory;
import org.apache.tajo.worker.TajoWorker;
import java.util.ArrayList;
@@ -393,58 +394,64 @@ public class QueryMaster extends CompositeService implements EventHandler {
}
public void stopQuery(QueryId queryId) {
- QueryMasterTask queryMasterTask;
- queryMasterTask = queryMasterTasks.remove(queryId);
- if(queryMasterTask == null) return;
+ QueryMasterTask queryMasterTask = queryMasterTasks.remove(queryId);
+ if(queryMasterTask == null) {
+ LOG.warn("No query info:" + queryId);
+ return;
+ }
finishedQueryMasterTasks.put(queryId, queryMasterTask);
- if(queryMasterTask != null) {
- TajoHeartbeat queryHeartbeat = buildTajoHeartBeat(queryMasterTask);
- CallFuture<TajoHeartbeatResponse> future = new CallFuture<TajoHeartbeatResponse>();
+ TajoHeartbeat queryHeartbeat = buildTajoHeartBeat(queryMasterTask);
+ CallFuture<TajoHeartbeatResponse> future = new CallFuture<TajoHeartbeatResponse>();
- NettyClientBase tmClient = null;
- try {
- // In TajoMaster HA mode, if backup master be active status,
- // worker may fail to connect existing active master. Thus,
- // if worker can't connect the master, worker should try to connect another master and
- // update master address in worker context.
- if (systemConf.getBoolVar(TajoConf.ConfVars.TAJO_MASTER_HA_ENABLE)) {
- try {
- tmClient = connPool.getConnection(queryMasterContext.getWorkerContext().getTajoMasterAddress(),
- TajoMasterProtocol.class, true);
- } catch (Exception e) {
- queryMasterContext.getWorkerContext().setWorkerResourceTrackerAddr(HAServiceUtil.getResourceTrackerAddress(systemConf));
- queryMasterContext.getWorkerContext().setTajoMasterAddress(HAServiceUtil.getMasterUmbilicalAddress(systemConf));
- tmClient = connPool.getConnection(queryMasterContext.getWorkerContext().getTajoMasterAddress(),
- TajoMasterProtocol.class, true);
- }
- } else {
+ NettyClientBase tmClient = null;
+ try {
+ // In TajoMaster HA mode, if backup master be active status,
+ // worker may fail to connect existing active master. Thus,
+ // if worker can't connect the master, worker should try to connect another master and
+ // update master address in worker context.
+ if (systemConf.getBoolVar(TajoConf.ConfVars.TAJO_MASTER_HA_ENABLE)) {
+ try {
+ tmClient = connPool.getConnection(queryMasterContext.getWorkerContext().getTajoMasterAddress(),
+ TajoMasterProtocol.class, true);
+ } catch (Exception e) {
+ queryMasterContext.getWorkerContext().setWorkerResourceTrackerAddr(HAServiceUtil.getResourceTrackerAddress(systemConf));
+ queryMasterContext.getWorkerContext().setTajoMasterAddress(HAServiceUtil.getMasterUmbilicalAddress(systemConf));
tmClient = connPool.getConnection(queryMasterContext.getWorkerContext().getTajoMasterAddress(),
TajoMasterProtocol.class, true);
}
+ } else {
+ tmClient = connPool.getConnection(queryMasterContext.getWorkerContext().getTajoMasterAddress(),
+ TajoMasterProtocol.class, true);
+ }
- TajoMasterProtocol.TajoMasterProtocolService masterClientService = tmClient.getStub();
- masterClientService.heartbeat(future.getController(), queryHeartbeat, future);
- } catch (Exception e) {
- //this function will be closed in new thread.
- //When tajo do stop cluster, tajo master maybe throw closed connection exception
+ TajoMasterProtocol.TajoMasterProtocolService masterClientService = tmClient.getStub();
+ masterClientService.heartbeat(future.getController(), queryHeartbeat, future);
+ } catch (Exception e) {
+ //this function will be closed in new thread.
+ //When tajo do stop cluster, tajo master maybe throw closed connection exception
- LOG.error(e.getMessage(), e);
- } finally {
- connPool.releaseConnection(tmClient);
- }
+ LOG.error(e.getMessage(), e);
+ } finally {
+ connPool.releaseConnection(tmClient);
+ }
- try {
- queryMasterTask.stop();
- if (!queryContext.getBool(SessionVars.DEBUG_ENABLED)) {
- cleanup(queryId);
- }
- } catch (Exception e) {
- LOG.error(e.getMessage(), e);
+ try {
+ queryMasterTask.stop();
+ if (!queryContext.getBool(SessionVars.DEBUG_ENABLED)) {
+ cleanup(queryId);
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ Query query = queryMasterTask.getQuery();
+ if (query != null) {
+ QueryHistory queryHisory = query.getQueryHistory();
+ if (queryHisory != null) {
+ query.context.getQueryMasterContext().getWorkerContext().
+ getTaskHistoryWriter().appendHistory(queryHisory);
}
- } else {
- LOG.warn("No query info:" + queryId);
}
if(workerContext.isYarnContainerMode()) {
stop();
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
index fe2752f..0f275e9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/QueryUnit.java
@@ -32,6 +32,7 @@ import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.QueryUnitId;
+import org.apache.tajo.TajoProtos.TaskAttemptState;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.ipc.TajoWorkerProtocol.FailureIntermediateProto;
import org.apache.tajo.ipc.TajoWorkerProtocol.IntermediateEntryProto;
@@ -42,8 +43,10 @@ import org.apache.tajo.master.event.QueryUnitAttemptScheduleEvent.QueryUnitAttem
import org.apache.tajo.plan.logical.*;
import org.apache.tajo.storage.DataLocation;
import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.fragment.FragmentConvertor;
import org.apache.tajo.util.Pair;
import org.apache.tajo.util.TajoIdUtils;
+import org.apache.tajo.util.history.QueryUnitHistory;
import org.apache.tajo.worker.FetchImpl;
import java.net.URI;
@@ -84,6 +87,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
private QueryUnitAttemptId successfulAttempt;
private String succeededHost;
+ private int succeededHostPort;
private int succeededPullServerPort;
private int failedAttempts;
@@ -96,6 +100,8 @@ public class QueryUnit implements EventHandler<TaskEvent> {
private static final AttemptKilledTransition ATTEMPT_KILLED_TRANSITION = new AttemptKilledTransition();
+ private QueryUnitHistory finalQueryUnitHistory;
+
protected static final StateMachineFactory
<QueryUnit, TaskState, TaskEventType, TaskEvent> stateMachineFactory =
new StateMachineFactory <QueryUnit, TaskState, TaskEventType, TaskEvent>(TaskState.NEW)
@@ -214,6 +220,76 @@ public class QueryUnit implements EventHandler<TaskEvent> {
}
}
+ public TaskAttemptState getLastAttemptStatus() {
+ QueryUnitAttempt lastAttempt = getLastAttempt();
+ if (lastAttempt != null) {
+ return lastAttempt.getState();
+ } else {
+ return TaskAttemptState.TA_ASSIGNED;
+ }
+ }
+
+ public QueryUnitHistory getQueryUnitHistory() {
+ if (finalQueryUnitHistory != null) {
+ if (finalQueryUnitHistory.getFinishTime() == 0) {
+ finalQueryUnitHistory = makeQueryUnitHistory();
+ }
+ return finalQueryUnitHistory;
+ } else {
+ return makeQueryUnitHistory();
+ }
+ }
+
+ private QueryUnitHistory makeQueryUnitHistory() {
+ QueryUnitHistory queryUnitHistory = new QueryUnitHistory();
+
+ QueryUnitAttempt lastAttempt = getLastAttempt();
+ if (lastAttempt != null) {
+ queryUnitHistory.setId(lastAttempt.getId().toString());
+ queryUnitHistory.setState(lastAttempt.getState().toString());
+ queryUnitHistory.setProgress(lastAttempt.getProgress());
+ }
+ queryUnitHistory.setHostAndPort(succeededHost + ":" + succeededHostPort);
+ queryUnitHistory.setRetryCount(this.getRetryCount());
+ queryUnitHistory.setLaunchTime(launchTime);
+ queryUnitHistory.setFinishTime(finishTime);
+
+ queryUnitHistory.setNumShuffles(getShuffleOutpuNum());
+ if (!getShuffleFileOutputs().isEmpty()) {
+ ShuffleFileOutput shuffleFileOutputs = getShuffleFileOutputs().get(0);
+ if (queryUnitHistory.getNumShuffles() > 0) {
+ queryUnitHistory.setShuffleKey("" + shuffleFileOutputs.getPartId());
+ queryUnitHistory.setShuffleFileName(shuffleFileOutputs.getFileName());
+ }
+ }
+
+ List<String> fragmentList = new ArrayList<String>();
+ for (FragmentProto eachFragment : getAllFragments()) {
+ FileFragment fileFragment = FragmentConvertor.convert(FileFragment.class, eachFragment);
+ fragmentList.add(fileFragment.toString());
+ }
+ queryUnitHistory.setFragments(fragmentList.toArray(new String[]{}));
+
+ List<String[]> fetchList = new ArrayList<String[]>();
+ for (Map.Entry<String, Set<FetchImpl>> e : getFetchMap().entrySet()) {
+ for (FetchImpl f : e.getValue()) {
+ for (URI uri : f.getSimpleURIs()){
+ fetchList.add(new String[] {e.getKey(), uri.toString()});
+ }
+ }
+ }
+
+ queryUnitHistory.setFetchs(fetchList.toArray(new String[][]{}));
+
+ List<String> dataLocationList = new ArrayList<String>();
+ for(DataLocation eachLocation: getDataLocations()) {
+ dataLocationList.add(eachLocation.toString());
+ }
+
+ queryUnitHistory.setDataLocations(dataLocationList.toArray(new String[]{}));
+ return queryUnitHistory;
+ }
+
public void setLogicalPlan(LogicalNode plan) {
this.plan = plan;
@@ -488,6 +564,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
private void finishTask() {
this.finishTime = System.currentTimeMillis();
+ finalQueryUnitHistory = makeQueryUnitHistory();
}
private static class KillNewTaskTransition implements SingleArcTransition<QueryUnit, TaskEvent> {
@@ -527,6 +604,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
task.successfulAttempt = attemptEvent.getTaskAttemptId();
task.succeededHost = attempt.getWorkerConnectionInfo().getHost();
+ task.succeededHostPort = attempt.getWorkerConnectionInfo().getPeerRpcPort();
task.succeededPullServerPort = attempt.getWorkerConnectionInfo().getPullServerPort();
task.finishTask();
@@ -542,6 +620,7 @@ public class QueryUnit implements EventHandler<TaskEvent> {
QueryUnitAttempt attempt = task.attempts.get(attemptEvent.getTaskAttemptId());
task.launchTime = System.currentTimeMillis();
task.succeededHost = attempt.getWorkerConnectionInfo().getHost();
+ task.succeededHostPort = attempt.getWorkerConnectionInfo().getPeerRpcPort();
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index 91fd22d..96534df 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@ -64,6 +64,8 @@ import org.apache.tajo.storage.StorageManager;
import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.KeyValueSet;
+import org.apache.tajo.util.history.QueryUnitHistory;
+import org.apache.tajo.util.history.SubQueryHistory;
import org.apache.tajo.worker.FetchImpl;
import java.io.IOException;
@@ -280,6 +282,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
private TaskSchedulerContext schedulerContext;
private List<IntermediateEntry> hashShuffleIntermediateEntries = new ArrayList<IntermediateEntry>();
private AtomicInteger completeReportReceived = new AtomicInteger(0);
+ private SubQueryHistory finalSubQueryHistory;
public SubQuery(QueryMasterTask.QueryMasterTaskContext context, MasterPlan masterPlan,
ExecutionBlock block, StorageManager sm) {
@@ -394,6 +397,76 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
tasks.put(task.getId(), task);
}
+ public SubQueryHistory getSubQueryHistory() {
+ if (finalSubQueryHistory != null) {
+ if (finalSubQueryHistory.getFinishTime() == 0) {
+ finalSubQueryHistory = makeSubQueryHistory();
+ finalSubQueryHistory.setQueryUnits(makeQueryUnitHistories());
+ }
+ return finalSubQueryHistory;
+ } else {
+ return makeSubQueryHistory();
+ }
+ }
+
+ private List<QueryUnitHistory> makeQueryUnitHistories() {
+ List<QueryUnitHistory> queryUnitHistories = new ArrayList<QueryUnitHistory>();
+
+ for(QueryUnit eachQueryUnit: getQueryUnits()) {
+ queryUnitHistories.add(eachQueryUnit.getQueryUnitHistory());
+ }
+
+ return queryUnitHistories;
+ }
+
+ private SubQueryHistory makeSubQueryHistory() {
+ SubQueryHistory subQueryHistory = new SubQueryHistory();
+
+ subQueryHistory.setExecutionBlockId(getId().toString());
+ subQueryHistory.setPlan(PlannerUtil.buildExplainString(block.getPlan()));
+ subQueryHistory.setState(getState().toString());
+ subQueryHistory.setStartTime(startTime);
+ subQueryHistory.setFinishTime(finishTime);
+ subQueryHistory.setSucceededObjectCount(succeededObjectCount);
+ subQueryHistory.setKilledObjectCount(killedObjectCount);
+ subQueryHistory.setFailedObjectCount(failedObjectCount);
+ subQueryHistory.setTotalScheduledObjectsCount(totalScheduledObjectsCount);
+ subQueryHistory.setHostLocalAssigned(getTaskScheduler().getHostLocalAssigned());
+ subQueryHistory.setRackLocalAssigned(getTaskScheduler().getRackLocalAssigned());
+
+ long totalInputBytes = 0;
+ long totalReadBytes = 0;
+ long totalReadRows = 0;
+ long totalWriteBytes = 0;
+ long totalWriteRows = 0;
+ int numShuffles = 0;
+ for(QueryUnit eachQueryUnit: getQueryUnits()) {
+ numShuffles = eachQueryUnit.getShuffleOutpuNum();
+ if (eachQueryUnit.getLastAttempt() != null) {
+ TableStats inputStats = eachQueryUnit.getLastAttempt().getInputStats();
+ if (inputStats != null) {
+ totalInputBytes += inputStats.getNumBytes();
+ totalReadBytes += inputStats.getReadBytes();
+ totalReadRows += inputStats.getNumRows();
+ }
+ TableStats outputStats = eachQueryUnit.getLastAttempt().getResultStats();
+ if (outputStats != null) {
+ totalWriteBytes += outputStats.getNumBytes();
+ totalWriteRows += outputStats.getNumRows();
+ }
+ }
+ }
+
+ subQueryHistory.setTotalInputBytes(totalInputBytes);
+ subQueryHistory.setTotalReadBytes(totalReadBytes);
+ subQueryHistory.setTotalReadRows(totalReadRows);
+ subQueryHistory.setTotalWriteBytes(totalWriteBytes);
+ subQueryHistory.setTotalWriteRows(totalWriteRows);
+ subQueryHistory.setNumShuffles(numShuffles);
+ subQueryHistory.setProgress(getProgress());
+ return subQueryHistory;
+ }
+
/**
* It finalizes this subquery. It is only invoked when the subquery is succeeded.
*/
@@ -1172,6 +1245,9 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
getContext().getQueryMasterContext().getQueryMaster().cleanupExecutionBlock(ebIds);
}
+
+ this.finalSubQueryHistory = makeSubQueryHistory();
+ this.finalSubQueryHistory.setQueryUnits(makeQueryUnitHistories());
}
public List<IntermediateEntry> getHashShuffleIntermediateEntries() {
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java b/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
index 9d0dcaa..6d3597d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/util/JSPUtil.java
@@ -23,10 +23,14 @@ import org.apache.tajo.catalog.FunctionDesc;
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.master.TajoMaster.MasterContext;
+import org.apache.tajo.master.ha.HAService;
import org.apache.tajo.master.querymaster.QueryInProgress;
import org.apache.tajo.master.querymaster.QueryMasterTask;
import org.apache.tajo.master.querymaster.QueryUnit;
import org.apache.tajo.master.querymaster.SubQuery;
+import org.apache.tajo.util.history.QueryUnitHistory;
+import org.apache.tajo.util.history.SubQueryHistory;
import org.apache.tajo.worker.TaskRunnerHistory;
import org.apache.tajo.worker.TaskRunner;
@@ -38,7 +42,7 @@ import static org.apache.tajo.conf.TajoConf.ConfVars;
public class JSPUtil {
static DecimalFormat decimalF = new DecimalFormat("###.0");
- public static void sortQueryUnit(QueryUnit[] queryUnits, String sortField, String sortOrder) {
+ public static void sortQueryUnitArray(QueryUnit[] queryUnits, String sortField, String sortOrder) {
if(sortField == null || sortField.isEmpty()) {
sortField = "id";
}
@@ -46,6 +50,22 @@ public class JSPUtil {
Arrays.sort(queryUnits, new QueryUnitComparator(sortField, "asc".equals(sortOrder)));
}
+ public static void sortQueryUnit(List<QueryUnit> queryUnits, String sortField, String sortOrder) {
+ if(sortField == null || sortField.isEmpty()) {
+ sortField = "id";
+ }
+
+ Collections.sort(queryUnits, new QueryUnitComparator(sortField, "asc".equals(sortOrder)));
+ }
+
+ public static void sortQueryUnitHistory(List<QueryUnitHistory> queryUnits, String sortField, String sortOrder) {
+ if(sortField == null || sortField.isEmpty()) {
+ sortField = "id";
+ }
+
+ Collections.sort(queryUnits, new QueryUnitHistoryComparator(sortField, "asc".equals(sortOrder)));
+ }
+
public static void sortTaskRunner(List<TaskRunner> taskRunners) {
Collections.sort(taskRunners, new Comparator<TaskRunner>() {
@Override
@@ -147,6 +167,43 @@ public class JSPUtil {
return subQueryList;
}
+ public static List<SubQueryHistory> sortSubQueryHistory(Collection<SubQueryHistory> subQueries) {
+ List<SubQueryHistory> subQueryList = new ArrayList<SubQueryHistory>(subQueries);
+ Collections.sort(subQueryList, new Comparator<SubQueryHistory>() {
+ @Override
+ public int compare(SubQueryHistory subQuery1, SubQueryHistory subQuery2) {
+ long q1StartTime = subQuery1.getStartTime();
+ long q2StartTime = subQuery2.getStartTime();
+
+ q1StartTime = (q1StartTime == 0 ? Long.MAX_VALUE : q1StartTime);
+ q2StartTime = (q2StartTime == 0 ? Long.MAX_VALUE : q2StartTime);
+
+ int result = compareLong(q1StartTime, q2StartTime);
+ if (result == 0) {
+ return subQuery1.getExecutionBlockId().compareTo(subQuery2.getExecutionBlockId());
+ } else {
+ return result;
+ }
+ }
+ });
+
+ return subQueryList;
+ }
+
+ public static String getMasterActiveLabel(MasterContext context) {
+ HAService haService = context.getHAService();
+ String activeLabel = "";
+ if (haService != null) {
+ if (haService.isActiveStatus()) {
+ activeLabel = "<font color='#1e90ff'>(active)</font>";
+ } else {
+ activeLabel = "<font color='#1e90ff'>(backup)</font>";
+ }
+ }
+
+ return activeLabel;
+ }
+
static class QueryUnitComparator implements Comparator<QueryUnit> {
private String sortField;
private boolean asc;
@@ -194,6 +251,53 @@ public class JSPUtil {
}
}
+ static class QueryUnitHistoryComparator implements Comparator<QueryUnitHistory> {
+ private String sortField;
+ private boolean asc;
+ public QueryUnitHistoryComparator(String sortField, boolean asc) {
+ this.sortField = sortField;
+ this.asc = asc;
+ }
+
+ @Override
+ public int compare(QueryUnitHistory queryUnit, QueryUnitHistory queryUnit2) {
+ if(asc) {
+ if("id".equals(sortField)) {
+ return queryUnit.getId().compareTo(queryUnit2.getId());
+ } else if("host".equals(sortField)) {
+ String host1 = queryUnit.getHostAndPort() == null ? "-" : queryUnit.getHostAndPort();
+ String host2 = queryUnit2.getHostAndPort() == null ? "-" : queryUnit2.getHostAndPort();
+ return host1.compareTo(host2);
+ } else if("runTime".equals(sortField)) {
+ return compareLong(queryUnit.getRunningTime(), queryUnit2.getRunningTime());
+ } else if("startTime".equals(sortField)) {
+ return compareLong(queryUnit.getLaunchTime(), queryUnit2.getLaunchTime());
+ } else {
+ return queryUnit.getId().compareTo(queryUnit2.getId());
+ }
+ } else {
+ if("id".equals(sortField)) {
+ return queryUnit2.getId().compareTo(queryUnit.getId());
+ } else if("host".equals(sortField)) {
+ String host1 = queryUnit.getHostAndPort() == null ? "-" : queryUnit.getHostAndPort();
+ String host2 = queryUnit2.getHostAndPort() == null ? "-" : queryUnit2.getHostAndPort();
+ return host2.compareTo(host1);
+ } else if("runTime".equals(sortField)) {
+ if(queryUnit2.getLaunchTime() == 0) {
+ return -1;
+ } else if(queryUnit.getLaunchTime() == 0) {
+ return 1;
+ }
+ return compareLong(queryUnit2.getRunningTime(), queryUnit.getRunningTime());
+ } else if("startTime".equals(sortField)) {
+ return compareLong(queryUnit2.getLaunchTime(), queryUnit.getLaunchTime());
+ } else {
+ return queryUnit2.getId().compareTo(queryUnit.getId());
+ }
+ }
+ }
+ }
+
static int compareLong(long a, long b) {
if(a > b) {
return 1;
@@ -246,4 +350,60 @@ public class JSPUtil {
return result;
}
+
+ public static String getPageNavigation(int currentPage, int totalPage, String url) {
+ StringBuilder sb = new StringBuilder();
+
+ int pageIndex = (currentPage - 1) / 10;
+ int totalPageIndex = (totalPage - 1) / 10;
+
+ String prefix = "";
+
+ if (pageIndex > 0) {
+ int prevPage = pageIndex * 10;
+ sb.append(prefix).append("<a href='").append(url)
+ .append("&page=").append(prevPage).append("'>")
+ .append("<</a>");
+ prefix = " ";
+ }
+
+ for (int i = 1; i <= 10; i++) {
+ int printPage = pageIndex * 10 + i;
+ if (printPage == currentPage) {
+ sb.append(prefix).append(printPage);
+ } else {
+ sb.append(prefix).append("<a href='").append(url)
+ .append("&page=").append(printPage).append("'>")
+ .append("[").append(printPage).append("]</a>");
+ }
+ prefix = " ";
+ if (printPage >= totalPage) {
+ break;
+ }
+ }
+
+ if(totalPageIndex > pageIndex) {
+ int nextPage = (pageIndex + 1) * 10 + 1;
+ sb.append(prefix).append("<a href='").append(url)
+ .append("&page=").append(nextPage).append("'>")
+ .append("></a>");
+ }
+ return sb.toString();
+ }
+
+ public static <T extends Object> List<T> getPageNavigationList(List<T> originList, int page, int pageSize) {
+ if (originList == null) {
+ return new ArrayList<T>();
+ }
+ int start = (page - 1) * pageSize;
+ int end = start + pageSize;
+ if (end > originList.size()) {
+ end = originList.size();
+ }
+ if (!originList.isEmpty()) {
+ return originList.subList(start, end);
+ }
+
+ return originList;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/History.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/History.java b/tajo-core/src/main/java/org/apache/tajo/util/history/History.java
new file mode 100644
index 0000000..0312d4c
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/History.java
@@ -0,0 +1,27 @@
+/**
+ * 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;
+
+public interface History {
+ public static enum HistoryType {
+ TASK, QUERY, QUERY_SUMMARY
+ }
+
+ public HistoryType getHistoryType();
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
new file mode 100644
index 0000000..868dfcd
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryCleaner.java
@@ -0,0 +1,136 @@
+/**
+ * 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 org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+
+import java.io.IOException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class HistoryCleaner extends Thread {
+ private static final Log LOG = LogFactory.getLog(HistoryCleaner.class);
+
+ private int historyExpireDays;
+ private AtomicBoolean stopped = new AtomicBoolean(false);
+ private Path historyParentPath;
+ private Path taskHistoryParentPath;
+ private TajoConf tajoConf;
+ private boolean isMaster;
+
+ public HistoryCleaner(TajoConf tajoConf, boolean isMaster) throws IOException {
+ super(HistoryCleaner.class.getName());
+
+ this.tajoConf = tajoConf;
+ this.isMaster = isMaster;
+ historyExpireDays = tajoConf.getIntVar(ConfVars.HISTORY_EXPIRY_TIME_DAY);
+ historyParentPath = tajoConf.getQueryHistoryDir(tajoConf);
+ taskHistoryParentPath = tajoConf.getTaskHistoryDir(tajoConf);
+ }
+
+ public void doStop() {
+ stopped.set(true);
+ this.interrupt();
+ }
+
+ @Override
+ public void run() {
+ LOG.info("History cleaner started: expiry day=" + historyExpireDays);
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
+
+ while (!stopped.get()) {
+ try {
+ Thread.sleep(60 * 60 * 12 * 1000); // 12 hours
+ } catch (InterruptedException e) {
+ }
+
+ if (stopped.get()) {
+ break;
+ }
+
+ try {
+ Calendar cal = Calendar.getInstance();
+ cal.add(Calendar.DAY_OF_MONTH, -historyExpireDays);
+
+ long cleanTargetTime = cal.getTime().getTime();
+
+ // Clean query history directory
+ if (isMaster) {
+ FileSystem fs = historyParentPath.getFileSystem(tajoConf);
+ if (fs.exists(historyParentPath)) {
+ FileStatus[] files = fs.listStatus(historyParentPath);
+ if (files != null) {
+ for (FileStatus eachFile : files) {
+ String pathName = eachFile.getPath().getName();
+ long pathTime;
+ try {
+ pathTime = df.parse(pathName).getTime();
+ } catch (ParseException e) {
+ LOG.warn(eachFile.getPath() + " is not History directory format.");
+ continue;
+ }
+
+ if (pathTime < cleanTargetTime) {
+ LOG.info("Cleaning query history dir: " + eachFile.getPath());
+ fs.delete(eachFile.getPath(), true);
+ }
+ }
+ }
+ }
+ }
+
+ if (!isMaster) {
+ // Clean task history directory
+ FileSystem fs = taskHistoryParentPath.getFileSystem(tajoConf);
+ if (fs.exists(taskHistoryParentPath)) {
+ FileStatus[] files = fs.listStatus(taskHistoryParentPath);
+ if (files != null) {
+ for (FileStatus eachFile : files) {
+ String pathName = eachFile.getPath().getName();
+ long pathTime;
+ try {
+ pathTime = df.parse(pathName).getTime();
+ } catch (ParseException e) {
+ LOG.warn(eachFile.getPath() + " is not History directory format.");
+ continue;
+ }
+
+ if (pathTime < cleanTargetTime) {
+ LOG.info("Cleaning task history dir: " + eachFile.getPath());
+ fs.delete(eachFile.getPath(), true);
+ }
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ LOG.info("History cleaner stopped");
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryReader.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryReader.java b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryReader.java
new file mode 100644
index 0000000..21bc725
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryReader.java
@@ -0,0 +1,308 @@
+/**
+ * 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 org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.QueryUnitAttemptId;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.ipc.TajoWorkerProtocol.TaskHistoryProto;
+import org.apache.tajo.master.querymaster.QueryInfo;
+import org.apache.tajo.worker.TaskHistory;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+public class HistoryReader {
+ private static final Log LOG = LogFactory.getLog(HistoryReader.class);
+
+ public static final int DEFAULT_PAGE_SIZE = 100;
+ public static final int DEFAULT_TASK_PAGE_SIZE = 2000;
+ private String processName;
+ private TajoConf tajoConf;
+ private Path historyParentPath;
+ private Path taskHistoryParentPath;
+
+ public HistoryReader(String processName, TajoConf tajoConf) throws IOException {
+ this.processName = processName.replaceAll(":", "_").toLowerCase();
+ this.tajoConf = tajoConf;
+
+ historyParentPath = tajoConf.getQueryHistoryDir(tajoConf);
+ taskHistoryParentPath = tajoConf.getTaskHistoryDir(tajoConf);
+ }
+
+ public List<QueryInfo> getQueries(String keyword) throws IOException {
+ List<QueryInfo> queryInfos = new ArrayList<QueryInfo>();
+
+ FileSystem fs = HistoryWriter.getNonCrcFileSystem(historyParentPath, tajoConf);
+ if (!fs.exists(historyParentPath)) {
+ return queryInfos;
+ }
+ FileStatus[] files = fs.listStatus(historyParentPath);
+ if (files == null || files.length == 0) {
+ return queryInfos;
+ }
+
+ for (FileStatus eachDateFile: files) {
+ if (eachDateFile.isFile()) {
+ continue;
+ }
+ FileStatus[] dateFiles = fs.listStatus(new Path(eachDateFile.getPath(), HistoryWriter.QUERY_LIST));
+ if (dateFiles == null || dateFiles.length == 0) {
+ continue;
+ }
+
+ for (FileStatus eachFile: dateFiles) {
+ if (eachFile.isDirectory()) {
+ continue;
+ }
+
+ Path path = eachFile.getPath();
+ if (!path.getName().endsWith(HistoryWriter.HISTORY_FILE_POSTFIX)) {
+ continue;
+ }
+
+ FSDataInputStream in = null;
+ try {
+ in = fs.open(path);
+
+ byte[] buf = new byte[100 * 1024];
+ while (true) {
+ int length = in.readInt();
+ if (length > buf.length) {
+ buf = new byte[length];
+ }
+ in.readFully(buf, 0, length);
+ String queryInfoJson = new String(buf, 0, length);
+ QueryInfo queryInfo = QueryInfo.fromJson(queryInfoJson);
+ if (keyword != null) {
+ if (queryInfo.getSql().indexOf(keyword) >= 0) {
+ queryInfos.add(queryInfo);
+ }
+ } else {
+ queryInfos.add(queryInfo);
+ }
+ }
+ } catch (EOFException e) {
+ } catch (Exception e) {
+ LOG.error("Reading error:" + path + ", " +e.getMessage(), e);
+ } finally {
+ in.close();
+ }
+ }
+ }
+
+ Collections.sort(queryInfos, new Comparator<QueryInfo>() {
+ @Override
+ public int compare(QueryInfo query1, QueryInfo query2) {
+ return query2.getQueryIdStr().toString().compareTo(query1.getQueryIdStr().toString());
+ }
+ });
+
+ return queryInfos;
+ }
+
+ private Path getQueryHistoryFilePath(String queryId, long startTime) throws IOException {
+ if (startTime == 0) {
+ String[] tokens = queryId.split("_");
+ if (tokens.length == 3) {
+ startTime = Long.parseLong(tokens[1]);
+ } else {
+ startTime = System.currentTimeMillis();
+ }
+ }
+ Path queryHistoryPath = HistoryWriter.getQueryHistoryFilePath(historyParentPath, queryId, startTime);
+ FileSystem fs = HistoryWriter.getNonCrcFileSystem(queryHistoryPath, tajoConf);
+
+ if (!fs.exists(queryHistoryPath)) {
+ LOG.info("No query history file: " + queryHistoryPath);
+ Calendar cal = Calendar.getInstance();
+ cal.setTimeInMillis(startTime);
+ cal.add(Calendar.DAY_OF_MONTH, -1);
+ queryHistoryPath = HistoryWriter.getQueryHistoryFilePath(historyParentPath, queryId, startTime);
+ if (!fs.exists(queryHistoryPath)) {
+ LOG.info("No query history file: " + queryHistoryPath);
+ cal.setTimeInMillis(startTime);
+ cal.add(Calendar.DAY_OF_MONTH, 1);
+ queryHistoryPath = HistoryWriter.getQueryHistoryFilePath(historyParentPath, queryId, startTime);
+ }
+ if (!fs.exists(queryHistoryPath)) {
+ LOG.info("No query history file: " + queryHistoryPath);
+ return null;
+ }
+ }
+ return queryHistoryPath;
+ }
+
+ public QueryHistory getQueryHistory(String queryId) throws IOException {
+ return getQueryHistory(queryId, 0);
+ }
+
+ public QueryHistory getQueryHistory(String queryId, long startTime) throws IOException {
+ Path queryHistoryFile = getQueryHistoryFilePath(queryId, startTime);
+ if (queryHistoryFile == null) {
+ return null;
+ }
+ FileSystem fs = HistoryWriter.getNonCrcFileSystem(queryHistoryFile, tajoConf);
+
+ FileStatus fileStatus = fs.getFileStatus(queryHistoryFile);
+ if (fileStatus.getLen() > 10 * 1024 * 1024) {
+ throw new IOException("QueryHistory file is too big: " +
+ queryHistoryFile + ", " + fileStatus.getLen() + " bytes");
+ }
+ FSDataInputStream in = null;
+ try {
+ in = fs.open(queryHistoryFile);
+ byte[] buf = new byte[(int)fileStatus.getLen()];
+
+ in.readFully(buf, 0, buf.length);
+
+ return QueryHistory.fromJson(new String(buf));
+ } finally {
+ if (in != null) {
+ in.close();
+ }
+ }
+ }
+
+ public List<QueryUnitHistory> getQueryUnitHistory(String queryId, String ebId) throws IOException {
+ Path queryHistoryFile = getQueryHistoryFilePath(queryId, 0);
+ if (queryHistoryFile == null) {
+ return new ArrayList<QueryUnitHistory>();
+ }
+ Path detailFile = new Path(queryHistoryFile.getParent(), ebId + HistoryWriter.HISTORY_FILE_POSTFIX);
+ FileSystem fs = HistoryWriter.getNonCrcFileSystem(detailFile, tajoConf);
+
+ if (!fs.exists(detailFile)) {
+ return new ArrayList<QueryUnitHistory>();
+ }
+
+ FileStatus fileStatus = fs.getFileStatus(detailFile);
+ if (fileStatus.getLen() > 100 * 1024 * 1024) { // 100MB
+ throw new IOException("QueryUnitHistory file is too big: " +
+ detailFile + ", " + fileStatus.getLen() + " bytes");
+ }
+
+ FSDataInputStream in = null;
+ try {
+ in = fs.open(detailFile);
+ byte[] buf = new byte[(int)fileStatus.getLen()];
+
+ in.readFully(buf, 0, buf.length);
+
+ return SubQueryHistory.fromJsonQueryUnits(new String(buf));
+ } finally {
+ if (in != null) {
+ in.close();
+ }
+ }
+ }
+
+ public TaskHistory getTaskHistory(String queryUnitAttemptId, long startTime) throws IOException {
+ FileSystem fs = HistoryWriter.getNonCrcFileSystem(taskHistoryParentPath, tajoConf);
+
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
+
+ Calendar cal = Calendar.getInstance();
+ cal.setTime(new Date(startTime));
+
+ //current, current-1, current+1 hour
+ String[] targetHistoryFileDates = new String[3];
+ targetHistoryFileDates[0] = df.format(cal.getTime());
+
+ cal.add(Calendar.HOUR_OF_DAY, -1);
+ targetHistoryFileDates[1] = df.format(cal.getTime());
+
+ cal.setTime(new Date(startTime));
+ cal.add(Calendar.HOUR_OF_DAY, 1);
+ targetHistoryFileDates[2] = df.format(cal.getTime());
+
+ for (String historyFileDate : targetHistoryFileDates) {
+ Path fileParent = new Path(taskHistoryParentPath, historyFileDate.substring(0, 8) + "/tasks/" + processName);
+ String hour = historyFileDate.substring(8, 10);
+
+ if (!fs.exists(fileParent)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Task history parent not exists:" + fileParent);
+ }
+ continue;
+ }
+
+ FileStatus[] files = fs.listStatus(fileParent);
+ if (files == null || files.length == 0) {
+ return null;
+ }
+
+ String filePrefix = processName + "_" + hour + "_";
+
+ for (FileStatus eachFile : files) {
+ if (eachFile.getPath().getName().indexOf(filePrefix) != 0) {
+ continue;
+ }
+
+ FSDataInputStream in = null;
+ TaskHistoryProto.Builder builder = TaskHistoryProto.newBuilder();
+ try {
+ FileStatus status = fs.getFileStatus(eachFile.getPath());
+ LOG.info("Finding TaskHistory from " + status.getLen() + "," + eachFile.getPath());
+
+ in = fs.open(eachFile.getPath());
+ while (true) {
+ int len = in.readInt();
+ byte[] buf = new byte[len];
+ in.readFully(buf, 0, len);
+
+ builder.clear();
+ TaskHistoryProto taskHistoryProto = builder.mergeFrom(buf).build();
+ QueryUnitAttemptId attemptId = new QueryUnitAttemptId(taskHistoryProto.getQueryUnitAttemptId());
+ if (attemptId.toString().equals(queryUnitAttemptId)) {
+ return new TaskHistory(taskHistoryProto);
+ }
+ }
+ } catch (EOFException e) {
+ } finally {
+ if (in != null) {
+ in.close();
+ }
+ }
+ }
+ }
+ return null;
+ }
+
+ public QueryInfo getQueryInfo(String queryId) throws IOException {
+ List<QueryInfo> queries = getQueries(null);
+
+ if (queries != null) {
+ for (QueryInfo queryInfo: queries) {
+ if (queryId.equals(queryInfo.getQueryId().toString())) {
+ return queryInfo;
+ }
+ }
+ }
+
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/e01b00a7/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
new file mode 100644
index 0000000..63a143b
--- /dev/null
+++ b/tajo-core/src/main/java/org/apache/tajo/util/history/HistoryWriter.java
@@ -0,0 +1,450 @@
+/**
+ * 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 org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.service.AbstractService;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.master.querymaster.QueryInfo;
+import org.apache.tajo.worker.TaskHistory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.*;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * History directory structure
+ * tajo.query-history.path: hdfs
+ * tajo.task-history.path: local or hdfs
+ *
+ * <tajo.history.query.dir>/<yyyyMMdd>/query-list/query-list-<HHmmss>.hist (TajoMaster's query list, hourly rolling)
+ * /query-detail/<QUERY_ID>/query.hist (QueryMaster's query detail)
+ * /<EB_ID>.hist (QueryMaster's subquery detail)
+ * <tajo.history.task.dir>/<yyyyMMdd>/tasks/<WORKER_HOST>_<WORKER_PORT>/<WORKER_HOST>_<WORKER_PORT>_<HH>_<seq>.hist
+ * History files are kept for "tajo.history.expiry-time-day" (default value is 7 days)
+ */
+public class HistoryWriter extends AbstractService {
+ private static final Log LOG = LogFactory.getLog(HistoryWriter.class);
+ public static final String QUERY_LIST = "query-list";
+ public static final String QUERY_DETAIL = "query-detail";
+ public static final String HISTORY_FILE_POSTFIX = ".hist";
+
+ private final LinkedBlockingQueue<History> historyQueue = new LinkedBlockingQueue<History>();
+ // key: yyyyMMddHH
+ private Map<String, WriterHolder> taskWriters = new HashMap<String, WriterHolder>();
+
+ // For TajoMaster's query list
+ private WriterHolder querySummaryWriter = null;
+
+ private WriterThread writerThread;
+ private AtomicBoolean stopped = new AtomicBoolean(false);
+ private Path historyParentPath;
+ private Path taskHistoryParentPath;
+ private String processName;
+ private TajoConf tajoConf;
+ private HistoryCleaner historyCleaner;
+ private boolean isMaster;
+
+ public HistoryWriter(String processName, boolean isMaster) {
+ super(HistoryWriter.class.getName() + ":" + processName);
+ this.processName = processName.replaceAll(":", "_").toLowerCase();
+ this.isMaster = isMaster;
+ }
+
+ @Override
+ public void serviceInit(Configuration conf) throws Exception {
+ tajoConf = (TajoConf)conf;
+ historyParentPath = tajoConf.getQueryHistoryDir(tajoConf);
+ taskHistoryParentPath = tajoConf.getTaskHistoryDir(tajoConf);
+ writerThread = new WriterThread();
+ historyCleaner = new HistoryCleaner(tajoConf, isMaster);
+ super.serviceInit(conf);
+ }
+
+ @Override
+ public void serviceStop() throws Exception {
+ for (WriterHolder eachWriter : taskWriters.values()) {
+ if (eachWriter.out != null) {
+ try {
+ eachWriter.out.close();
+ } catch (Exception err) {
+ LOG.error(err.getMessage(), err);
+ }
+ }
+ }
+ taskWriters.clear();
+ stopped.set(true);
+ writerThread.interrupt();
+
+ if (querySummaryWriter != null && querySummaryWriter.out != null) {
+ try {
+ querySummaryWriter.out.close();
+ } catch (Exception err) {
+ LOG.error(err.getMessage(), err);
+ }
+ }
+
+ if (historyCleaner != null) {
+ historyCleaner.doStop();
+ }
+ super.serviceStop();
+ }
+
+ @Override
+ public void serviceStart() throws Exception {
+ writerThread.start();
+ historyCleaner.start();
+ }
+
+ public void appendHistory(History history) {
+ synchronized (historyQueue) {
+ historyQueue.add(history);
+ historyQueue.notifyAll();
+ }
+ }
+
+ public static FileSystem getNonCrcFileSystem(Path path, Configuration conf) throws IOException {
+ // https://issues.apache.org/jira/browse/HADOOP-7844
+ // If FileSystem is a local and CheckSumFileSystem, flushing doesn't touch file length.
+ // So HistoryReader can't read until closing the file.
+ FileSystem fs = path.getFileSystem(conf);
+ if (path.toUri().getScheme().equals("file")) {
+ fs.setWriteChecksum(false);
+ }
+
+ return fs;
+ }
+
+ public static Path getQueryHistoryFilePath(Path historyParentPath, String queryId, long startTime) {
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
+
+ Path datePath = new Path(historyParentPath, df.format(startTime) + "/" + QUERY_DETAIL);
+ return new Path(datePath, queryId + "/query" + HISTORY_FILE_POSTFIX);
+ }
+
+ public static Path getQueryHistoryFilePath(Path historyParentPath, String queryId) {
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
+
+ Path datePath = null;
+ try {
+ String[] tokens = queryId.split("_");
+ //q_1412483083972_0005 = q_<timestamp>_<seq>
+ if (tokens.length == 3) {
+ datePath = new Path(historyParentPath, df.format(tokens[1]) + "/" + QUERY_DETAIL);
+ } else {
+ datePath = new Path(historyParentPath, df.format(new Date(System.currentTimeMillis())) + "/" + QUERY_DETAIL);
+ }
+ } catch (Exception e) {
+ datePath = new Path(historyParentPath, df.format(new Date(System.currentTimeMillis())) + "/" + QUERY_DETAIL);
+ }
+ return new Path(datePath, queryId + "/query" + HISTORY_FILE_POSTFIX);
+ }
+
+ class WriterThread extends Thread {
+ public void run() {
+ LOG.info("HistoryWriter_"+ processName + " started.");
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
+ while (!stopped.get()) {
+ List<History> histories = new ArrayList<History>();
+ synchronized (historyQueue) {
+ historyQueue.drainTo(histories);
+ if (histories.isEmpty()) {
+ try {
+ historyQueue.wait(60 * 1000);
+ } catch (InterruptedException e) {
+ if (stopped.get()) {
+ break;
+ }
+ }
+ }
+ }
+ if (stopped.get()) {
+ break;
+ }
+ try {
+ if (!histories.isEmpty()) {
+ writeHistory(histories);
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+
+ //clean up history file
+
+ // closing previous writer
+ synchronized (taskWriters) {
+ Calendar cal = Calendar.getInstance();
+ cal.add(Calendar.HOUR_OF_DAY, -2);
+ String closeTargetTime = df.format(cal.getTime());
+ List<String> closingTargets = new ArrayList<String>();
+ synchronized (taskWriters) {
+ for (String eachWriterTime : taskWriters.keySet()) {
+ if (eachWriterTime.compareTo(closeTargetTime) <= 0) {
+ closingTargets.add(eachWriterTime);
+ }
+ }
+ }
+
+ for (String eachWriterTime : closingTargets) {
+ WriterHolder writerHolder = null;
+ synchronized (taskWriters) {
+ writerHolder = taskWriters.remove(eachWriterTime);
+ }
+ if (writerHolder != null) {
+ LOG.info("Closing task history file: " + writerHolder.path);
+ if (writerHolder.out != null) {
+ try {
+ writerHolder.out.close();
+ } catch (IOException e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ }
+ }
+ }
+ }
+ LOG.info("HistoryWriter_"+ processName + " stopped.");
+ }
+
+ public void writeHistory(List<History> histories) {
+ if (histories.isEmpty()) {
+ return;
+ }
+ for (History eachHistory : histories) {
+ switch(eachHistory.getHistoryType()) {
+ case TASK:
+ try {
+ writeTaskHistory((TaskHistory) eachHistory);
+ } catch (Exception e) {
+ LOG.error("Error while saving task history: " +
+ ((TaskHistory) eachHistory).getQueryUnitAttemptId() + ":" + e.getMessage(), e);
+ }
+ break;
+ case QUERY:
+ try {
+ writeQueryHistory((QueryHistory) eachHistory);
+ } catch (Exception e) {
+ LOG.error("Error while saving query history: " +
+ ((QueryHistory) eachHistory).getQueryId() + ":" + e.getMessage(), e);
+ }
+ break;
+ case QUERY_SUMMARY:
+ try {
+ writeQuerySummary((QueryInfo) eachHistory);
+ } catch (Exception e) {
+ LOG.error("Error while saving query summary: " +
+ ((QueryInfo) eachHistory).getQueryId() + ":" + e.getMessage(), e);
+ }
+ break;
+ default:
+ LOG.warn("Wrong history type: " + eachHistory.getHistoryType());
+ }
+ }
+ }
+
+ private synchronized void writeQueryHistory(QueryHistory queryHistory) throws Exception {
+ // QueryMaster's query detail history (json format)
+ // <tajo.query-history.path>/<yyyyMMdd>/query-detail/<QUERY_ID>/query.hist
+
+ // QueryMaster's subquery detail history(proto binary format)
+ // <tajo.query-history.path>/<yyyyMMdd>/query-detail/<QUERY_ID>/<EB_ID>.hist
+
+ Path queryHistoryFile = getQueryHistoryFilePath(historyParentPath, queryHistory.getQueryId());
+ FileSystem fs = getNonCrcFileSystem(queryHistoryFile, tajoConf);
+
+ if (!fs.exists(queryHistoryFile.getParent())) {
+ if (!fs.mkdirs(queryHistoryFile.getParent())) {
+ LOG.error("Can't make QueryHistory dir: " + queryHistoryFile.getParent());
+ return;
+ }
+ }
+
+ FSDataOutputStream out = null;
+ try {
+ LOG.info("Saving query summary: " + queryHistoryFile);
+ out = fs.create(queryHistoryFile);
+ out.write(queryHistory.toJson().getBytes());
+ } finally {
+ if (out != null) {
+ try {
+ out.close();
+ } catch (Exception err) {
+ LOG.error(err.getMessage(), err);
+ }
+ }
+ }
+
+ if (queryHistory.getSubQueryHistories() != null) {
+ for (SubQueryHistory subQueryHistory : queryHistory.getSubQueryHistories()) {
+ Path path = new Path(queryHistoryFile.getParent(), subQueryHistory.getExecutionBlockId() + HISTORY_FILE_POSTFIX);
+ out = null;
+ try {
+ out = fs.create(path);
+ out.write(subQueryHistory.toQueryUnitsJson().getBytes());
+ LOG.info("Saving query unit: " + path);
+ } finally {
+ if (out != null) {
+ try {
+ out.close();
+ } catch (Exception err) {
+ LOG.error(err.getMessage(), err);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ private synchronized void writeQuerySummary(QueryInfo queryInfo) throws Exception {
+ // writing to HDFS and rolling hourly
+ if (querySummaryWriter == null) {
+ querySummaryWriter = new WriterHolder();
+ rollingQuerySummaryWriter();
+ } else {
+ if (querySummaryWriter.out == null) {
+ rollingQuerySummaryWriter();
+ } else if (System.currentTimeMillis() - querySummaryWriter.lastWritingTime >= 60 * 60 * 1000) {
+ if (querySummaryWriter.out != null) {
+ LOG.info("Close query history file: " + querySummaryWriter.path);
+ querySummaryWriter.out.close();
+ }
+ rollingQuerySummaryWriter();
+ }
+ }
+ byte[] jsonBytes = ("\n" + queryInfo.toJson() + "\n").getBytes();
+
+ querySummaryWriter.out.writeInt(jsonBytes.length);
+ querySummaryWriter.out.write(jsonBytes);
+ querySummaryWriter.out.hflush();
+ }
+
+ private synchronized void rollingQuerySummaryWriter() throws Exception {
+ // finding largest file sequence
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");
+ String currentDateTime = df.format(new Date(System.currentTimeMillis()));
+
+ Path datePath = new Path(historyParentPath, currentDateTime.substring(0, 8) + "/" + QUERY_LIST);
+ FileSystem fs = getNonCrcFileSystem(datePath, tajoConf);
+ if (!fs.exists(datePath)) {
+ if (!fs.mkdirs(datePath)) {
+ LOG.error("Can't make QueryList history dir: " + datePath.getParent());
+ return;
+ }
+ }
+
+ Path historyFile = new Path(datePath, QUERY_LIST + "-" + currentDateTime.substring(8, 14) + HISTORY_FILE_POSTFIX);
+ querySummaryWriter.path = historyFile;
+ querySummaryWriter.lastWritingTime = System.currentTimeMillis();
+ LOG.info("Create query history file: " + historyFile);
+ querySummaryWriter.out = fs.create(historyFile);
+ }
+
+ private synchronized void writeTaskHistory(TaskHistory taskHistory) throws Exception {
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
+
+ String taskStartTime = df.format(new Date(taskHistory.getStartTime()));
+
+ // taskWriters variable has three Writer(currentTime-2, currentTime-1, currentTime)
+ // because Task history writer is rolled by hourly
+ WriterHolder writerHolder = taskWriters.get(taskStartTime);
+ if (writerHolder == null) {
+ writerHolder = new WriterHolder();
+ writerHolder.out = createTaskHistoryFile(taskStartTime, writerHolder);
+ taskWriters.put(taskStartTime, writerHolder);
+ }
+ writerHolder.lastWritingTime = System.currentTimeMillis();
+
+ if (writerHolder.out != null) {
+ byte[] taskHistoryBytes = taskHistory.getProto().toByteArray();
+ writerHolder.out.writeInt(taskHistoryBytes.length);
+ writerHolder.out.write(taskHistoryBytes);
+ writerHolder.out.flush();
+ }
+ }
+
+ private FSDataOutputStream createTaskHistoryFile(String taskStartTime, WriterHolder writerHolder) throws IOException {
+ FileSystem fs = getNonCrcFileSystem(taskHistoryParentPath, tajoConf);
+ Path path = getQueryTaskHistoryPath(fs, taskHistoryParentPath, processName, taskStartTime);
+ if (!fs.exists(path)) {
+ if (!fs.mkdirs(path.getParent())) {
+ LOG.error("Can't make Query history directory: " + path);
+ return null;
+ }
+ }
+ writerHolder.path = path;
+ return fs.create(path, false);
+ }
+ }
+
+ public static Path getQueryTaskHistoryPath(FileSystem fs, Path parent,
+ String processName, String taskStartTime) throws IOException {
+ // <tajo.task-history.path>/<yyyyMMdd>/tasks/<WORKER_HOST>_<WORKER_PORT>/<WORKER_HOST>_<WORKER_PORT>_<HH>_<seq>.hist
+
+ // finding largest sequence path
+ Path fileParent = new Path(parent, taskStartTime.substring(0, 8) + "/tasks/" + processName);
+
+ String hour = taskStartTime.substring(8, 10);
+ int maxSeq = -1;
+
+ if (!fs.exists(fileParent)) {
+ maxSeq++;
+ return new Path(fileParent, processName + "_" + hour + "_" + maxSeq + HISTORY_FILE_POSTFIX);
+ }
+
+ if (!fs.isDirectory(fileParent)) {
+ throw new IOException("Task history path is not directory: " + fileParent);
+ }
+ FileStatus[] files = fs.listStatus(fileParent);
+ if (files != null) {
+ for (FileStatus eachFile: files) {
+ String[] nameTokens = eachFile.getPath().getName().split("_");
+ if (nameTokens.length != 4) {
+ continue;
+ }
+
+ if (nameTokens[2].equals(hour)) {
+ int prefixIndex = nameTokens[3].indexOf(".");
+ if (prefixIndex > 0) {
+ try {
+ int fileSeq = Integer.parseInt(nameTokens[3].substring(0, prefixIndex));
+ if (fileSeq > maxSeq) {
+ maxSeq = fileSeq;
+ }
+ } catch (NumberFormatException e) {
+ }
+ }
+ }
+ }
+ }
+
+ maxSeq++;
+ return new Path(fileParent, processName + "_" + hour + "_" + maxSeq + HISTORY_FILE_POSTFIX);
+ }
+
+ class WriterHolder {
+ long lastWritingTime;
+ Path path;
+ FSDataOutputStream out;
+ }
+}
[6/6] tajo git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/tajo into index_support
Posted by ji...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/tajo into index_support
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/4a364cf0
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/4a364cf0
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/4a364cf0
Branch: refs/heads/index_support
Commit: 4a364cf08a22ab50eb50eb6b8ca373e2f6f0ce42
Parents: a5298fb 2b8d30c
Author: Jihoon Son <ji...@apache.org>
Authored: Tue Nov 11 23:24:30 2014 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Tue Nov 11 23:24:30 2014 +0900
----------------------------------------------------------------------
CHANGES | 7 +
.../org/apache/tajo/client/QueryClient.java | 6 +
.../org/apache/tajo/client/QueryClientImpl.java | 52 +++
.../org/apache/tajo/client/TajoClientImpl.java | 13 +-
tajo-client/src/main/proto/ClientProtos.proto | 60 +++
.../main/proto/QueryMasterClientProtocol.proto | 1 +
.../main/proto/TajoMasterClientProtocol.proto | 1 +
tajo-common/pom.xml | 4 -
.../java/org/apache/tajo/conf/TajoConf.java | 26 ++
.../tajo/util/datetime/DateTimeConstants.java | 22 +
.../tajo/util/datetime/DateTimeFormat.java | 2 +-
.../apache/tajo/util/datetime/DateTimeUtil.java | 96 ++--
.../apache/tajo/util/TestDateTimeFormat.java | 12 +-
.../org/apache/tajo/util/TestDateTimeUtil.java | 79 ++--
.../datetime/DateTimePartFromUnixTimestamp.java | 18 +-
.../java/org/apache/tajo/master/TajoMaster.java | 21 +
.../tajo/master/TajoMasterClientService.java | 33 ++
.../apache/tajo/master/querymaster/Query.java | 39 ++
.../master/querymaster/QueryInProgress.java | 3 +
.../tajo/master/querymaster/QueryInfo.java | 79 +++-
.../tajo/master/querymaster/QueryMaster.java | 89 ++--
.../tajo/master/querymaster/QueryUnit.java | 79 ++++
.../tajo/master/querymaster/SubQuery.java | 76 ++++
.../main/java/org/apache/tajo/util/JSPUtil.java | 162 ++++++-
.../org/apache/tajo/util/history/History.java | 27 ++
.../tajo/util/history/HistoryCleaner.java | 136 ++++++
.../apache/tajo/util/history/HistoryReader.java | 308 +++++++++++++
.../apache/tajo/util/history/HistoryWriter.java | 450 +++++++++++++++++++
.../apache/tajo/util/history/QueryHistory.java | 151 +++++++
.../tajo/util/history/QueryUnitHistory.java | 167 +++++++
.../tajo/util/history/SubQueryHistory.java | 270 +++++++++++
.../java/org/apache/tajo/worker/TajoWorker.java | 20 +
.../tajo/worker/TajoWorkerClientService.java | 34 ++
.../main/java/org/apache/tajo/worker/Task.java | 5 +-
.../org/apache/tajo/worker/TaskHistory.java | 8 +-
.../src/main/resources/webapps/admin/query.jsp | 72 +--
.../resources/webapps/admin/querydetail.jsp | 116 +++++
.../main/resources/webapps/admin/querytasks.jsp | 249 ++++++++++
.../main/resources/webapps/admin/queryunit.jsp | 134 ++++++
.../resources/webapps/worker/querydetail.jsp | 43 +-
.../resources/webapps/worker/querytasks.jsp | 126 ++++--
.../resources/webapps/worker/taskhistory.jsp | 123 +++++
.../org/apache/tajo/client/TestTajoClient.java | 54 +++
.../engine/function/TestDateTimeFunctions.java | 15 +-
.../java/org/apache/tajo/util/TestJSPUtil.java | 37 +-
.../util/history/TestHistoryWriterReader.java | 251 +++++++++++
tajo-project/pom.xml | 5 -
tajo-rpc/pom.xml | 4 -
48 files changed, 3549 insertions(+), 236 deletions(-)
----------------------------------------------------------------------
[5/6] tajo git commit: TAJO-1161: Remove joda time dependency from
tajo-core. (Jihun Kang via hyunsik)
Posted by ji...@apache.org.
TAJO-1161: Remove joda time dependency from tajo-core. (Jihun Kang via hyunsik)
Closes #233
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/2b8d30ca
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/2b8d30ca
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/2b8d30ca
Branch: refs/heads/index_support
Commit: 2b8d30ca2ca4fbf1dfda58f482c952441178390d
Parents: f45e74f
Author: Hyunsik Choi <hy...@apache.org>
Authored: Tue Nov 11 00:55:08 2014 -0800
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Tue Nov 11 00:55:08 2014 -0800
----------------------------------------------------------------------
CHANGES | 3 +
tajo-common/pom.xml | 4 -
.../tajo/util/datetime/DateTimeConstants.java | 22 +++++
.../apache/tajo/util/datetime/DateTimeUtil.java | 96 +++++++++++++-------
.../apache/tajo/util/TestDateTimeFormat.java | 12 +--
.../org/apache/tajo/util/TestDateTimeUtil.java | 79 ++++++++++------
.../datetime/DateTimePartFromUnixTimestamp.java | 18 ++--
.../engine/function/TestDateTimeFunctions.java | 9 +-
tajo-project/pom.xml | 5 -
tajo-rpc/pom.xml | 4 -
10 files changed, 157 insertions(+), 95 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 950a108..d71905b 100644
--- a/CHANGES
+++ b/CHANGES
@@ -13,6 +13,9 @@ Release 0.9.1 - unreleased
IMPROVEMENT
+ TAJO-1161: Remove joda time dependency from tajo-core.
+ (Jihun Kang via hyunsik)
+
TAJO-1133: Add 'bin/tajo version' command. (Jihun Kang via hyunsik)
TAJO-1145: Add 'bin/tajo --help' command. (Jihun Kang via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-common/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-common/pom.xml b/tajo-common/pom.xml
index 820220c..9bbd5a7 100644
--- a/tajo-common/pom.xml
+++ b/tajo-common/pom.xml
@@ -216,10 +216,6 @@
<artifactId>gson</artifactId>
</dependency>
<dependency>
- <groupId>joda-time</groupId>
- <artifactId>joda-time</artifactId>
- </dependency>
- <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeConstants.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeConstants.java b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeConstants.java
index 353a500..e5a51ef 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeConstants.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeConstants.java
@@ -67,6 +67,7 @@ public class DateTimeConstants {
public static final long USECS_PER_HOUR = 3600000000L;
public static final long USECS_PER_MINUTE = 60000000L;
public static final long USECS_PER_SEC = 1000000L;
+ public static final long USECS_PER_MSEC = 1000L;
public static final int JULIAN_MINYEAR = -4713;
public static final int JULIAN_MINMONTH = 11;
@@ -75,6 +76,27 @@ public class DateTimeConstants {
/** == DateTimeUtil.toJulianDate(JULIAN_MAXYEAR, 1, 1) */
public static final int JULIAN_MAX = 2147483494;
+
+ /** the first ISO day of week */
+ public static final int MONDAY = 1;
+
+ /** the second ISO day of week */
+ public static final int TUESDAY = 2;
+
+ /** the third ISO day of week */
+ public static final int WEDNESDAY = 3;
+
+ /** the fourth ISO day of week */
+ public static final int THURSDAY = 4;
+
+ /** the fifth ISO day of week */
+ public static final int FRIDAY = 5;
+
+ /** the sixth ISO day of week */
+ public static final int SATURDAY = 6;
+
+ /** the seventh ISO day of week */
+ public static final int SUNDAY = 7;
// Julian-date equivalents of Day 0 in Unix and Postgres reckoning
/** == DateTimeUtil.toJulianDate(1970, 1, 1) */
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeUtil.java b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeUtil.java
index 9198f5b..a9dc4e8 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeUtil.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/datetime/DateTimeUtil.java
@@ -24,8 +24,6 @@ import org.apache.tajo.exception.ValueOutOfRangeException;
import org.apache.tajo.util.datetime.DateTimeConstants.DateStyle;
import org.apache.tajo.util.datetime.DateTimeConstants.DateToken;
import org.apache.tajo.util.datetime.DateTimeConstants.TokenField;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
import java.util.TimeZone;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -2088,54 +2086,88 @@ public class DateTimeUtil {
minutes = minutes - hours * DateTimeConstants.MINS_PER_HOUR;
StringBuilder sb = new StringBuilder();
- String prefix = "";
-
sb.append(totalSecs > 0 ? "+" : "-").append(String.format("%02d", hours));
if (minutes > 0) {
sb.append(":").append(String.format("%02d", minutes));
- prefix = ":";
}
return sb.toString();
}
- public static long getDay(DateTime dateTime) {
- return convertToMicroSeconds(dateTime.withTimeAtStartOfDay());
- }
-
- public static long getHour(DateTime dateTime) {
- return convertToMicroSeconds(dateTime.withTime(dateTime.get(org.joda.time.DateTimeFieldType.hourOfDay()), 0, 0, 0));
- }
-
- public static long getMinute(DateTime dateTime) {
- return convertToMicroSeconds(dateTime.withTime(dateTime.get(org.joda.time.DateTimeFieldType.hourOfDay()),
- dateTime.get(org.joda.time.DateTimeFieldType.minuteOfHour()), 0, 0));
+ public static long getDay(TimeMeta dateTime) {
+ long usecs = 0;
+
+ usecs = julianTimeToJavaTime(toJulianTimestamp(dateTime.years, dateTime.monthOfYear, dateTime.dayOfMonth,
+ 0, 0, 0, 0)) * DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
}
- public static long getSecond(DateTime dateTime) {
- return convertToMicroSeconds(dateTime.withTime(dateTime.get(org.joda.time.DateTimeFieldType.hourOfDay()),
- dateTime.get(org.joda.time.DateTimeFieldType.minuteOfHour()), dateTime.get(org.joda.time.DateTimeFieldType.secondOfMinute()), 0));
+ public static long getHour(TimeMeta dateTime) {
+ long usecs = 0;
+
+ usecs = julianTimeToJavaTime(toJulianTimestamp(dateTime.years, dateTime.monthOfYear, dateTime.dayOfMonth,
+ dateTime.hours, 0, 0, 0)) * DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
}
- public static long getMonth(DateTime dateTime) {
- return convertToMicroSeconds(dateTime.withTimeAtStartOfDay().withDate(dateTime.getYear(),
- dateTime.getMonthOfYear(),1));
+ public static long getMinute(TimeMeta dateTime) {
+ long usecs = 0;
+
+ usecs = julianTimeToJavaTime(toJulianTimestamp(dateTime.years, dateTime.monthOfYear, dateTime.dayOfMonth,
+ dateTime.hours, dateTime.minutes, 0, 0)) * DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
}
- public static long getDayOfWeek(DateTime dateTime,int week) {
- return convertToMicroSeconds(dateTime.withTimeAtStartOfDay().withDayOfWeek(week));
+ public static long getSecond(TimeMeta dateTime) {
+ long usecs = 0;
+
+ usecs = julianTimeToJavaTime(toJulianTimestamp(dateTime.years, dateTime.monthOfYear, dateTime.dayOfMonth,
+ dateTime.hours, dateTime.minutes, dateTime.secs, 0)) * DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
}
- public static long getYear (DateTime dateTime) {
- return convertToMicroSeconds(dateTime.withTimeAtStartOfDay().withDate(dateTime.getYear(), 1, 1));
+ public static long getMonth(TimeMeta dateTime) {
+ long usecs = 0;
+
+ usecs = julianTimeToJavaTime(toJulianTimestamp(dateTime.years, dateTime.monthOfYear, 1, 0, 0, 0, 0)) *
+ DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
}
- public static DateTime getUTCDateTime(Int8Datum int8Datum){
- return new DateTime(int8Datum.asInt8()/1000, DateTimeZone.UTC);
- }
-
- public static long convertToMicroSeconds(DateTime dateTime) {
- return dateTime.getMillis() * 1000;
+ public static long getDayOfWeek(TimeMeta dateTime, int weekday) {
+ if (weekday < 1 || weekday > 7) {
+ throw new RuntimeException("Weekday is out of range. Actual : " + weekday);
+ }
+
+ int week = date2isoweek(dateTime.years, dateTime.monthOfYear, dateTime.dayOfMonth);
+ int jday = isoweek2j(dateTime.years, week);
+ long usecs = 0;
+
+ jday += (weekday - 1);
+
+ jday -= DateTimeConstants.POSTGRES_EPOCH_JDATE;
+ usecs = julianTimeToJavaTime(toJulianTimestamp(jday, 0, 0, 0, 0)) *
+ DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
+ }
+
+ public static long getYear(TimeMeta dateTime) {
+ long usecs = 0;
+
+ usecs = julianTimeToJavaTime(toJulianTimestamp(dateTime.years, 1, 1, 0, 0, 0, 0)) *
+ DateTimeConstants.USECS_PER_MSEC;
+ return usecs;
+ }
+
+ public static TimeMeta getUTCDateTime(Int8Datum int8Datum){
+ long usecs = int8Datum.asInt8()%DateTimeConstants.USECS_PER_MSEC;
+ long julianTimestamp = javaTimeToJulianTime(int8Datum.asInt8()/DateTimeConstants.USECS_PER_MSEC);
+ TimeMeta tm = new TimeMeta();
+
+ julianTimestamp += usecs;
+ toJulianTimeMeta(julianTimestamp, tm);
+ return tm;
}
+
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeFormat.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeFormat.java b/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeFormat.java
index 0bfe708..2b5dba0 100644
--- a/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeFormat.java
+++ b/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeFormat.java
@@ -21,8 +21,6 @@ package org.apache.tajo.util;
import org.apache.tajo.datum.TimestampDatum;
import org.apache.tajo.util.datetime.DateTimeFormat;
import org.apache.tajo.util.datetime.TimeMeta;
-import org.joda.time.DateTime;
-import org.joda.time.format.DateTimeFormatter;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -117,19 +115,11 @@ public class TestDateTimeFormat {
@Test
public void testPerformance() {
- DateTimeFormatter jodaFormat = org.joda.time.format.DateTimeFormat.forPattern("YYYY-MM-DD HH:mm:ss.SSS");
long startTime = System.currentTimeMillis();
for (int i = 0; i < 10000000; i++) {
- DateTime dateTime = jodaFormat.parseDateTime("1997-12-30 11:40:50.345");
+ DateTimeFormat.toTimestamp("1997-12-30 11:40:50.345", "YYYY-MM-DD HH24:MI:SS.MS");
}
long endTime = System.currentTimeMillis();
- System.out.println("total parse time with JodaTime:" + (endTime - startTime) + " ms");
-
- startTime = System.currentTimeMillis();
- for (int i = 0; i < 10000000; i++) {
- TimestampDatum datum = DateTimeFormat.toTimestamp("1997-12-30 11:40:50.345", "YYYY-MM-DD HH24:MI:SS.MS");
- }
- endTime = System.currentTimeMillis();
System.out.println("total parse time with TajoDateTimeFormat:" + (endTime - startTime) + " ms");
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeUtil.java b/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeUtil.java
index e953d44..a54fb94 100644
--- a/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeUtil.java
+++ b/tajo-common/src/test/java/org/apache/tajo/util/TestDateTimeUtil.java
@@ -19,30 +19,21 @@
package org.apache.tajo.util;
import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.Int8Datum;
import org.apache.tajo.util.datetime.DateTimeConstants;
import org.apache.tajo.util.datetime.DateTimeUtil;
import org.apache.tajo.util.datetime.TimeMeta;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
-import org.joda.time.format.DateTimeFormat;
-import org.joda.time.format.DateTimeFormatter;
import org.junit.Test;
import java.util.Calendar;
+import java.util.Date;
import java.util.TimeZone;
import static org.junit.Assert.*;
public class TestDateTimeUtil {
- private static final int TEST_YEAR = 2014;
- private static final int TEST_MONTH_OF_YEAR = 4;
- private static final int TEST_DAY_OF_MONTH = 18;
- private static final int TEST_HOUR_OF_DAY = 0;
- private static final int TEST_MINUTE_OF_HOUR = 15;
- private static final int TEST_SECOND_OF_MINUTE = 25;
- private static final DateTime TEST_DATETIME = new DateTime(TEST_YEAR, TEST_MONTH_OF_YEAR, TEST_DAY_OF_MONTH,
- TEST_HOUR_OF_DAY, TEST_MINUTE_OF_HOUR, TEST_SECOND_OF_MINUTE, DateTimeZone.UTC);
- private static final DateTimeFormatter fmt = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
+ private static final TimeMeta TEST_DATETIME = DateTimeUtil.decodeDateTime("2014-04-18 01:15:25.69148");
@Test
public void testDecodeDateTime() {
@@ -400,40 +391,76 @@ public class TestDateTimeUtil {
assertEquals("-09", DateTimeUtil.getTimeZoneDisplayTime(TimeZone.getTimeZone("GMT-9")));
assertEquals("-09:10", DateTimeUtil.getTimeZoneDisplayTime(TimeZone.getTimeZone("GMT-9:10")));
}
-
+
@Test
public void testGetYear() {
- assertEquals(DateTime.parse("2014-01-01 00:00:00", fmt.withZoneUTC()).getMillis() * 1000,
- DateTimeUtil.getYear(TEST_DATETIME));
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-01-01 00:00:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getYear(TEST_DATETIME));
}
@Test
public void testGetMonth() {
- assertEquals(DateTime.parse("2014-04-01 00:00:00", fmt.withZoneUTC()).getMillis() * 1000,
- DateTimeUtil.getMonth(TEST_DATETIME));
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-01 00:00:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getMonth(TEST_DATETIME));
}
@Test
public void testGetDay() {
- assertEquals(DateTime.parse("2014-04-18 00:00:00", fmt.withZoneUTC()).getMillis() * 1000,
- DateTimeUtil.getDay(TEST_DATETIME));
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-18 00:00:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getDay(TEST_DATETIME));
+ }
+
+ @Test
+ public void testGetDayOfWeek() {
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-14 00:00:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getDayOfWeek(TEST_DATETIME, DateTimeConstants.MONDAY));
+
+ javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-15 00:00:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getDayOfWeek(TEST_DATETIME, DateTimeConstants.TUESDAY));
}
@Test
public void testGetHour() {
- assertEquals(DateTime.parse("2014-04-18 00:00:00",fmt.withZoneUTC()).getMillis() * 1000,
- DateTimeUtil.getHour(TEST_DATETIME));
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-18 01:00:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getHour(TEST_DATETIME));
}
@Test
public void testGetMinute() {
- assertEquals(DateTime.parse("2014-04-18 00:15:00",fmt.withZoneUTC()).getMillis() * 1000,
- DateTimeUtil.getMinute(TEST_DATETIME));
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-18 01:15:00")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getMinute(TEST_DATETIME));
}
@Test
public void testGetSecond() {
- assertEquals(DateTime.parse("2014-04-18 00:15:25",fmt.withZoneUTC()).getMillis() * 1000,
- DateTimeUtil.getSecond(TEST_DATETIME));
+ long javaTimestamp =
+ DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp("2014-04-18 01:15:25")) *
+ DateTimeConstants.USECS_PER_MSEC;
+ assertEquals(javaTimestamp, DateTimeUtil.getSecond(TEST_DATETIME));
+ }
+
+ @Test
+ public void testGetUTCDateTime() {
+ long javaTimestamp = DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp(TEST_DATETIME)) *
+ DateTimeConstants.USECS_PER_MSEC;
+ javaTimestamp += (TEST_DATETIME.fsecs%DateTimeConstants.USECS_PER_MSEC);
+ Int8Datum datum = DatumFactory.createInt8(javaTimestamp);
+
+ assertTrue(TEST_DATETIME.equals(DateTimeUtil.getUTCDateTime(datum)));
}
+
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
index 97b5b88..4e0449d 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/function/datetime/DateTimePartFromUnixTimestamp.java
@@ -26,7 +26,7 @@ import org.apache.tajo.engine.function.annotation.Description;
import org.apache.tajo.engine.function.annotation.ParamTypes;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.util.datetime.DateTimeUtil;
-import org.joda.time.DateTime;
+import org.apache.tajo.util.datetime.TimeMeta;
import static org.apache.tajo.common.TajoDataTypes.Type.*;
@@ -58,7 +58,7 @@ public class DateTimePartFromUnixTimestamp extends GeneralFunction {
public Datum eval(Tuple params) {
Datum target = params.get(0);
- DateTime dateTime;
+ TimeMeta dateTime;
Int4Datum dayOfWeek = null;
if (target instanceof NullDatum || params.get(1) instanceof NullDatum) {
@@ -97,44 +97,44 @@ public class DateTimePartFromUnixTimestamp extends GeneralFunction {
}
private interface DateTimePartExtractorFromUnixTime {
- public Datum extract(DateTime dateTime);
+ public Datum extract(TimeMeta dateTime);
}
private interface WeekPartExtractorFromUnixTime {
- public Datum extract(DateTime dateTime, int week);
+ public Datum extract(TimeMeta dateTime, int week);
}
private class DayExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
- public Datum extract(DateTime dateTime) {
+ public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getDay(dateTime));
}
}
private class HourExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
- public Datum extract(DateTime dateTime) {
+ public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getHour(dateTime));
}
}
private class MonthExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
- public Datum extract(DateTime dateTime) {
+ public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getMonth(dateTime));
}
}
private class YearExtractorFromTime implements DateTimePartExtractorFromUnixTime {
@Override
- public Datum extract(DateTime dateTime) {
+ public Datum extract(TimeMeta dateTime) {
return DatumFactory.createInt8(DateTimeUtil.getYear(dateTime));
}
}
private class WeekExtractorFromTime implements WeekPartExtractorFromUnixTime {
@Override
- public Datum extract(DateTime dateTime , int week) {
+ public Datum extract(TimeMeta dateTime , int week) {
return DatumFactory.createInt8(DateTimeUtil.getDayOfWeek(dateTime,week));
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
index 619bc5b..7cca13d 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/function/TestDateTimeFunctions.java
@@ -24,7 +24,8 @@ import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.datum.TimestampDatum;
import org.apache.tajo.engine.eval.ExprTestBase;
-import org.joda.time.DateTime;
+import org.apache.tajo.util.datetime.DateTimeUtil;
+import org.apache.tajo.util.datetime.TimeMeta;
import org.junit.Test;
import java.io.IOException;
@@ -99,11 +100,11 @@ public class TestDateTimeFunctions extends ExprTestBase {
@Test
public void testToChar() throws IOException {
long expectedTimestamp = System.currentTimeMillis();
- DateTime expectedDateTime = new DateTime(expectedTimestamp);
- String dateFormatStr = "yyyy-MM";
+ TimeMeta tm = new TimeMeta();
+ DateTimeUtil.toJulianTimeMeta(DateTimeUtil.javaTimeToJulianTime(expectedTimestamp), tm);
// (expectedTimestamp / 1000) means the translation from millis seconds to unix timestamp
String q = String.format("select to_char(to_timestamp(%d), 'yyyy-MM');", (expectedTimestamp / 1000));
- testSimpleEval(q, new String[]{expectedDateTime.toString(dateFormatStr)});
+ testSimpleEval(q, new String[]{String.format("%04d-%02d", tm.years, tm.monthOfYear)});
q = "select to_char(to_timestamp('1997-12-30 11:40:00', 'YYYY-MM-DD HH24:MI:SS'), 'YYYY-MM-DD HH24:MI:SS')";
testSimpleEval(q, new String[]{"1997-12-30 11:40:00"});
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-project/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-project/pom.xml b/tajo-project/pom.xml
index aa8110f..5493095 100644
--- a/tajo-project/pom.xml
+++ b/tajo-project/pom.xml
@@ -1011,11 +1011,6 @@
<version>4.0.23.Final</version>
</dependency>
<dependency>
- <groupId>joda-time</groupId>
- <artifactId>joda-time</artifactId>
- <version>2.3</version>
- </dependency>
- <dependency>
<groupId>org.apache.derby</groupId>
<artifactId>derby</artifactId>
<version>10.8.2.2</version>
http://git-wip-us.apache.org/repos/asf/tajo/blob/2b8d30ca/tajo-rpc/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-rpc/pom.xml b/tajo-rpc/pom.xml
index 9079ecd..6a7f379 100644
--- a/tajo-rpc/pom.xml
+++ b/tajo-rpc/pom.xml
@@ -141,10 +141,6 @@
<artifactId>tajo-common</artifactId>
<exclusions>
<exclusion>
- <groupId>joda-time</groupId>
- <artifactId>joda-time</artifactId>
- </exclusion>
- <exclusion>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</exclusion>
[2/6] tajo git commit: TAJO-1026: Implement Query history persistency
manager.
Posted by ji...@apache.org.
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"/>
+ <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>
+
+ Page Size: <input type="text" name="pageSize" value="<%=pageSize%>" size="5"/>
+
+ <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>
+ Page Size: <input type="text" name="pageSize" value="<%=pageSize%>" size="5"/>
+
<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
+ <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));
+ }
}