diff --git a/itests/hive-unit/src/test/java/org/apache/hive/service/cli/session/TestQueryDisplay.java b/itests/hive-unit/src/test/java/org/apache/hive/service/cli/session/TestQueryDisplay.java index 98581e0e09c20b30eafd6aae99068fddf1596d8e..cc18ce79568941caeb20ea7e5708adbb13078547 100644 --- a/itests/hive-unit/src/test/java/org/apache/hive/service/cli/session/TestQueryDisplay.java +++ b/itests/hive-unit/src/test/java/org/apache/hive/service/cli/session/TestQueryDisplay.java @@ -154,8 +154,8 @@ private void verifyDDL(SQLOperationDisplay display, String stmt, String handle, Assert.assertTrue(qDisplay1.getPerfLogStarts(QueryDisplay.Phase.COMPILATION).size() > 0); Assert.assertTrue(qDisplay1.getPerfLogEnds(QueryDisplay.Phase.COMPILATION).size() > 0); - Assert.assertEquals(qDisplay1.getTaskDisplays().size(), 2); - QueryDisplay.TaskDisplay tInfo1 = qDisplay1.getTaskDisplays().get(1); + Assert.assertEquals(qDisplay1.getTaskDisplays().size(), 1); + QueryDisplay.TaskDisplay tInfo1 = qDisplay1.getTaskDisplays().get(0); Assert.assertEquals(tInfo1.getTaskId(), "Stage-0"); Assert.assertEquals(tInfo1.getTaskType(), StageType.DDL); Assert.assertTrue(tInfo1.getBeginTime() > 0 && tInfo1.getBeginTime() <= System.currentTimeMillis()); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java index dad43fbbab0359711320a5c4656d67fda6fb73f4..32d2cb268406332b96cdc6b15c72b3716caa75d7 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java @@ -450,7 +450,7 @@ public void run() { schema = getSchema(sem, conf); plan = new QueryPlan(queryStr, sem, perfLogger.getStartTime(PerfLogger.DRIVER_RUN), queryId, - queryState.getHiveOperation(), schema, queryDisplay); + queryState.getHiveOperation(), schema); conf.setQueryString(queryStr); @@ -1507,7 +1507,7 @@ public int execute() throws CommandNeedRetryException { perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.PRE_HOOK + peh.getClass().getName()); } } - + setQueryDisplays(plan.getRootTasks()); int mrJobs = Utilities.getMRTasks(plan.getRootTasks()).size(); int jobs = mrJobs + Utilities.getTezTasks(plan.getRootTasks()).size() @@ -1739,6 +1739,15 @@ public int execute() throws CommandNeedRetryException { return (0); } + private void setQueryDisplays(List> tasks) { + if (tasks != null) { + for (Task task : tasks) { + task.setQueryDisplay(queryDisplay); + setQueryDisplays(task.getDependentTasks()); + } + } + } + private void logMrWarning(int mrJobs) { if (mrJobs <= 0 || !("mr".equals(HiveConf.getVar(conf, ConfVars.HIVE_EXECUTION_ENGINE)))) { return; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/QueryDisplay.java b/ql/src/java/org/apache/hadoop/hive/ql/QueryDisplay.java index d582bc063fc150002a01d63451ae6632fca29ac1..703e99715713a54437ca1b2dbf5dfce432d32ee4 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/QueryDisplay.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/QueryDisplay.java @@ -79,6 +79,8 @@ private String name; private boolean requireLock; private boolean retryIfFail; + private String statusMessage; + // required for jackson public TaskDisplay() { @@ -158,15 +160,28 @@ public synchronized String getExternalHandle() { if (externalHandle == null && tTask.getExternalHandle() != null) { this.externalHandle = tTask.getExternalHandle(); } + setStatusMessage(tTask.getStatusMessage()); switch (taskState) { case RUNNING: - beginTime = System.currentTimeMillis(); + if (beginTime == null) { + beginTime = System.currentTimeMillis(); + } break; case FINISHED: - endTime = System.currentTimeMillis(); + if (endTime == null) { + endTime = System.currentTimeMillis(); + } break; } } + + public synchronized String getStatusMessage() { + return statusMessage; + } + + public synchronized void setStatusMessage(String statusMessage) { + this.statusMessage = statusMessage; + } } public synchronized void setTaskResult(String taskId, TaskResult result) { TaskDisplay taskDisplay = tasks.get(taskId); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/QueryPlan.java b/ql/src/java/org/apache/hadoop/hive/ql/QueryPlan.java index ef0923d555ba662b4ed30ef45a3d72760cdfad52..e8c8ae64561db26404acf8bd7939c5e2c5fe0069 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/QueryPlan.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/QueryPlan.java @@ -113,26 +113,12 @@ public QueryPlan() { } public QueryPlan(String queryString, BaseSemanticAnalyzer sem, Long startTime, String queryId, - HiveOperation operation, Schema resultSchema) { - this(queryString, sem, startTime, queryId, operation, resultSchema, null); - } - public QueryPlan(String queryString, BaseSemanticAnalyzer sem, Long startTime, String queryId, - HiveOperation operation, Schema resultSchema, QueryDisplay queryDisplay) { + HiveOperation operation, Schema resultSchema) { this.queryString = queryString; rootTasks = new ArrayList>(sem.getAllRootTasks()); reducerTimeStatsPerJobList = new ArrayList(); fetchTask = sem.getFetchTask(); - if (queryDisplay != null) { - if (fetchTask != null) { - fetchTask.setQueryDisplay(queryDisplay); - } - if (rootTasks!= null) { - for (Task t : rootTasks) { - t.setQueryDisplay(queryDisplay); - } - } - } // Note that inputs and outputs can be changed when the query gets executed inputs = sem.getAllInputs(); outputs = sem.getAllOutputs(); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java index c96c8135a344049e57167559c4d760b876a42ca5..52cb4457542d063a4862d1aecce6eb7405242eba 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java @@ -200,6 +200,7 @@ public String getName() { public boolean addDependentTask(Task dependent) { boolean ret = false; if (getListTasks() != null) { + ret = true; for (Task tsk : getListTasks()) { ret = ret & tsk.addDependentTask(dependent); } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java index 897af5e8c3e876f7b4f6a7aa5d462f8115a4e238..34bdafda2c4b6f0d0971bdb457843f1d0e682746 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Task.java @@ -83,8 +83,18 @@ protected String id; protected T work; private TaskState taskState = TaskState.CREATED; + private String statusMessage; private transient boolean fetchSource; + public void setStatusMessage(String statusMessage) { + this.statusMessage = statusMessage; + updateStatusInQueryDisplay(); + } + + public String getStatusMessage() { + return statusMessage; + } + public enum FeedType { DYNAMIC_PARTITIONS, // list of dynamic partitions } @@ -138,13 +148,13 @@ public void setQueryDisplay(QueryDisplay queryDisplay) { this.queryDisplay = queryDisplay; } - private void updateStatusInQueryDisplay() { + protected void updateStatusInQueryDisplay() { if (queryDisplay != null) { queryDisplay.updateTaskStatus(this); } } - private void setState(TaskState state) { + protected void setState(TaskState state) { this.taskState = state; updateStatusInQueryDisplay(); } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java index 639b0da6c8da2a0eec018ce57c12e21637f1c0de..926f6e883030b5a01d025994bd02c67f0f5a275c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java @@ -432,7 +432,7 @@ public int execute(DriverContext driverContext) { // Finally SUBMIT the JOB! rj = jc.submitJob(job); this.jobID = rj.getJobID(); - + updateStatusInQueryDisplay(); returnVal = jobExecHelper.progress(rj, jc); success = (returnVal == 0); } catch (Exception e) { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java index 760ba6c8f5afb2d3e8c533ffce8be5f73d13a9e2..11f5cfd026ecd3bc51ad7832914a789df81f7423 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HadoopJobExecHelper.java @@ -367,12 +367,14 @@ private MapRedStats progress(ExecDriverTaskHandle th) throws IOException { } } console.printInfo(output); + task.setStatusMessage(output); reportTime = System.currentTimeMillis(); } if (cpuMsec > 0) { - console.printInfo("MapReduce Total cumulative CPU time: " - + Utilities.formatMsecToStr(cpuMsec)); + String status = "MapReduce Total cumulative CPU time: " + Utilities.formatMsecToStr(cpuMsec); + console.printInfo(status); + task.setStatusMessage(status); } boolean success; diff --git a/service/src/test/org/apache/hive/service/cli/CLIServiceTest.java b/service/src/test/org/apache/hive/service/cli/CLIServiceTest.java index ff7e9a4e1c1ec61454fed173aaab0c344d1241d4..fb8ee4cd8a5d65d134bc7686c7e1a1e73008de95 100644 --- a/service/src/test/org/apache/hive/service/cli/CLIServiceTest.java +++ b/service/src/test/org/apache/hive/service/cli/CLIServiceTest.java @@ -641,7 +641,8 @@ public void testTaskStatus() throws Exception { SessionHandle sessionHandle = setupTestData(tableName, columnDefinitions, confOverlay); assertNotNull(sessionHandle); // nonblocking execute - String select = "SELECT ID + ' ' FROM TEST_EXEC_ASYNC"; + String select = "select a.id, b.id from (SELECT ID + ' ' `ID` FROM TEST_EXEC_ASYNC) a full outer join " + + "(SELECT ID + ' ' `ID` FROM TEST_EXEC_ASYNC) b on a.ID=b.ID"; OperationHandle ophandle = client.executeStatementAsync(sessionHandle, select, confOverlay); @@ -697,6 +698,7 @@ private void checkTaskStatuses(List taskDisplays) { case FINISHED: if (taskDisplay.getTaskType() == StageType.MAPRED || taskDisplay.getTaskType() == StageType.MAPREDLOCAL) { assertNotNull(taskDisplay.getExternalHandle()); + assertNotNull(taskDisplay.getStatusMessage()); } assertNotNull(taskDisplay.getBeginTime()); assertNotNull(taskDisplay.getEndTime());