You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lens.apache.org by ra...@apache.org on 2015/10/09 06:17:35 UTC
[30/50] [abbrv] lens git commit: LENS-693 : Queries get purged as
soon as they are finished
LENS-693 : Queries get purged as soon as they are finished
Project: http://git-wip-us.apache.org/repos/asf/lens/repo
Commit: http://git-wip-us.apache.org/repos/asf/lens/commit/669e8727
Tree: http://git-wip-us.apache.org/repos/asf/lens/tree/669e8727
Diff: http://git-wip-us.apache.org/repos/asf/lens/diff/669e8727
Branch: refs/heads/current-release-line
Commit: 669e8727292a297487dbeb2b3edc6f17c2cea651
Parents: d457dd0
Author: Rajat Khandelwal <pr...@apache.org>
Authored: Wed Sep 23 14:36:40 2015 +0530
Committer: Amareshwari Sriramadasu <am...@apache.org>
Committed: Wed Sep 23 14:36:40 2015 +0530
----------------------------------------------------------------------
.../main/java/org/apache/lens/api/LensConf.java | 7 +-
.../lens/api/query/InMemoryQueryResult.java | 24 +-
.../org/apache/lens/api/query/LensQuery.java | 8 +-
.../lens/api/query/PersistentQueryResult.java | 18 +
.../org/apache/lens/api/query/QueryResult.java | 4 +-
.../org/apache/lens/api/query/QueryStatus.java | 1 +
.../lens/cli/commands/LensQueryCommands.java | 9 +-
.../org/apache/lens/driver/hive/HiveDriver.java | 5 +-
.../driver/hive/HivePersistentResultSet.java | 2 +-
.../apache/lens/driver/hive/TestHiveDriver.java | 2 +-
.../lens/server/api/LensConfConstants.java | 8 +-
.../server/api/driver/InMemoryResultSet.java | 24 +-
.../lens/server/api/driver/LensResultSet.java | 9 +
.../server/api/driver/PersistentResultSet.java | 19 +-
.../server/api/query/FinishedLensQuery.java | 5 +-
.../lens/server/api/query/QueryCancelled.java | 10 +-
.../lens/server/api/query/QueryClosed.java | 10 +-
.../lens/server/api/query/QueryContext.java | 6 +-
.../lens/server/api/query/QueryEnded.java | 8 +-
.../lens/server/api/query/QueryFailed.java | 10 +-
.../lens/server/api/query/QuerySuccess.java | 10 +-
.../lens/server/api/driver/MockDriver.java | 9 +-
.../lens/server/query/LensPersistentResult.java | 73 +++-
.../lens/server/query/QueryEndNotifier.java | 79 ++---
.../server/query/QueryExecutionServiceImpl.java | 345 +++++++++----------
.../QueryExecutionStatisticsGenerator.java | 11 +-
.../lens/server/query/ResultFormatter.java | 17 +-
.../src/main/resources/lensserver-default.xml | 6 +-
.../org/apache/lens/server/LensJerseyTest.java | 21 +-
.../lens/server/common/RestAPITestUtil.java | 122 +++++--
.../lens/server/query/TestEventService.java | 11 +-
.../server/query/TestQueryEndEmailNotifier.java | 196 +++++------
.../lens/server/query/TestQueryService.java | 253 +++++---------
.../lens/server/query/TestResultFormatting.java | 11 +-
lens-server/src/test/resources/lens-site.xml | 6 +-
src/site/apt/admin/config.apt | 42 +--
36 files changed, 754 insertions(+), 647 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-api/src/main/java/org/apache/lens/api/LensConf.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/LensConf.java b/lens-api/src/main/java/org/apache/lens/api/LensConf.java
index 3b1ad87..ff965d6 100644
--- a/lens-api/src/main/java/org/apache/lens/api/LensConf.java
+++ b/lens-api/src/main/java/org/apache/lens/api/LensConf.java
@@ -63,11 +63,14 @@ public class LensConf implements Serializable {
properties.put(key, value);
}
+ public void addProperty(Object key, Object value) {
+ properties.put(String.valueOf(key), String.valueOf(value));
+ }
+
/**
* Adds Map of properties.
*
- * @param key the key
- * @param value the value
+ * @param props the properties
*/
public void addProperties(Map<String, String> props) {
properties.putAll(props);
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-api/src/main/java/org/apache/lens/api/query/InMemoryQueryResult.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/query/InMemoryQueryResult.java b/lens-api/src/main/java/org/apache/lens/api/query/InMemoryQueryResult.java
index 6a9f455..36d0ba0 100644
--- a/lens-api/src/main/java/org/apache/lens/api/query/InMemoryQueryResult.java
+++ b/lens-api/src/main/java/org/apache/lens/api/query/InMemoryQueryResult.java
@@ -27,10 +27,7 @@ import java.util.List;
import javax.xml.bind.annotation.XmlElementWrapper;
import javax.xml.bind.annotation.XmlRootElement;
-import lombok.AccessLevel;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
+import lombok.*;
/**
* The Class InMemoryQueryResult.
@@ -48,11 +45,26 @@ import lombok.NoArgsConstructor;
*/
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class InMemoryQueryResult extends QueryResult {
-
+ public static final String DECLARATION = "Result available in memory, attaching here: \n\n";
/**
* The rows.
*/
@XmlElementWrapper
@Getter
- private List<ResultRow> rows = new ArrayList<ResultRow>();
+ private List<ResultRow> rows = new ArrayList<>();
+
+ public String toPrettyString() {
+ StringBuilder b = new StringBuilder();
+ b.append(DECLARATION);
+ int numRows = 0;
+ for (ResultRow row : getRows()) {
+ for (Object col : row.getValues()) {
+ b.append(col).append("\t");
+ }
+ numRows++;
+ b.append("\n");
+ }
+ b.append(numRows).append(" rows ");
+ return b.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-api/src/main/java/org/apache/lens/api/query/LensQuery.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/query/LensQuery.java b/lens-api/src/main/java/org/apache/lens/api/query/LensQuery.java
index af439ff..204ecee 100644
--- a/lens-api/src/main/java/org/apache/lens/api/query/LensQuery.java
+++ b/lens-api/src/main/java/org/apache/lens/api/query/LensQuery.java
@@ -27,10 +27,7 @@ import javax.xml.bind.annotation.XmlRootElement;
import org.apache.lens.api.LensConf;
import org.apache.lens.api.Priority;
-import lombok.AccessLevel;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
+import lombok.*;
/**
* The Class LensQuery.
@@ -81,6 +78,7 @@ import lombok.NoArgsConstructor;
* Instantiates a new lens query.
*/
@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@EqualsAndHashCode
public class LensQuery {
/**
@@ -214,7 +212,7 @@ public class LensQuery {
}
public String getErrorMessage() {
- return (this.status!=null) ? this.status.getLensErrorTOErrorMsg() : null;
+ return (this.status != null) ? this.status.getLensErrorTOErrorMsg() : null;
}
public String getQueryHandleString() {
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-api/src/main/java/org/apache/lens/api/query/PersistentQueryResult.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/query/PersistentQueryResult.java b/lens-api/src/main/java/org/apache/lens/api/query/PersistentQueryResult.java
index a2e10c6..cb72cdc 100644
--- a/lens-api/src/main/java/org/apache/lens/api/query/PersistentQueryResult.java
+++ b/lens-api/src/main/java/org/apache/lens/api/query/PersistentQueryResult.java
@@ -69,4 +69,22 @@ public class PersistentQueryResult extends QueryResult {
@Getter
private Long fileSize;
+ @XmlElement
+ @Getter
+ private String httpResultUrl;
+
+ @Override
+ public String toPrettyString() {
+ StringBuilder sb = new StringBuilder().append("Result available at ").append(persistedURI).append(".");
+ if (numRows != null) {
+ sb.append(" Number of rows: ").append(numRows).append(".");
+ }
+ if (fileSize != null) {
+ sb.append(" File size: ").append(fileSize).append(".");
+ }
+ if (httpResultUrl != null) {
+ sb.append(" Downloadable from ").append(httpResultUrl).append(".");
+ }
+ return sb.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-api/src/main/java/org/apache/lens/api/query/QueryResult.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/query/QueryResult.java b/lens-api/src/main/java/org/apache/lens/api/query/QueryResult.java
index 3ecf0c2..3bc7cc7 100644
--- a/lens-api/src/main/java/org/apache/lens/api/query/QueryResult.java
+++ b/lens-api/src/main/java/org/apache/lens/api/query/QueryResult.java
@@ -24,6 +24,8 @@ package org.apache.lens.api.query;
import javax.xml.bind.annotation.XmlRootElement;
import javax.xml.bind.annotation.XmlSeeAlso;
+import org.apache.lens.api.result.PrettyPrintable;
+
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
@@ -36,5 +38,5 @@ import lombok.NoArgsConstructor;
* Instantiates a new query result.
*/
@NoArgsConstructor(access = AccessLevel.PROTECTED)
-public class QueryResult {
+public abstract class QueryResult implements PrettyPrintable {
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-api/src/main/java/org/apache/lens/api/query/QueryStatus.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/query/QueryStatus.java b/lens-api/src/main/java/org/apache/lens/api/query/QueryStatus.java
index 3c8531f..91cbe39 100644
--- a/lens-api/src/main/java/org/apache/lens/api/query/QueryStatus.java
+++ b/lens-api/src/main/java/org/apache/lens/api/query/QueryStatus.java
@@ -264,6 +264,7 @@ public class QueryStatus implements Serializable {
break;
case EXECUTED:
switch (newState) {
+ case EXECUTED:
case SUCCESSFUL:
case FAILED:
case CANCELED:
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-cli/src/main/java/org/apache/lens/cli/commands/LensQueryCommands.java
----------------------------------------------------------------------
diff --git a/lens-cli/src/main/java/org/apache/lens/cli/commands/LensQueryCommands.java b/lens-cli/src/main/java/org/apache/lens/cli/commands/LensQueryCommands.java
index 006eaed..b6e3fcf 100644
--- a/lens-cli/src/main/java/org/apache/lens/cli/commands/LensQueryCommands.java
+++ b/lens-cli/src/main/java/org/apache/lens/cli/commands/LensQueryCommands.java
@@ -120,14 +120,7 @@ public class LensQueryCommands extends BaseLensCommand {
QueryResult r = rs.getResultSet().getResult();
if (r instanceof InMemoryQueryResult) {
InMemoryQueryResult temp = (InMemoryQueryResult) r;
- for (ResultRow row : temp.getRows()) {
- for (Object col : row.getValues()) {
- b.append(col).append("\t");
- }
- numRows++;
- b.append("\n");
- }
- b.append(numRows + " rows ");
+ b.append(temp.toPrettyString());
} else {
PersistentQueryResult temp = (PersistentQueryResult) r;
b.append("Results of query stored at : ").append(temp.getPersistedURI()).append(" ");
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HiveDriver.java
----------------------------------------------------------------------
diff --git a/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HiveDriver.java b/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HiveDriver.java
index 31c343a..4561ccf 100644
--- a/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HiveDriver.java
+++ b/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HiveDriver.java
@@ -842,7 +842,7 @@ public class HiveDriver implements LensDriver {
log.info("Creating result set for hiveHandle:{}", op);
try {
if (context.isDriverPersistent()) {
- return new HivePersistentResultSet(new Path(context.getHdfsoutPath()), op, getClient());
+ return new HivePersistentResultSet(new Path(context.getDriverResultPath()), op, getClient());
} else if (op.hasResultSet()) {
return new HiveInMemoryResultSet(op, getClient(), closeAfterFetch);
} else {
@@ -874,7 +874,8 @@ public class HiveDriver implements LensDriver {
Path resultSetPath = context.getHDFSResultDir();
// create query
StringBuilder builder = new StringBuilder("INSERT OVERWRITE DIRECTORY ");
- context.setHdfsoutPath(resultSetPath.makeQualified(resultSetPath.getFileSystem(context.getConf())).toString());
+ context.setDriverResultPath(
+ resultSetPath.makeQualified(resultSetPath.getFileSystem(context.getConf())).toString());
builder.append('"').append(resultSetPath).append("\" ");
String outputDirFormat = qdconf.get(LensConfConstants.QUERY_OUTPUT_DIRECTORY_FORMAT);
if (outputDirFormat != null) {
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HivePersistentResultSet.java
----------------------------------------------------------------------
diff --git a/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HivePersistentResultSet.java b/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HivePersistentResultSet.java
index 00e1e53..746e17e 100644
--- a/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HivePersistentResultSet.java
+++ b/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/HivePersistentResultSet.java
@@ -64,7 +64,7 @@ public class HivePersistentResultSet extends PersistentResultSet {
}
@Override
- public Long fileSize() throws LensException {
+ public Long getFileSize() throws LensException {
return null;
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestHiveDriver.java
----------------------------------------------------------------------
diff --git a/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestHiveDriver.java b/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestHiveDriver.java
index 78b3320..2cb3736 100644
--- a/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestHiveDriver.java
+++ b/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestHiveDriver.java
@@ -793,7 +793,7 @@ public class TestHiveDriver {
assertEquals(0, driver.getHiveHandleSize());
HivePersistentResultSet persistentResultSet = (HivePersistentResultSet) resultSet;
String path = persistentResultSet.getOutputPath();
- assertEquals(ctx.getHdfsoutPath(), path);
+ assertEquals(ctx.getDriverResultPath(), path);
driver.closeQuery(plan2.getHandle());
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java b/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
index 096d26e..f202603 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
@@ -774,14 +774,14 @@ public final class LensConfConstants {
// Query Purge Configuration
/**
- * The Constant MAX_NUMBER_OF_FINISHED_QUERY.
+ * The Constant PURGE_INTERVAL.
*/
- public static final String MAX_NUMBER_OF_FINISHED_QUERY = SERVER_PFX + "max.finished.queries";
+ public static final String PURGE_INTERVAL = SERVER_PFX + "querypurger.sleep.interval";
/**
- * The Constant DEFAULT_FINISHED_QUERIES.
+ * The Constant DEFAULT_PURGE_INTERVAL.
*/
- public static final int DEFAULT_FINISHED_QUERIES = 100;
+ public static final int DEFAULT_PURGE_INTERVAL = 10000;
// Server DB configuration
/**
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/driver/InMemoryResultSet.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/InMemoryResultSet.java b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/InMemoryResultSet.java
index 3b76126..c64a3dd 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/InMemoryResultSet.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/InMemoryResultSet.java
@@ -22,16 +22,31 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.lens.api.query.InMemoryQueryResult;
-import org.apache.lens.api.query.QueryResult;
import org.apache.lens.api.query.ResultRow;
import org.apache.lens.server.api.error.LensException;
+import lombok.Setter;
+
/**
* The Class InMemoryResultSet.
*/
public abstract class InMemoryResultSet extends LensResultSet {
public abstract boolean seekToStart() throws LensException;
+
+ @Setter
+ private boolean fullyAccessed = false;
+
+ @Override
+ public boolean canBePurged() {
+ return fullyAccessed;
+ }
+
+ @Override
+ public String getOutputPath() throws LensException {
+ return null;
+ }
+
/**
* Whether there is another result row available.
*
@@ -60,12 +75,15 @@ public abstract class InMemoryResultSet extends LensResultSet {
*
* @see org.apache.lens.server.api.driver.LensResultSet#toQueryResult()
*/
- public QueryResult toQueryResult() throws LensException {
+ public InMemoryQueryResult toQueryResult() throws LensException {
List<ResultRow> rows = new ArrayList<ResultRow>();
while (hasNext()) {
rows.add(next());
}
+ fullyAccessed = true;
return new InMemoryQueryResult(rows);
}
-
+ public boolean isHttpResultAvailable() throws LensException {
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/driver/LensResultSet.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/LensResultSet.java b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/LensResultSet.java
index 929a302..805b0c1 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/LensResultSet.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/LensResultSet.java
@@ -25,6 +25,11 @@ import org.apache.lens.server.api.error.LensException;
* Result set returned by driver.
*/
public abstract class LensResultSet {
+ /**
+ *
+ * @return true if the result can be purged
+ */
+ public abstract boolean canBePurged();
/**
* Get the size of the result set.
@@ -39,8 +44,11 @@ public abstract class LensResultSet {
*
* @return Returns {@link LensResultSetMetadata}
*/
+
public abstract LensResultSetMetadata getMetadata() throws LensException;
+ public abstract String getOutputPath() throws LensException;
+
/**
* Get the corresponding query result object.
*
@@ -49,4 +57,5 @@ public abstract class LensResultSet {
*/
public abstract QueryResult toQueryResult() throws LensException;
+ public abstract boolean isHttpResultAvailable() throws LensException;
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/driver/PersistentResultSet.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/PersistentResultSet.java b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/PersistentResultSet.java
index bbde170..774f1ee 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/PersistentResultSet.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/PersistentResultSet.java
@@ -19,7 +19,6 @@
package org.apache.lens.server.api.driver;
import org.apache.lens.api.query.PersistentQueryResult;
-import org.apache.lens.api.query.QueryResult;
import org.apache.lens.server.api.error.LensException;
/**
@@ -27,22 +26,32 @@ import org.apache.lens.server.api.error.LensException;
*/
public abstract class PersistentResultSet extends LensResultSet {
+ @Override
+ public boolean canBePurged() {
+ return true;
+ }
+
/**
* Get the size of the result set file.
*
* @return The size if available, null if not available.
* @throws LensException the lens exception
*/
- public abstract Long fileSize() throws LensException;
+ public abstract Long getFileSize() throws LensException;
- public abstract String getOutputPath() throws LensException;
+ public String getHttpResultUrl() {
+ return null;
+ }
/*
* (non-Javadoc)
*
* @see org.apache.lens.server.api.driver.LensResultSet#toQueryResult()
*/
- public QueryResult toQueryResult() throws LensException {
- return new PersistentQueryResult(getOutputPath(), size(), fileSize());
+ public PersistentQueryResult toQueryResult() throws LensException {
+ return new PersistentQueryResult(getOutputPath(), size(), getFileSize(), getHttpResultUrl());
+ }
+ public boolean isHttpResultAvailable() throws LensException {
+ return false;
}
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/FinishedLensQuery.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/FinishedLensQuery.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/FinishedLensQuery.java
index d8c04db..7a06c44 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/FinishedLensQuery.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/FinishedLensQuery.java
@@ -24,7 +24,6 @@ import org.apache.lens.api.LensConf;
import org.apache.lens.api.query.QueryHandle;
import org.apache.lens.api.query.QueryStatus;
import org.apache.lens.server.api.driver.LensDriver;
-import org.apache.lens.server.api.error.LensException;
import org.apache.lens.server.api.query.collect.WaitingQueriesSelectionPolicy;
import org.apache.hadoop.conf.Configuration;
@@ -44,7 +43,7 @@ import lombok.ToString;
*
* @see java.lang.Object#hashCode()
*/
-@EqualsAndHashCode
+@EqualsAndHashCode(exclude = "selectedDriver")
/*
* (non-Javadoc)
*
@@ -198,7 +197,7 @@ public class FinishedLensQuery {
this.selectedDriver = ctx.getSelectedDriver();
}
- public QueryContext toQueryContext(Configuration conf, Collection<LensDriver> drivers) throws LensException {
+ public QueryContext toQueryContext(Configuration conf, Collection<LensDriver> drivers) {
QueryContext qctx = new QueryContext(userQuery, submitter, new LensConf(), conf, drivers, null, submissionTime,
false);
qctx.setQueryHandle(QueryHandle.fromString(handle));
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryCancelled.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryCancelled.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryCancelled.java
index bc0465c..a473a47 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryCancelled.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryCancelled.java
@@ -29,6 +29,7 @@ public class QueryCancelled extends QueryEnded {
/**
* Instantiates a new query cancelled.
*
+ * @param ctx the query context
* @param eventTime the event time
* @param prev the prev
* @param current the current
@@ -36,10 +37,15 @@ public class QueryCancelled extends QueryEnded {
* @param user the user
* @param cause the cause
*/
- public QueryCancelled(long eventTime, QueryStatus.Status prev, QueryStatus.Status current, QueryHandle handle,
+ public QueryCancelled(QueryContext ctx, long eventTime, QueryStatus.Status prev, QueryStatus.Status current,
+ QueryHandle handle,
String user, String cause) {
- super(eventTime, prev, current, handle, user, cause);
+ super(ctx, eventTime, prev, current, handle, user, cause);
checkCurrentState(QueryStatus.Status.CANCELED);
}
+ public QueryCancelled(QueryContext ctx, QueryStatus.Status prevState, QueryStatus.Status currState, String cause) {
+ // TODO: correct username. put who cancelled it, not the submitter. Similar for others
+ this(ctx, ctx.getEndTime(), prevState, currState, ctx.getQueryHandle(), ctx.getSubmittedUser(), cause);
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryClosed.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryClosed.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryClosed.java
index ea8f70f..3837087 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryClosed.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryClosed.java
@@ -29,6 +29,7 @@ public class QueryClosed extends QueryEnded {
/**
* Instantiates a new query closed.
*
+ * @param ctx the query context
* @param eventTime the event time
* @param prev the prev
* @param current the current
@@ -36,9 +37,14 @@ public class QueryClosed extends QueryEnded {
* @param user the user
* @param cause the cause
*/
- public QueryClosed(long eventTime, QueryStatus.Status prev, QueryStatus.Status current, QueryHandle handle,
+ public QueryClosed(QueryContext ctx, long eventTime, QueryStatus.Status prev, QueryStatus.Status current,
+ QueryHandle handle,
String user, String cause) {
- super(eventTime, prev, current, handle, user, cause);
+ super(ctx, eventTime, prev, current, handle, user, cause);
checkCurrentState(QueryStatus.Status.CLOSED);
}
+
+ public QueryClosed(QueryContext ctx, QueryStatus.Status prevState, QueryStatus.Status currState, String cause) {
+ this(ctx, ctx.getClosedTime(), prevState, currState, ctx.getQueryHandle(), ctx.getSubmittedUser(), cause);
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryContext.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryContext.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryContext.java
index 12de0a5..bed79ac 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryContext.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryContext.java
@@ -42,10 +42,12 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
+import lombok.ToString;
/**
* The Class QueryContext.
*/
+@ToString
public class QueryContext extends AbstractQueryContext {
/**
@@ -95,7 +97,7 @@ public class QueryContext extends AbstractQueryContext {
*/
@Getter
@Setter
- private String hdfsoutPath;
+ private String driverResultPath;
/**
* The submission time.
@@ -326,7 +328,7 @@ public class QueryContext extends AbstractQueryContext {
/*
* Introduced for Recovering finished query.
*/
- public void setStatusSkippingTransitionTest(final QueryStatus newStatus) throws LensException {
+ public void setStatusSkippingTransitionTest(final QueryStatus newStatus) {
this.status = newStatus;
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryEnded.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryEnded.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryEnded.java
index 3e9474c..e80da6d 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryEnded.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryEnded.java
@@ -33,6 +33,8 @@ import lombok.Getter;
*/
public class QueryEnded extends StatusChange {
+ @Getter
+ private final QueryContext queryContext;
/**
* The user.
*/
@@ -54,6 +56,7 @@ public class QueryEnded extends StatusChange {
/**
* Instantiates a new query ended.
*
+ * @param ctx
* @param eventTime the event time
* @param prev the prev
* @param current the current
@@ -61,9 +64,10 @@ public class QueryEnded extends StatusChange {
* @param user the user
* @param cause the cause
*/
- public QueryEnded(long eventTime, QueryStatus.Status prev, QueryStatus.Status current, QueryHandle handle,
- String user, String cause) {
+ public QueryEnded(QueryContext ctx, long eventTime, QueryStatus.Status prev, QueryStatus.Status current,
+ QueryHandle handle, String user, String cause) {
super(eventTime, prev, current, handle);
+ this.queryContext = ctx;
this.user = user;
this.cause = cause;
if (!END_STATES.contains(current)) {
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryFailed.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryFailed.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryFailed.java
index 40a1c0f..bdffbc2 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryFailed.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QueryFailed.java
@@ -29,6 +29,7 @@ public class QueryFailed extends QueryEnded {
/**
* Instantiates a new query failed.
*
+ * @param ctx the query context
* @param eventTime the event time
* @param prev the prev
* @param current the current
@@ -36,9 +37,14 @@ public class QueryFailed extends QueryEnded {
* @param user the user
* @param cause the cause
*/
- public QueryFailed(long eventTime, QueryStatus.Status prev, QueryStatus.Status current, QueryHandle handle,
+ public QueryFailed(QueryContext ctx, long eventTime, QueryStatus.Status prev, QueryStatus.Status current,
+ QueryHandle handle,
String user, String cause) {
- super(eventTime, prev, current, handle, user, cause);
+ super(ctx, eventTime, prev, current, handle, user, cause);
checkCurrentState(QueryStatus.Status.FAILED);
}
+
+ public QueryFailed(QueryContext ctx, QueryStatus.Status prevState, QueryStatus.Status currState, String cause) {
+ this(ctx, ctx.getEndTime(), prevState, currState, ctx.getQueryHandle(), ctx.getSubmittedUser(), cause);
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/main/java/org/apache/lens/server/api/query/QuerySuccess.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QuerySuccess.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QuerySuccess.java
index c551dba..298fdbb 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/QuerySuccess.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/QuerySuccess.java
@@ -29,13 +29,19 @@ public class QuerySuccess extends QueryEnded {
/**
* Instantiates a new query success.
*
+ * @param ctx the query context
* @param eventTime the event time
* @param prev the prev
* @param current the current
* @param handle the handle
*/
- public QuerySuccess(long eventTime, QueryStatus.Status prev, QueryStatus.Status current, QueryHandle handle) {
- super(eventTime, prev, current, handle, null, null);
+ public QuerySuccess(QueryContext ctx, long eventTime, QueryStatus.Status prev, QueryStatus.Status current,
+ QueryHandle handle) {
+ super(ctx, eventTime, prev, current, handle, null, null);
checkCurrentState(QueryStatus.Status.SUCCESSFUL);
}
+
+ public QuerySuccess(QueryContext ctx, QueryStatus.Status prevState, QueryStatus.Status currState) {
+ this(ctx, ctx.getEndTime(), prevState, currState, ctx.getQueryHandle());
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server-api/src/test/java/org/apache/lens/server/api/driver/MockDriver.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/test/java/org/apache/lens/server/api/driver/MockDriver.java b/lens-server-api/src/test/java/org/apache/lens/server/api/driver/MockDriver.java
index b28669e..2d86589 100644
--- a/lens-server-api/src/test/java/org/apache/lens/server/api/driver/MockDriver.java
+++ b/lens-server-api/src/test/java/org/apache/lens/server/api/driver/MockDriver.java
@@ -255,7 +255,7 @@ public class MockDriver implements LensDriver {
}
@Override
- public Long fileSize() throws LensException {
+ public Long getFileSize() throws LensException {
// TODO Auto-generated method stub
return null;
}
@@ -297,7 +297,7 @@ public class MockDriver implements LensDriver {
* @see org.apache.lens.server.api.driver.LensDriver#fetchResultSet(org.apache.lens.server.api.query.QueryContext)
*/
@Override
- public LensResultSet fetchResultSet(QueryContext context) throws LensException {
+ public LensResultSet fetchResultSet(final QueryContext context) throws LensException {
return new InMemoryResultSet() {
@Override
@@ -340,6 +340,11 @@ public class MockDriver implements LensDriver {
// TODO Auto-generated method stub
return false;
}
+
+ @Override
+ public boolean canBePurged() {
+ return true;
+ }
};
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/main/java/org/apache/lens/server/query/LensPersistentResult.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/LensPersistentResult.java b/lens-server/src/main/java/org/apache/lens/server/query/LensPersistentResult.java
index b65a5f4..1e9a182 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/LensPersistentResult.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/LensPersistentResult.java
@@ -18,13 +18,27 @@
*/
package org.apache.lens.server.query;
+import java.io.IOException;
+
+import org.apache.lens.api.query.QueryHandle;
+import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.driver.LensResultSetMetadata;
import org.apache.lens.server.api.driver.PersistentResultSet;
import org.apache.lens.server.api.error.LensException;
+import org.apache.lens.server.api.query.FinishedLensQuery;
+import org.apache.lens.server.api.query.QueryContext;
-/**
- * The Class LensPersistentResult.
- */
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.codehaus.jackson.map.ObjectMapper;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+/** The Class LensPersistentResult. */
+@Slf4j
public class LensPersistentResult extends PersistentResultSet {
/** The metadata. */
@@ -38,19 +52,45 @@ public class LensPersistentResult extends PersistentResultSet {
/** The file size. */
private final Long fileSize;
+ private final Configuration conf;
+ @Getter
+ private String httpResultUrl = null;
/**
* Instantiates a new lens persistent result.
- *
- * @param metadata the metadata
- * @param outputPath the output path
- * @param numRows the num rows
+ * @param queryHandle the query handle
+ * @param metadata the metadata
+ * @param outputPath the output path
+ * @param numRows the num rows
+ * @param conf the lens server conf
*/
- public LensPersistentResult(LensResultSetMetadata metadata, String outputPath, Integer numRows, Long fileSize) {
+ public LensPersistentResult(QueryHandle queryHandle, LensResultSetMetadata metadata, String outputPath, Integer
+ numRows, Long fileSize,
+ Configuration conf) {
this.metadata = metadata;
this.outputPath = outputPath;
this.numRows = numRows;
this.fileSize = fileSize;
+ this.conf = conf;
+ if (isHttpResultAvailable()) {
+ this.httpResultUrl = conf.get(LensConfConstants.SERVER_BASE_URL, LensConfConstants.DEFAULT_SERVER_BASE_URL)
+ + "queryapi/queries/" + queryHandle + "/httpresultset";
+ }
+ }
+
+ public LensPersistentResult(QueryContext ctx, Configuration conf) {
+ this(ctx.getQueryHandle(),
+ ctx.getQueryOutputFormatter().getMetadata(),
+ ctx.getQueryOutputFormatter().getFinalOutputPath(),
+ ctx.getQueryOutputFormatter().getNumRows(),
+ ctx.getQueryOutputFormatter().getFileSize(), conf);
+ }
+
+ public LensPersistentResult(FinishedLensQuery query, Configuration conf, ObjectMapper mapper) throws
+ ClassNotFoundException, IOException {
+ this(QueryHandle.fromString(query.getHandle()),
+ mapper.readValue(query.getMetadata(), (Class<LensResultSetMetadata>) Class.forName(query.getMetadataClass())),
+ query.getResult(), query.getRows(), query.getFileSize(), conf);
}
@Override
@@ -69,7 +109,7 @@ public class LensPersistentResult extends PersistentResultSet {
}
@Override
- public Long fileSize() throws LensException {
+ public Long getFileSize() throws LensException {
return fileSize;
}
@@ -77,4 +117,19 @@ public class LensPersistentResult extends PersistentResultSet {
public LensResultSetMetadata getMetadata() throws LensException {
return metadata;
}
+
+ @Override
+ public boolean isHttpResultAvailable() {
+ try {
+ final Path resultPath = new Path(getOutputPath());
+ FileSystem fs = resultPath.getFileSystem(conf);
+ if (fs.isDirectory(resultPath)) {
+ return false;
+ }
+ } catch (IOException | LensException e) {
+ log.warn("Unable to get status for Result Directory", e);
+ return false;
+ }
+ return true;
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/main/java/org/apache/lens/server/query/QueryEndNotifier.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/QueryEndNotifier.java b/lens-server/src/main/java/org/apache/lens/server/query/QueryEndNotifier.java
index 45ba7ac..110624a 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/QueryEndNotifier.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/QueryEndNotifier.java
@@ -18,6 +18,8 @@
*/
package org.apache.lens.server.query;
+import static org.apache.lens.server.api.LensConfConstants.*;
+
import java.util.Date;
import java.util.Properties;
@@ -32,7 +34,7 @@ import javax.mail.internet.MimeMultipart;
import org.apache.lens.api.query.QueryStatus;
import org.apache.lens.server.LensServices;
-import org.apache.lens.server.api.LensConfConstants;
+import org.apache.lens.server.api.error.LensException;
import org.apache.lens.server.api.events.AsyncEventListener;
import org.apache.lens.server.api.metrics.MetricsService;
import org.apache.lens.server.api.query.QueryContext;
@@ -58,9 +60,6 @@ public class QueryEndNotifier extends AsyncEventListener<QueryEnded> {
/** The Constant EMAIL_ERROR_COUNTER. */
public static final String EMAIL_ERROR_COUNTER = "email-send-errors";
- /** The conf. */
- private final HiveConf conf;
-
/** The from. */
private final String from;
@@ -78,37 +77,33 @@ public class QueryEndNotifier extends AsyncEventListener<QueryEnded> {
private final LogSegregationContext logSegregationContext;
- /**
- * Instantiates a new query end notifier.
+ /** Instantiates a new query end notifier.
*
* @param queryService the query service
- * @param hiveConf the hive conf
- */
+ * @param hiveConf the hive conf */
public QueryEndNotifier(QueryExecutionServiceImpl queryService, HiveConf hiveConf,
- @NonNull final LogSegregationContext logSegregationContext) {
+ @NonNull final LogSegregationContext logSegregationContext) {
this.queryService = queryService;
- this.conf = hiveConf;
- from = conf.get(LensConfConstants.MAIL_FROM_ADDRESS);
- host = conf.get(LensConfConstants.MAIL_HOST);
- port = conf.get(LensConfConstants.MAIL_PORT);
- mailSmtpTimeout = Integer.parseInt(
- conf.get(LensConfConstants.MAIL_SMTP_TIMEOUT, LensConfConstants.MAIL_DEFAULT_SMTP_TIMEOUT));
- mailSmtpConnectionTimeout = Integer.parseInt(conf.get(LensConfConstants.MAIL_SMTP_CONNECTIONTIMEOUT,
- LensConfConstants.MAIL_DEFAULT_SMTP_CONNECTIONTIMEOUT));
+ HiveConf conf = hiveConf;
+ from = conf.get(MAIL_FROM_ADDRESS);
+ host = conf.get(MAIL_HOST);
+ port = conf.get(MAIL_PORT);
+ mailSmtpTimeout = Integer.parseInt(conf.get(MAIL_SMTP_TIMEOUT, MAIL_DEFAULT_SMTP_TIMEOUT));
+ mailSmtpConnectionTimeout = Integer.parseInt(conf.get(MAIL_SMTP_CONNECTIONTIMEOUT,
+ MAIL_DEFAULT_SMTP_CONNECTIONTIMEOUT));
this.logSegregationContext = logSegregationContext;
}
/*
* (non-Javadoc)
*
- * @see org.apache.lens.server.api.events.AsyncEventListener#process(org.apache.lens.server.api.events.LensEvent)
- */
+ * @see org.apache.lens.server.api.events.AsyncEventListener#process(org.apache.lens.server.api.events.LensEvent) */
@Override
- public void process(QueryEnded event) {
+ public void process(final QueryEnded event) {
if (event.getCurrentValue() == QueryStatus.Status.CLOSED) {
return;
}
- QueryContext queryContext = queryService.getQueryContext(event.getQueryHandle());
+ QueryContext queryContext = event.getQueryContext();
if (queryContext == null) {
log.warn("Could not find the context for {} for event:{}. No email generated", event.getQueryHandle(),
event.getCurrentValue());
@@ -116,45 +111,36 @@ public class QueryEndNotifier extends AsyncEventListener<QueryEnded> {
}
this.logSegregationContext.setLogSegragationAndQueryId(queryContext.getQueryHandleString());
- boolean whetherMailNotify = Boolean.parseBoolean(queryContext.getConf().get(LensConfConstants.QUERY_MAIL_NOTIFY,
- LensConfConstants.WHETHER_MAIL_NOTIFY_DEFAULT));
+ boolean whetherMailNotify = Boolean.parseBoolean(queryContext.getConf().get(QUERY_MAIL_NOTIFY,
+ WHETHER_MAIL_NOTIFY_DEFAULT));
if (!whetherMailNotify) {
return;
}
String queryName = queryContext.getQueryName();
- queryName = queryName == null ? "" : queryName;
- String mailSubject = "Query " + queryName + " " + queryContext.getStatus().getStatus() + ": "
- + event.getQueryHandle();
+ String mailSubject = "Query " + (StringUtils.isBlank(queryName) ? "" : (queryName + " "))
+ + queryContext.getStatus().getStatus() + ": " + event.getQueryHandle();
String mailMessage = createMailMessage(queryContext);
String to = queryContext.getSubmittedUser() + "@" + queryService.getServerDomain();
- String cc = queryContext.getConf().get(LensConfConstants.QUERY_RESULT_EMAIL_CC,
- LensConfConstants.QUERY_RESULT_DEFAULT_EMAIL_CC);
+ String cc = queryContext.getConf().get(QUERY_RESULT_EMAIL_CC, QUERY_RESULT_DEFAULT_EMAIL_CC);
log.info("Sending completion email for query handle: {}", event.getQueryHandle());
sendMail(host, port, new Email(from, to, cc, mailSubject, mailMessage), mailSmtpTimeout, mailSmtpConnectionTimeout);
}
- /**
- * Creates the mail message.
+ /** Creates the mail message.
*
* @param queryContext the query context
- * @return the string
- */
+ * @return the string */
private String createMailMessage(QueryContext queryContext) {
StringBuilder msgBuilder = new StringBuilder();
switch (queryContext.getStatus().getStatus()) {
case SUCCESSFUL:
- msgBuilder.append("Result available at ");
- String baseURI = conf.get(LensConfConstants.SERVER_BASE_URL, LensConfConstants.DEFAULT_SERVER_BASE_URL);
- msgBuilder.append(baseURI);
- msgBuilder.append("queryapi/queries/");
- msgBuilder.append(queryContext.getQueryHandle());
- msgBuilder.append("/httpresultset");
+ msgBuilder.append(getResultMessage(queryContext));
break;
case FAILED:
msgBuilder.append(queryContext.getStatus().getStatusMessage());
@@ -166,13 +152,21 @@ public class QueryEndNotifier extends AsyncEventListener<QueryEnded> {
case CANCELED:
msgBuilder.append(queryContext.getStatus().getStatusMessage());
break;
- case CLOSED:
default:
break;
}
return msgBuilder.toString();
}
+ private String getResultMessage(QueryContext queryContext) {
+ try {
+ return queryService.getResultset(queryContext.getQueryHandle()).toQueryResult().toPrettyString();
+ } catch (LensException e) {
+ log.error("Error retrieving result of query handle {} for sending e-mail", queryContext.getQueryHandle(), e);
+ return "Error retrieving result.";
+ }
+ }
+
@Data
public static class Email {
private final String from;
@@ -181,15 +175,14 @@ public class QueryEndNotifier extends AsyncEventListener<QueryEnded> {
private final String subject;
private final String message;
}
- /**
- * Send mail.
+
+ /** Send mail.
*
* @param host the host
* @param port the port
* @param email the email
* @param mailSmtpTimeout the mail smtp timeout
- * @param mailSmtpConnectionTimeout the mail smtp connection timeout
- */
+ * @param mailSmtpConnectionTimeout the mail smtp connection timeout */
public static void sendMail(String host, String port,
Email email, int mailSmtpTimeout, int mailSmtpConnectionTimeout) {
Properties props = System.getProperties();
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java b/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
index 9e27dd4..2ba54e0 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionServiceImpl.java
@@ -88,10 +88,12 @@ import org.codehaus.jackson.map.module.SimpleModule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import lombok.Getter;
import lombok.NonNull;
+import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
/**
@@ -159,18 +161,17 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
/**
* The finished queries.
*/
- private DelayQueue<FinishedQuery> finishedQueries = new DelayQueue<FinishedQuery>();
+ ConcurrentLinkedQueue<FinishedQuery> finishedQueries = new ConcurrentLinkedQueue<>();
/**
* The prepared query queue.
*/
- private DelayQueue<PreparedQueryContext> preparedQueryQueue = new DelayQueue<PreparedQueryContext>();
+ private DelayQueue<PreparedQueryContext> preparedQueryQueue = new DelayQueue<>();
/**
* The prepared queries.
*/
- private Map<QueryPrepareHandle, PreparedQueryContext> preparedQueries
- = new HashMap<QueryPrepareHandle, PreparedQueryContext>();
+ private Map<QueryPrepareHandle, PreparedQueryContext> preparedQueries = new HashMap<>();
/**
* The all queries.
@@ -180,7 +181,8 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
/**
* The conf.
*/
- private Configuration conf;
+ @VisibleForTesting
+ Configuration conf;
/**
* The query submitter runnable.
@@ -240,7 +242,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
/**
* The max finished queries.
*/
- private int maxFinishedQueries;
+ int purgeInterval;
/**
* The lens server dao.
@@ -328,7 +330,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
// Add result formatter
getEventService().addListenerForType(new ResultFormatter(this, this.logSegregationContext), QueryExecuted.class);
- getEventService().addListenerForType(new QueryExecutionStatisticsGenerator(this, getEventService()),
+ getEventService().addListenerForType(new QueryExecutionStatisticsGenerator(getEventService()),
QueryEnded.class);
getEventService().addListenerForType(
new QueryEndNotifier(this, getCliService().getHiveConf(), this.logSegregationContext), QueryEnded.class);
@@ -435,11 +437,13 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
/**
* The Class FinishedQuery.
*/
- private class FinishedQuery implements Delayed {
+ @ToString
+ public class FinishedQuery {
/**
* The ctx.
*/
+ @Getter
private final QueryContext ctx;
/**
@@ -462,29 +466,25 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
}
- /*
- * (non-Javadoc)
- *
- * @see java.lang.Comparable#compareTo(java.lang.Object)
- */
- @Override
- public int compareTo(Delayed o) {
- return (int) (this.finishTime.getTime() - ((FinishedQuery) o).finishTime.getTime());
+ public boolean canBePurged() {
+ try {
+ if (getCtx().getStatus().getStatus().equals(SUCCESSFUL)) {
+ if (getCtx().getStatus().isResultSetAvailable()) {
+ LensResultSet rs = getResultset();
+ log.info("Resultset for {} is {}", getQueryHandle(), rs);
+ return rs.canBePurged();
+ }
+ }
+ return true;
+ } catch (Throwable e) {
+ log.error("Error while accessing result set for query handle while purging: {}."
+ + " Hence, going ahead with purge", getQueryHandle(), e);
+ return true;
+ }
}
- /*
- * (non-Javadoc)
- *
- * @see java.util.concurrent.Delayed#getDelay(java.util.concurrent.TimeUnit)
- */
- @Override
- public long getDelay(TimeUnit units) {
- int size = finishedQueries.size();
- if (size > maxFinishedQueries) {
- return 0;
- } else {
- return Integer.MAX_VALUE;
- }
+ private LensResultSet getResultset() throws LensException {
+ return QueryExecutionServiceImpl.this.getResultset(getQueryHandle());
}
/**
@@ -494,16 +494,13 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
return finishTime;
}
- /**
- * @return the ctx
- */
- public QueryContext getCtx() {
- return ctx;
- }
-
public String getQueryHandleString() {
return ctx.getQueryHandleString();
}
+
+ public QueryHandle getQueryHandle() {
+ return ctx.getQueryHandle();
+ }
}
/**
@@ -523,8 +520,8 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
private final QueryLaunchingConstraintsChecker constraintsChecker;
public QuerySubmitter(@NonNull final ErrorCollection errorCollection,
- @NonNull final EstimatedQueryCollection waitingQueries,
- @NonNull final QueryLaunchingConstraintsChecker constraintsChecker) {
+ @NonNull final EstimatedQueryCollection waitingQueries,
+ @NonNull final QueryLaunchingConstraintsChecker constraintsChecker) {
this.errorCollection = errorCollection;
this.waitingQueries = waitingQueries;
@@ -639,9 +636,9 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
private void checkEstimatedQueriesState(final QueryContext query) throws LensException {
- if (query.getSelectedDriver() == null || query.getSelectedDriverQueryCost() == null) {
- throw new LensException("selected driver: " + query.getSelectedDriver() +" OR selected driver query cost: "
- + query.getSelectedDriverQueryCost() + " is null. Query doesn't appear to be an estimated query.");
+ if (query.getSelectedDriver() == null || query.getSelectedDriverQueryCost() == null) {
+ throw new LensException("selected driver: " + query.getSelectedDriver() + " OR selected driver query cost: "
+ + query.getSelectedDriverQueryCost() + " is null. Query doesn't appear to be an estimated query.");
}
}
}
@@ -789,10 +786,9 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
ctx.setStatus(ctx.getDriverStatus().toQueryStatus());
} catch (LensException exc) {
// Driver gave exception while updating status
-
setFailedStatus(ctx, "Status update failed", exc.getMessage(), exc.buildLensErrorTO(this.errorCollection));
log.error("Status update failed for {}", handle, exc);
-
+ return;
}
// query is successfully executed by driver and
// if query result need not be persisted or there is no result available in driver, move the query to
@@ -832,10 +828,9 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
QueryHandle query = ctx.getQueryHandle();
switch (currState) {
case CANCELED:
- // TODO: correct username. put who cancelled it, not the submitter. Similar for others
- return new QueryCancelled(ctx.getEndTime(), prevState, currState, query, ctx.getSubmittedUser(), null);
+ return new QueryCancelled(ctx, prevState, currState, null);
case CLOSED:
- return new QueryClosed(ctx.getClosedTime(), prevState, currState, query, ctx.getSubmittedUser(), null);
+ return new QueryClosed(ctx, prevState, currState, null);
case FAILED:
StringBuilder msgBuilder = new StringBuilder();
msgBuilder.append(ctx.getStatus().getStatusMessage());
@@ -843,8 +838,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
msgBuilder.append("\n Reason:\n");
msgBuilder.append(ctx.getStatus().getErrorMessage());
}
- return new QueryFailed(ctx.getEndTime(), prevState, currState, query, ctx.getSubmittedUser(),
- msgBuilder.toString());
+ return new QueryFailed(ctx, prevState, currState, msgBuilder.toString());
case LAUNCHED:
return new QueryLaunched(ctx.getLaunchTime(), prevState, currState, query);
case QUEUED:
@@ -855,7 +849,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
case EXECUTED:
return new QueryExecuted(ctx.getDriverStatus().getDriverFinishTime(), prevState, currState, query);
case SUCCESSFUL:
- return new QuerySuccess(ctx.getEndTime(), prevState, currState, query);
+ return new QuerySuccess(ctx, prevState, currState);
default:
log.warn("Query {} transitioned to {} state from {} state", query, currState, prevState);
return null;
@@ -905,64 +899,67 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
public void run() {
log.info("Starting Query purger thread");
while (!stopped && !queryPurger.isInterrupted()) {
- FinishedQuery finished = null;
try {
- finished = finishedQueries.take();
- logSegregationContext.setLogSegragationAndQueryId(finished.getQueryHandleString());
- } catch (InterruptedException e) {
- log.info("QueryPurger has been interrupted, exiting");
- return;
- }
- try {
- FinishedLensQuery finishedQuery = new FinishedLensQuery(finished.getCtx());
- if (finished.ctx.getStatus().getStatus() == SUCCESSFUL) {
- if (finished.ctx.getStatus().isResultSetAvailable()) {
- LensResultSet set = getResultset(finished.getCtx().getQueryHandle());
- if (set != null && PersistentResultSet.class.isAssignableFrom(set.getClass())) {
- LensResultSetMetadata metadata = set.getMetadata();
- String outputPath = ((PersistentResultSet) set).getOutputPath();
- Long fileSize = ((PersistentResultSet) set).fileSize();
- Integer rows = set.size();
- finishedQuery.setMetadataClass(metadata.getClass().getName());
- finishedQuery.setResult(outputPath);
- finishedQuery.setMetadata(MAPPER.writeValueAsString(metadata));
- finishedQuery.setRows(rows);
- finishedQuery.setFileSize(fileSize);
+ Iterator<FinishedQuery> iter = finishedQueries.iterator();
+ FinishedQuery finished;
+ while (iter.hasNext()) {
+ finished = iter.next();
+ if (finished.canBePurged()) {
+ try {
+ FinishedLensQuery finishedQuery = new FinishedLensQuery(finished.getCtx());
+ if (finished.ctx.getStatus().getStatus() == SUCCESSFUL) {
+ if (finished.ctx.getStatus().isResultSetAvailable()) {
+ try {
+ LensResultSet set = finished.getResultset();
+ if (set != null && PersistentResultSet.class.isAssignableFrom(set.getClass())) {
+ LensResultSetMetadata metadata = set.getMetadata();
+ String outputPath = set.getOutputPath();
+ Long fileSize = ((PersistentResultSet) set).getFileSize();
+ Integer rows = set.size();
+ finishedQuery.setMetadataClass(metadata.getClass().getName());
+ finishedQuery.setResult(outputPath);
+ finishedQuery.setMetadata(MAPPER.writeValueAsString(metadata));
+ finishedQuery.setRows(rows);
+ finishedQuery.setFileSize(fileSize);
+ }
+ } catch (Exception e) {
+ log.error("Couldn't obtain result set info for the query: {}. Going ahead with purge",
+ finished.getQueryHandle(), e);
+ }
+ }
+ }
+ lensServerDao.insertFinishedQuery(finishedQuery);
+ log.info("Saved query {} to DB", finishedQuery.getHandle());
+ iter.remove();
+ } catch (Exception e) {
+ log.warn("Exception while purging query {}", finished.getQueryHandle(), e);
+ continue;
}
- }
- }
- try {
- lensServerDao.insertFinishedQuery(finishedQuery);
- log.info("Saved query {} to DB", finishedQuery.getHandle());
- } catch (Exception e) {
- log.warn("Exception while purging query ", e);
- finishedQueries.add(finished);
- continue;
- }
-
- synchronized (finished.ctx) {
- finished.ctx.setFinishedQueryPersisted(true);
- try {
- if (finished.getCtx().getSelectedDriver() != null) {
- finished.getCtx().getSelectedDriver().closeQuery(finished.getCtx().getQueryHandle());
+ synchronized (finished.ctx) {
+ finished.ctx.setFinishedQueryPersisted(true);
+ try {
+ if (finished.getCtx().getSelectedDriver() != null) {
+ finished.getCtx().getSelectedDriver().closeQuery(finished.getQueryHandle());
+ }
+ } catch (Exception e) {
+ log.warn("Exception while closing query with selected driver.", e);
+ }
+ log.info("Purging: {}", finished.getQueryHandle());
+ allQueries.remove(finished.getQueryHandle());
+ resultSets.remove(finished.getQueryHandle());
}
- } catch (Exception e) {
- log.warn("Exception while closing query with selected driver.", e);
+ fireStatusChangeEvent(finished.getCtx(),
+ new QueryStatus(1f, null, CLOSED, "Query purged", false, null, null, null), finished.getCtx()
+ .getStatus());
+ log.info("Query purged: {}", finished.getQueryHandle());
}
- log.info("Purging: {}", finished.getCtx().getQueryHandle());
- allQueries.remove(finished.getCtx().getQueryHandle());
- resultSets.remove(finished.getCtx().getQueryHandle());
}
- fireStatusChangeEvent(finished.getCtx(),
- new QueryStatus(1f, null, CLOSED, "Query purged", false, null, null, null), finished.getCtx().getStatus());
- log.info("Query purged: {}", finished.getCtx().getQueryHandle());
-
- } catch (LensException e) {
- incrCounter(QUERY_PURGER_COUNTER);
- log.error("Error closing query ", e);
- } catch (Exception e) {
+ Thread.sleep(purgeInterval);
+ } catch (InterruptedException e) {
+ log.error("purger interrupted", e);
+ } catch (Throwable e) {
+ log.error("Purger giving error", e);
incrCounter(QUERY_PURGER_COUNTER);
- log.error("Error in query purger", e);
}
}
log.info("QueryPurger exited");
@@ -1018,16 +1015,16 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
new DefaultQueryCollection(new TreeSet<QueryContext>(new QueryContextPriorityComparator()))));
ImmutableSet<QueryLaunchingConstraint> queryConstraints = getImplementations(
- QUERY_LAUNCHING_CONSTRAINT_FACTORIES_KEY, hiveConf);
+ QUERY_LAUNCHING_CONSTRAINT_FACTORIES_KEY, hiveConf);
this.queryConstraintsChecker = new DefaultQueryLaunchingConstraintsChecker(queryConstraints);
this.querySubmitterRunnable = new QuerySubmitter(LensServices.get().getErrorCollection(), this.waitingQueries,
- this.queryConstraintsChecker);
+ this.queryConstraintsChecker);
this.querySubmitter = new Thread(querySubmitterRunnable, "QuerySubmitter");
ImmutableSet<WaitingQueriesSelectionPolicy> selectionPolicies = getImplementations(
- WAITING_QUERIES_SELECTION_POLICY_FACTORIES_KEY, hiveConf);
+ WAITING_QUERIES_SELECTION_POLICY_FACTORIES_KEY, hiveConf);
this.waitingQueriesSelector = new IntersectingWaitingQueriesSelector(selectionPolicies);
@@ -1048,8 +1045,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
log.error("Error while loading drivers", e);
throw new IllegalStateException("Could not load drivers", e);
}
- maxFinishedQueries = conf.getInt(MAX_NUMBER_OF_FINISHED_QUERY,
- DEFAULT_FINISHED_QUERIES);
+ purgeInterval = conf.getInt(PURGE_INTERVAL, DEFAULT_PURGE_INTERVAL);
initalizeFinishedQueryStore(conf);
log.info("Query execution service initialized");
}
@@ -1207,7 +1203,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
*/
private void rewriteAndSelect(final AbstractQueryContext ctx) throws LensException {
MethodMetricsContext parallelCallGauge = MethodMetricsFactory.createMethodGauge(ctx.getConf(), false,
- PARALLEL_CALL_GAUGE);
+ PARALLEL_CALL_GAUGE);
try {
userQueryToCubeQueryRewriter.rewrite(ctx);
// Initially we obtain individual runnables for rewrite and estimate calls
@@ -1293,7 +1289,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
MethodMetricsContext selectGauge = MethodMetricsFactory.createMethodGauge(ctx.getConf(), false,
- DRIVER_SELECTOR_GAUGE);
+ DRIVER_SELECTOR_GAUGE);
// 2. select driver to run the query
LensDriver driver = driverSelector.select(ctx, conf);
ctx.setSelectedDriver(driver);
@@ -1420,16 +1416,14 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
* @return the resultset from dao
* @throws LensException the lens exception
*/
- private LensResultSet getResultsetFromDAO(QueryHandle queryHandle) throws LensException {
+ private LensPersistentResult getResultsetFromDAO(QueryHandle queryHandle) throws LensException {
FinishedLensQuery query = lensServerDao.getQuery(queryHandle.toString());
if (query != null) {
if (query.getResult() == null) {
throw new NotFoundException("InMemory Query result purged " + queryHandle);
}
try {
- Class<LensResultSetMetadata> mdKlass = (Class<LensResultSetMetadata>) Class.forName(query.getMetadataClass());
- return new LensPersistentResult(MAPPER.readValue(query.getMetadata(), mdKlass), query.getResult(),
- query.getRows(), query.getFileSize());
+ return new LensPersistentResult(query, conf, MAPPER);
} catch (Exception e) {
throw new LensException(e);
}
@@ -1444,7 +1438,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
* @return the resultset
* @throws LensException the lens exception
*/
- private LensResultSet getResultset(QueryHandle queryHandle) throws LensException {
+ LensResultSet getResultset(QueryHandle queryHandle) throws LensException {
QueryContext ctx = allQueries.get(queryHandle);
if (ctx == null) {
return getResultsetFromDAO(queryHandle);
@@ -1456,15 +1450,9 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
LensResultSet resultSet = resultSets.get(queryHandle);
if (resultSet == null) {
if (ctx.isPersistent() && ctx.getQueryOutputFormatter() != null) {
- resultSets
- .put(queryHandle,
- new LensPersistentResult(
- ctx.getQueryOutputFormatter().getMetadata(),
- ctx.getQueryOutputFormatter().getFinalOutputPath(),
- ctx.getQueryOutputFormatter().getNumRows(),
- ctx.getQueryOutputFormatter().getFileSize()));
+ resultSets.put(queryHandle, new LensPersistentResult(ctx, conf));
} else if (allQueries.get(queryHandle).isResultAvailableInDriver()) {
- resultSet = allQueries.get(queryHandle).getSelectedDriver().fetchResultSet(allQueries.get(queryHandle));
+ resultSet = getDriverResultset(queryHandle);
resultSets.put(queryHandle, resultSet);
} else {
throw new NotFoundException("Result set not available for query:" + queryHandle);
@@ -1585,7 +1573,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
public QueryHandle executePrepareAsync(LensSessionHandle sessionHandle, QueryPrepareHandle prepareHandle,
LensConf conf, String queryName) throws LensException {
try {
- log.info("ExecutePrepareAsync: session:{} prepareHandle:{}", sessionHandle, prepareHandle.getPrepareHandleId());
+ log.info("ExecutePrepareAsync: session:{} prepareHandle:{}", sessionHandle, prepareHandle.getPrepareHandleId());
acquire(sessionHandle);
PreparedQueryContext pctx = getPreparedQueryContext(sessionHandle, prepareHandle);
Configuration qconf = getLensConf(sessionHandle, conf);
@@ -1771,13 +1759,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
acquire(sessionHandle);
QueryContext ctx = allQueries.get(queryHandle);
if (ctx == null) {
- FinishedLensQuery query = lensServerDao.getQuery(queryHandle.toString());
- log.info("FinishedLensQuery:{}", query);
- if (query == null) {
- throw new NotFoundException("Query not found " + queryHandle);
- }
- // pass the query conf instead of service conf
- return query.toQueryContext(conf, drivers.values());
+ return getQueryContextOfFinishedQuery(queryHandle);
}
updateStatus(queryHandle);
return ctx;
@@ -1786,6 +1768,16 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
}
+ QueryContext getQueryContextOfFinishedQuery(QueryHandle queryHandle) {
+ FinishedLensQuery query = lensServerDao.getQuery(queryHandle.toString());
+ log.info("FinishedLensQuery:{}", query);
+ if (query == null) {
+ throw new NotFoundException("Query not found " + queryHandle);
+ }
+ // pass the query conf instead of service conf
+ return query.toQueryContext(conf, drivers.values());
+ }
+
/**
* Gets the query context.
*
@@ -1981,6 +1973,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
this.notify();
}
}
+
}
/*
@@ -1994,7 +1987,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
public QueryResultSetMetadata getResultSetMetadata(LensSessionHandle sessionHandle, QueryHandle queryHandle)
throws LensException {
try {
- log.info("GetResultSetMetadata: session:{} query: {}", sessionHandle, queryHandle);
+ log.info("GetResultSetMetadata: session:{} query: {}", sessionHandle, queryHandle);
acquire(sessionHandle);
LensResultSet resultSet = getResultset(queryHandle);
if (resultSet != null) {
@@ -2267,7 +2260,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
acquire(sessionHandle);
Configuration qconf = getLensConf(sessionHandle, lensConf);
ExplainQueryContext explainQueryContext = new ExplainQueryContext(requestId, query, getSession(sessionHandle)
- .getLoggedInUser(), lensConf, qconf, drivers.values());
+ .getLoggedInUser(), lensConf, qconf, drivers.values());
explainQueryContext.setLensSessionIdentifier(sessionHandle.getPublicId().toString());
accept(query, qconf, SubmitOp.EXPLAIN);
rewriteAndSelect(explainQueryContext);
@@ -2383,6 +2376,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
break;
case LAUNCHED:
case RUNNING:
+ case EXECUTED:
try {
launchedQueries.add(ctx);
} catch (final Exception e) {
@@ -2486,69 +2480,54 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
return isHealthy
- ? new HealthStatus(isHealthy, "QueryExecution service is healthy.")
- : new HealthStatus(isHealthy, details.toString());
+ ? new HealthStatus(isHealthy, "QueryExecution service is healthy.")
+ : new HealthStatus(isHealthy, details.toString());
}
-
/*
* (non-Javadoc)
*
* @see org.apache.lens.server.api.query.QueryExecutionService#getHttpResultSet(org.apache.lens.api.LensSessionHandle,
* org.apache.lens.api.query.QueryHandle)
*/
+
@Override
public Response getHttpResultSet(LensSessionHandle sessionHandle, QueryHandle queryHandle) throws LensException {
+ LensResultSet resultSet = getResultset(queryHandle);
+ if (!resultSet.isHttpResultAvailable()) {
+ throw new NotFoundException("http result not available");
+ }
+ final Path resultPath = new Path(resultSet.getOutputPath());
final QueryContext ctx = getQueryContext(sessionHandle, queryHandle);
- LensResultSet result = getResultset(queryHandle);
- if (result instanceof LensPersistentResult) {
- final Path resultPath = new Path(((PersistentResultSet) result).getOutputPath());
+ String resultFSReadUrl = conf.get(RESULT_FS_READ_URL);
+ if (resultFSReadUrl != null) {
try {
- FileSystem fs = resultPath.getFileSystem(conf);
- if (fs.isDirectory(resultPath)) {
- throw new NotFoundException("Http result not available for query:" + queryHandle.toString());
- }
- } catch (IOException e) {
- log.warn("Unable to get status for Result Directory", e);
- throw new NotFoundException("Http result not available for query:" + queryHandle.toString());
- }
- String resultFSReadUrl = ctx.getConf().get(RESULT_FS_READ_URL);
- if (resultFSReadUrl != null) {
- try {
- URI resultReadPath = new URI(resultFSReadUrl + resultPath.toUri().getPath() + "?op=OPEN&user.name="
- + getSession(sessionHandle).getClusterUser());
- return Response.seeOther(resultReadPath)
- .header("content-disposition", "attachment; filename = " + resultPath.getName())
- .type(MediaType.APPLICATION_OCTET_STREAM).build();
- } catch (URISyntaxException e) {
- throw new LensException(e);
- }
- } else {
- StreamingOutput stream = new StreamingOutput() {
- @Override
- public void write(OutputStream os) throws IOException {
- FSDataInputStream fin = null;
- try {
- FileSystem fs = resultPath.getFileSystem(ctx.getConf());
- fin = fs.open(resultPath);
- UtilityMethods.pipe(fin, os);
- } finally {
- if (fin != null) {
- fin.close();
- }
-
- }
- }
- };
- return Response.ok(stream).header("content-disposition", "attachment; filename = " + resultPath.getName())
+ URI resultReadPath = new URI(resultFSReadUrl + resultPath.toUri().getPath() + "?op=OPEN&user.name="
+ + getSession(sessionHandle).getClusterUser());
+ return Response.seeOther(resultReadPath)
+ .header("content-disposition", "attachment; filename = " + resultPath.getName())
.type(MediaType.APPLICATION_OCTET_STREAM).build();
+ } catch (URISyntaxException e) {
+ throw new LensException(e);
}
} else {
- String entity = "";
- if (result instanceof InMemoryResultSet || result instanceof PersistentResultSet) {
- entity = "Result is available in driver's "
- + (result instanceof InMemoryResultSet ? "memory" : "persistence") + ".";
- }
- return Response.status(Response.Status.NOT_FOUND).entity(entity).build();
+ StreamingOutput stream = new StreamingOutput() {
+ @Override
+ public void write(OutputStream os) throws IOException {
+ FSDataInputStream fin = null;
+ try {
+ FileSystem fs = resultPath.getFileSystem(ctx.getConf());
+ fin = fs.open(resultPath);
+ UtilityMethods.pipe(fin, os);
+ } finally {
+ if (fin != null) {
+ fin.close();
+ }
+
+ }
+ }
+ };
+ return Response.ok(stream).header("content-disposition", "attachment; filename = " + resultPath.getName())
+ .type(MediaType.APPLICATION_OCTET_STREAM).build();
}
}
@@ -2698,7 +2677,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
Collection<ResourceEntry> pendingResources =
session.getPendingSessionResourcesForDatabase(ctx.getDatabase());
log.info("Adding pending {} session resources for session {} for database {}", pendingResources.size(),
- sessionIdentifier, ctx.getDatabase());
+ sessionIdentifier, ctx.getDatabase());
List<ResourceEntry> failedResources = addResources(pendingResources, sessionHandle, hiveDriver);
// Mark added resources so that we don't add them again. If any of the resources failed
// to be added, then they will be added again
@@ -2761,7 +2740,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
}
log.debug("launchedQueries.remove(finishedQuery) has returned [{}] for finished query with query id:[{}]", modified,
- finishedQuery.getQueryHandleString());
+ finishedQuery.getQueryHandleString());
return modified;
}
@@ -2801,7 +2780,7 @@ public class QueryExecutionServiceImpl extends BaseLensService implements QueryE
private void processWaitingQueries(final FinishedLensQuery finishedQuery) {
Set<QueryContext> eligibleWaitingQueries = this.waitingQueriesSelector
- .selectQueries(finishedQuery, this.waitingQueries);
+ .selectQueries(finishedQuery, this.waitingQueries);
if (eligibleWaitingQueries.isEmpty()) {
log.debug("No queries eligible to move out of waiting state.");
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionStatisticsGenerator.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionStatisticsGenerator.java b/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionStatisticsGenerator.java
index b57bc64..55cabe2 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionStatisticsGenerator.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/QueryExecutionStatisticsGenerator.java
@@ -39,20 +39,15 @@ public class QueryExecutionStatisticsGenerator extends AsyncEventListener<QueryE
/** The Constant LOG. */
private static final Logger LOG = LoggerFactory.getLogger(QueryExecutionStatisticsGenerator.class);
- /** The query service. */
- private final QueryExecutionServiceImpl queryService;
-
/** The event service. */
private final LensEventService eventService;
/**
* Instantiates a new query execution statistics generator.
*
- * @param queryService the query service
* @param eventService the event service
*/
- public QueryExecutionStatisticsGenerator(QueryExecutionServiceImpl queryService, LensEventService eventService) {
- this.queryService = queryService;
+ public QueryExecutionStatisticsGenerator(LensEventService eventService) {
this.eventService = eventService;
}
@@ -68,7 +63,7 @@ public class QueryExecutionStatisticsGenerator extends AsyncEventListener<QueryE
}
QueryHandle handle = ended.getQueryHandle();
QueryExecutionStatistics event = new QueryExecutionStatistics(System.currentTimeMillis());
- QueryContext ctx = queryService.getQueryContext(handle);
+ QueryContext ctx = ended.getQueryContext();
if (ctx == null) {
LOG.warn("Could not find the context for " + handle + " for event:" + ended.getCurrentValue()
+ ". No stat generated");
@@ -88,7 +83,7 @@ public class QueryExecutionStatisticsGenerator extends AsyncEventListener<QueryE
QueryDriverStatistics driverStats = new QueryDriverStatistics();
driverStats.setDriverQuery(ctx.getSelectedDriverQuery());
driverStats.setStartTime(ctx.getDriverStatus().getDriverStartTime());
- driverStats.setEndTime(ctx.getDriverStatus().getDriverStartTime());
+ driverStats.setEndTime(ctx.getDriverStatus().getDriverFinishTime());
event.setDriverStats(driverStats);
try {
if (LOG.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/main/java/org/apache/lens/server/query/ResultFormatter.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/query/ResultFormatter.java b/lens-server/src/main/java/org/apache/lens/server/query/ResultFormatter.java
index af42eb0..f568b17 100644
--- a/lens-server/src/main/java/org/apache/lens/server/query/ResultFormatter.java
+++ b/lens-server/src/main/java/org/apache/lens/server/query/ResultFormatter.java
@@ -65,17 +65,16 @@ public class ResultFormatter extends AsyncEventListener<QueryExecuted> {
*/
@Override
public void process(QueryExecuted event) {
- formatOutput(event);
+ formatOutput(queryService.getQueryContext(event.getQueryHandle()));
}
/**
* Format output.
*
- * @param event the event
+ * @param ctx the query context
*/
- private void formatOutput(QueryExecuted event) {
- QueryHandle queryHandle = event.getQueryHandle();
- QueryContext ctx = queryService.getQueryContext(queryHandle);
+ private void formatOutput(QueryContext ctx) {
+ QueryHandle queryHandle = ctx.getQueryHandle();
this.logSegregationContext.setLogSegragationAndQueryId(ctx.getQueryHandleString());
try {
if (!ctx.isPersistent()) {
@@ -86,9 +85,8 @@ public class ResultFormatter extends AsyncEventListener<QueryExecuted> {
log.info("Result formatter for {}", queryHandle);
LensResultSet resultSet = queryService.getDriverResultset(queryHandle);
boolean isPersistedInDriver = resultSet instanceof PersistentResultSet;
- if (isPersistedInDriver) {
- // skip result formatting if persisted size is huge
- Path persistedDirectory = new Path(ctx.getHdfsoutPath());
+ if (isPersistedInDriver) { // skip result formatting if persisted size is huge
+ Path persistedDirectory = new Path(ctx.getDriverResultPath());
FileSystem fs = persistedDirectory.getFileSystem(ctx.getConf());
long size = fs.getContentSummary(persistedDirectory).getLength();
long threshold = ctx.getConf().getLong(LensConfConstants.RESULT_FORMAT_SIZE_THRESHOLD,
@@ -112,7 +110,7 @@ public class ResultFormatter extends AsyncEventListener<QueryExecuted> {
}
if (isPersistedInDriver) {
log.info("Result formatter for {} in persistent result", queryHandle);
- Path persistedDirectory = new Path(ctx.getHdfsoutPath());
+ Path persistedDirectory = new Path(ctx.getDriverResultPath());
// write all files from persistent directory
((PersistedOutputFormatter) formatter).addRowsFromPersistedPath(persistedDirectory);
} else {
@@ -121,6 +119,7 @@ public class ResultFormatter extends AsyncEventListener<QueryExecuted> {
while (inmemory.hasNext()) {
((InMemoryOutputFormatter) formatter).writeRow(inmemory.next());
}
+ inmemory.setFullyAccessed(true);
}
if (ctx.getConf().getBoolean(LensConfConstants.QUERY_OUTPUT_WRITE_FOOTER,
LensConfConstants.DEFAULT_OUTPUT_WRITE_FOOTER)) {
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/main/resources/lensserver-default.xml
----------------------------------------------------------------------
diff --git a/lens-server/src/main/resources/lensserver-default.xml b/lens-server/src/main/resources/lensserver-default.xml
index 3c8fc09..1e6bd10 100644
--- a/lens-server/src/main/resources/lensserver-default.xml
+++ b/lens-server/src/main/resources/lensserver-default.xml
@@ -279,9 +279,9 @@
<!-- Finished Query Purging Configurations -->
<property>
- <name>lens.server.max.finished.queries</name>
- <value>100</value>
- <description>Maximum number of finished queries which lens server will keep in memory before purging.</description>
+ <name>lens.server.querypurger.sleep.interval</name>
+ <value>10000</value>
+ <description>The interval(milliseconds) with which purger to run periodically. Default 10 sec. </description>
</property>
<property>
<name>lens.server.domain</name>
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/java/org/apache/lens/server/LensJerseyTest.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/LensJerseyTest.java b/lens-server/src/test/java/org/apache/lens/server/LensJerseyTest.java
index 05bd99f..3dad050 100644
--- a/lens-server/src/test/java/org/apache/lens/server/LensJerseyTest.java
+++ b/lens-server/src/test/java/org/apache/lens/server/LensJerseyTest.java
@@ -23,15 +23,17 @@ import static org.testng.Assert.*;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.URI;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
import javax.ws.rs.core.UriBuilder;
import org.apache.lens.driver.hive.TestRemoteHiveDriver;
-import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.metrics.LensMetricsUtil;
import org.apache.lens.server.api.metrics.MetricsService;
import org.apache.lens.server.model.LogSegregationContext;
import org.apache.lens.server.model.MappedDiagnosticLogSegregationContext;
+import org.apache.lens.server.query.QueryExecutionServiceImpl;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hive.service.Service;
@@ -41,6 +43,7 @@ import org.glassfish.jersey.test.JerseyTest;
import org.testng.annotations.AfterSuite;
import org.testng.annotations.BeforeSuite;
+import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
/**
@@ -186,7 +189,6 @@ public abstract class LensJerseyTest extends JerseyTest {
*/
public void restartLensServer() {
HiveConf h = getServerConf();
- h.set(LensConfConstants.MAX_NUMBER_OF_FINISHED_QUERY, "0");
restartLensServer(h);
}
@@ -204,4 +206,19 @@ public abstract class LensJerseyTest extends JerseyTest {
LensServices.get().start();
System.out.println("Lens services restarted!");
}
+ public static void waitForPurge(int allowUnpurgable,
+ ConcurrentLinkedQueue<QueryExecutionServiceImpl.FinishedQuery> finishedQueries) throws InterruptedException {
+ List<QueryExecutionServiceImpl.FinishedQuery> unPurgable = Lists.newArrayList();
+ for (QueryExecutionServiceImpl.FinishedQuery finishedQuery : finishedQueries) {
+ if (!finishedQuery.canBePurged()) {
+ unPurgable.add(finishedQuery);
+ }
+ }
+ if (unPurgable.size() > allowUnpurgable) {
+ throw new RuntimeException("finished queries can't be purged: " + unPurgable);
+ }
+ while (finishedQueries.size() > allowUnpurgable) {
+ Thread.sleep(5000);
+ }
+ }
}