You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lens.apache.org by am...@apache.org on 2015/09/23 11:06:58 UTC
[1/2] lens git commit: LENS-693 : Queries get purged as soon as they
are finished
Repository: lens
Updated Branches:
refs/heads/master d457dd0de -> 669e87272
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/java/org/apache/lens/server/common/RestAPITestUtil.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/common/RestAPITestUtil.java b/lens-server/src/test/java/org/apache/lens/server/common/RestAPITestUtil.java
index 0945598..e266ef2 100644
--- a/lens-server/src/test/java/org/apache/lens/server/common/RestAPITestUtil.java
+++ b/lens-server/src/test/java/org/apache/lens/server/common/RestAPITestUtil.java
@@ -22,9 +22,13 @@ package org.apache.lens.server.common;
import static org.apache.lens.server.common.FormDataMultiPartFactory.createFormDataMultiPartForFact;
import static org.apache.lens.server.common.FormDataMultiPartFactory.createFormDataMultiPartForSession;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.Invocation;
import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.GenericType;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
@@ -36,7 +40,9 @@ import org.apache.lens.api.metastore.XCube;
import org.apache.lens.api.metastore.XFactTable;
import org.apache.lens.api.query.LensQuery;
import org.apache.lens.api.query.QueryHandle;
+import org.apache.lens.api.query.QueryResult;
import org.apache.lens.api.query.QueryStatus;
+import org.apache.lens.api.result.LensAPIResult;
import org.glassfish.jersey.media.multipart.FormDataMultiPart;
@@ -53,49 +59,72 @@ public class RestAPITestUtil {
public static LensSessionHandle openFooBarSession(final WebTarget target) {
return openSession(target, "foo", "bar");
}
+
public static LensSessionHandle openSession(final WebTarget target, final String userName, final String passwd) {
return openSession(target, userName, passwd, new LensConf());
}
public static LensSessionHandle openSession(final WebTarget target, final String userName, final String passwd,
- final LensConf conf) {
+ final LensConf conf) {
final FormDataMultiPart mp = createFormDataMultiPartForSession(Optional.<LensSessionHandle>absent(),
- Optional.of(userName), Optional.of(passwd), Optional.of(conf));
+ Optional.of(userName), Optional.of(passwd), Optional.of(conf));
return target.path("session").request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE),
- LensSessionHandle.class);
+ LensSessionHandle.class);
}
public static Response estimate(final WebTarget target, final Optional<LensSessionHandle> sessionId,
- final Optional<String> query) {
+ final Optional<String> query) {
- return postQuery(target, sessionId, query, Optional.of("estimate"));
+ return postQuery(target, sessionId, query, Optional.of("estimate"), Optional.<LensConf>absent());
}
public static Response execute(final WebTarget target, final Optional<LensSessionHandle> sessionId,
final Optional<String> query) {
- return postQuery(target, sessionId, query, Optional.of("execute"));
+ return execute(target, sessionId, query, Optional.<LensConf>absent());
}
- public static Response postQuery(final WebTarget target, final Optional<LensSessionHandle> sessionId,
- final Optional<String> query, final Optional<String> operation) {
+ public static Response execute(final WebTarget target, final Optional<LensSessionHandle> sessionId,
+ final Optional<String> query, final Optional<LensConf> lensConf) {
+ return postQuery(target, sessionId, query, Optional.of("execute"), lensConf);
+ }
+
+ public static <T> T executeAndGetHandle(final WebTarget target, final Optional<LensSessionHandle> sessionId,
+ final Optional<String> query, final Optional<LensConf> lensConf) {
+ Response resp = postQuery(target, sessionId, query, Optional.of("execute"), lensConf);
+ assertEquals(resp.getStatus(), Response.Status.OK.getStatusCode());
+ T handle = resp.readEntity(new GenericType<LensAPIResult<T>>() {}).getData();
+ assertNotNull(handle);
+ return handle;
+ }
- return postQuery(target, sessionId, query, operation, new LensConf());
+ public static Response postQuery(final WebTarget target, final Optional<LensSessionHandle> sessionId,
+ final Optional<String> query, final Optional<String> operation) {
+ return postQuery(target, sessionId, query, operation, Optional.<LensConf>absent());
}
public static Response postQuery(final WebTarget target, final Optional<LensSessionHandle> sessionId,
- final Optional<String> query, final Optional<String> operation, final LensConf conf) {
+ final Optional<String> query, final Optional<String> operation, Optional<LensConf> lensConfOptional) {
FormDataMultiPart mp = FormDataMultiPartFactory
- .createFormDataMultiPartForQuery(sessionId, query, operation, conf);
+ .createFormDataMultiPartForQuery(sessionId, query, operation, lensConfOptional.or(new LensConf()));
return target.path("queryapi/queries").request(MediaType.APPLICATION_XML).post(
- Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
+ Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
}
- public static void closeSessionFailFast(final WebTarget target, final LensSessionHandle sessionId) {
+ public static LensQuery executeAndWaitForQueryToFinish(WebTarget target, LensSessionHandle lensSessionId,
+ String query, Optional<LensConf> conf, Optional<QueryStatus.Status> statusOptional) throws InterruptedException {
+ QueryHandle handle = executeAndGetHandle(target, Optional.of(lensSessionId), Optional.of(query), conf);
+ if (statusOptional.isPresent()) {
+ return waitForQueryToFinish(target, lensSessionId, handle, statusOptional.get());
+ } else {
+ return waitForQueryToFinish(target, lensSessionId, handle);
+ }
+ }
+ public static void closeSessionFailFast(final WebTarget target, final LensSessionHandle sessionId) {
APIResult result = closeSession(target, sessionId);
checkResponse(result);
}
@@ -114,7 +143,7 @@ public class RestAPITestUtil {
public static APIResult createCube(final WebTarget target, final LensSessionHandle sessionId, final XCube cube) {
return target.path("metastore").path("cubes").queryParam("sessionid", sessionId).request(MediaType.APPLICATION_XML)
- .post(Entity.xml(cubeObjectFactory.createXCube(cube)), APIResult.class);
+ .post(Entity.xml(cubeObjectFactory.createXCube(cube)), APIResult.class);
}
public static void createCubeFailFast(final WebTarget target, final LensSessionHandle sessionId, final XCube cube) {
@@ -123,64 +152,64 @@ public class RestAPITestUtil {
}
public static APIResult createFact(final WebTarget target, final LensSessionHandle sessionId,
- final XFactTable factTable) {
+ final XFactTable factTable) {
FormDataMultiPart mp = createFormDataMultiPartForFact(sessionId, factTable);
return target.path("metastore").path("facts").queryParam("sessionid", sessionId).request(MediaType.APPLICATION_XML)
- .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE), APIResult.class);
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE), APIResult.class);
}
public static void createFactFailFast(final WebTarget target, final LensSessionHandle sessionId,
- final XFactTable factTable) {
+ final XFactTable factTable) {
APIResult result = createFact(target, sessionId, factTable);
checkResponse(result);
}
public static APIResult setCurrentDatabase(final WebTarget target, final LensSessionHandle sessionId,
- final String dbName) {
+ final String dbName) {
WebTarget dbTarget = target.path("metastore").path("databases/current");
return dbTarget.queryParam("sessionid", sessionId).request(MediaType.APPLICATION_XML)
- .put(Entity.xml(dbName),
+ .put(Entity.xml(dbName),
APIResult.class);
}
public static void setCurrentDatabaseFailFast(final WebTarget target, final LensSessionHandle sessionId,
- final String dbName) {
+ final String dbName) {
APIResult result = setCurrentDatabase(target, sessionId, dbName);
checkResponse(result);
}
public static APIResult createDatabase(final WebTarget target, final LensSessionHandle sessionId,
- final String dbName) {
+ final String dbName) {
WebTarget dbTarget = target.path("metastore").path("databases");
return dbTarget.queryParam("sessionid", sessionId).request(MediaType.APPLICATION_XML)
- .post(Entity.xml(dbName), APIResult.class);
+ .post(Entity.xml(dbName), APIResult.class);
}
public static void createDatabaseFailFast(final WebTarget target, final LensSessionHandle sessionId,
- final String dbName) {
+ final String dbName) {
APIResult result = createDatabase(target, sessionId, dbName);
checkResponse(result);
}
public static void createAndSetCurrentDbFailFast(final WebTarget target, final LensSessionHandle sessionId,
- final String dbName) {
+ final String dbName) {
createDatabaseFailFast(target, sessionId, dbName);
setCurrentDatabaseFailFast(target, sessionId, dbName);
}
public static APIResult dropDatabaseFailFast(final WebTarget target, final LensSessionHandle sessionId,
- String dbName) {
+ String dbName) {
WebTarget dbTarget = target.path("metastore").path("databases").path(dbName);
return dbTarget.queryParam("cascade", "true")
- .queryParam("sessionid", sessionId).request(MediaType.APPLICATION_XML).delete(APIResult.class);
+ .queryParam("sessionid", sessionId).request(MediaType.APPLICATION_XML).delete(APIResult.class);
}
private static void checkResponse(final APIResult result) {
@@ -191,22 +220,45 @@ public class RestAPITestUtil {
public static LensQuery waitForQueryToFinish(final WebTarget target, final LensSessionHandle lensSessionHandle,
final QueryHandle handle) throws InterruptedException {
-
- LensQuery lensQuery = getLensQuery(target, lensSessionHandle, handle);
-
- QueryStatus stat = lensQuery.getStatus();
- while (!stat.finished()) {
- lensQuery = getLensQuery(target, lensSessionHandle, handle);
- stat = lensQuery.getStatus();
+ LensQuery ctx = getLensQuery(target, lensSessionHandle, handle);
+ while (!ctx.getStatus().finished()) {
+ ctx = getLensQuery(target, lensSessionHandle, handle);
Thread.sleep(1000);
}
+ return ctx;
+ }
+
+ public static LensQuery waitForQueryToFinish(final WebTarget target, final LensSessionHandle lensSessionHandle,
+ final QueryHandle handle, QueryStatus.Status status) throws InterruptedException {
+ LensQuery lensQuery = waitForQueryToFinish(target, lensSessionHandle, handle);
+ assertEquals(lensQuery.getStatus().getStatus(), status);
return lensQuery;
}
public static LensQuery getLensQuery(final WebTarget target, final LensSessionHandle lensSessionHandle,
final QueryHandle handle) {
-
return target.path("queryapi/queries").path(handle.toString()).queryParam("sessionid", lensSessionHandle).request()
- .get(LensQuery.class);
+ .get(LensQuery.class);
+ }
+
+ public static QueryResult getLensQueryResult(final WebTarget target, final LensSessionHandle lensSessionHandle,
+ final QueryHandle handle) {
+ return target.path("queryapi/queries").path(handle.toString()).path("resultset")
+ .queryParam("sessionid", lensSessionHandle).request().get(QueryResult.class);
+ }
+
+ public static Response getLensQueryHttpResult(final WebTarget target, final LensSessionHandle lensSessionHandle,
+ final QueryHandle handle) {
+ return target.path("queryapi/queries").path(handle.toString()).path("httpresultset")
+ .queryParam("sessionid", lensSessionHandle).request().get(Response.class);
+ }
+
+ public static LensConf getLensConf(Object... args) {
+ assertEquals(args.length % 2, 0);
+ LensConf conf = new LensConf();
+ for (int i = 0; i < args.length; i += 2) {
+ conf.addProperty(args[i], args[i + 1]);
+ }
+ return conf;
}
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/java/org/apache/lens/server/query/TestEventService.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/query/TestEventService.java b/lens-server/src/test/java/org/apache/lens/server/query/TestEventService.java
index b8012be..1dab35e 100644
--- a/lens-server/src/test/java/org/apache/lens/server/query/TestEventService.java
+++ b/lens-server/src/test/java/org/apache/lens/server/query/TestEventService.java
@@ -92,7 +92,6 @@ public class TestEventService {
* the genericEvent event occurs, that object's appropriate
* method is invoked.
*
- * @see GenericEventEvent
*/
class GenericEventListener extends AsyncEventListener<LensEvent> {
@@ -119,7 +118,6 @@ public class TestEventService {
* the mockFailed event occurs, that object's appropriate
* method is invoked.
*
- * @see MockFailedEvent
*/
class MockFailedListener implements LensEventListener<QueryFailed> {
@@ -146,7 +144,6 @@ public class TestEventService {
* the mockEnded event occurs, that object's appropriate
* method is invoked.
*
- * @see MockEndedEvent
*/
class MockEndedListener implements LensEventListener<QueryEnded> {
@@ -290,7 +287,7 @@ public class TestEventService {
*/
@Test
public void testAddListener() {
- int listenersBefore = ((EventServiceImpl) service).getEventListeners().keySet().size();
+ int listenersBefore = service.getEventListeners().keySet().size();
genericEventListener = new GenericEventListener();
service.addListenerForType(genericEventListener, LensEvent.class);
endedListener = new MockEndedListener();
@@ -417,8 +414,10 @@ public class TestEventService {
QueryHandle query = new QueryHandle(UUID.randomUUID());
String user = "user";
long now = System.currentTimeMillis();
- QueryFailed failed = new QueryFailed(now, QueryStatus.Status.RUNNING, QueryStatus.Status.FAILED, query, user, null);
- QuerySuccess success = new QuerySuccess(now, QueryStatus.Status.RUNNING, QueryStatus.Status.SUCCESSFUL, query);
+ QueryFailed failed
+ = new QueryFailed(null, now, QueryStatus.Status.RUNNING, QueryStatus.Status.FAILED, query, user, null);
+ QuerySuccess success
+ = new QuerySuccess(null, now, QueryStatus.Status.RUNNING, QueryStatus.Status.SUCCESSFUL, query);
QueuePositionChange positionChange = new QueuePositionChange(now, 1, 0, query);
try {
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/java/org/apache/lens/server/query/TestQueryEndEmailNotifier.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/query/TestQueryEndEmailNotifier.java b/lens-server/src/test/java/org/apache/lens/server/query/TestQueryEndEmailNotifier.java
index 82ce205..3bb90ef 100644
--- a/lens-server/src/test/java/org/apache/lens/server/query/TestQueryEndEmailNotifier.java
+++ b/lens-server/src/test/java/org/apache/lens/server/query/TestQueryEndEmailNotifier.java
@@ -18,28 +18,25 @@
*/
package org.apache.lens.server.query;
-import static org.apache.lens.server.common.RestAPITestUtil.execute;
-import static org.apache.lens.server.common.RestAPITestUtil.waitForQueryToFinish;
+import static org.apache.lens.server.common.RestAPITestUtil.*;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import javax.ws.rs.client.Entity;
-import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Application;
import javax.ws.rs.core.GenericType;
-import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensSessionHandle;
-import org.apache.lens.api.query.LensQuery;
+import org.apache.lens.api.query.InMemoryQueryResult;
import org.apache.lens.api.query.QueryHandle;
-import org.apache.lens.api.query.QueryStatus;
+import org.apache.lens.api.query.QueryResult;
import org.apache.lens.api.query.QueryStatus.Status;
import org.apache.lens.api.result.LensAPIResult;
import org.apache.lens.server.LensJerseyTest;
@@ -52,16 +49,10 @@ import org.apache.lens.server.common.TestResourceFile;
import org.apache.hadoop.hive.conf.HiveConf;
import org.glassfish.jersey.client.ClientConfig;
-import org.glassfish.jersey.media.multipart.FormDataBodyPart;
-import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
-import org.glassfish.jersey.media.multipart.FormDataMultiPart;
import org.glassfish.jersey.media.multipart.MultiPartFeature;
import org.subethamail.wiser.Wiser;
import org.subethamail.wiser.WiserMessage;
-import org.testng.Assert;
-import org.testng.annotations.AfterTest;
-import org.testng.annotations.BeforeTest;
-import org.testng.annotations.Test;
+import org.testng.annotations.*;
import com.google.common.base.Optional;
@@ -93,7 +84,7 @@ public class TestQueryEndEmailNotifier extends LensJerseyTest {
wiser.setHostname("localhost");
wiser.setPort(25000);
queryService = LensServices.get().getService(QueryExecutionService.NAME);
- Map<String, String> sessionconf = new HashMap<String, String>();
+ Map<String, String> sessionconf = new HashMap<>();
sessionconf.put("test.session.key", "svalue");
sessionconf.put(LensConfConstants.QUERY_MAIL_NOTIFY, "true");
sessionconf.put(LensConfConstants.QUERY_RESULT_EMAIL_CC, "foo1@localhost,foo2@localhost,foo3@localhost");
@@ -101,6 +92,12 @@ public class TestQueryEndEmailNotifier extends LensJerseyTest {
// automatically
createTable(TEST_TABLE);
loadData(TEST_TABLE, TestResourceFile.TEST_DATA2_FILE.getValue());
+ wiser.start();
+ }
+
+ @BeforeMethod
+ public void clearWiserMessages() {
+ wiser.getMessages().clear();
}
/*
@@ -110,6 +107,7 @@ public class TestQueryEndEmailNotifier extends LensJerseyTest {
*/
@AfterTest
public void tearDown() throws Exception {
+ wiser.stop();
dropTable(TEST_TABLE);
queryService.closeSession(lensSessionId);
super.tearDown();
@@ -171,28 +169,29 @@ public class TestQueryEndEmailNotifier extends LensJerseyTest {
private QueryHandle launchAndWaitForQuery(LensConf conf, String query, Status expectedStatus)
throws InterruptedException {
- final WebTarget target = target().path("queryapi/queries");
- final FormDataMultiPart mp = new FormDataMultiPart();
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("sessionid").build(), lensSessionId,
- MediaType.APPLICATION_XML_TYPE));
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("query").build(), query));
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("operation").build(), "execute"));
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("conf").fileName("conf").build(), conf,
- MediaType.APPLICATION_XML_TYPE));
- final QueryHandle handle = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE),
- new GenericType<LensAPIResult<QueryHandle>>(){}).getData();
+ return executeAndWaitForQueryToFinish(target(), lensSessionId, query, Optional.of(conf),
+ Optional.of(expectedStatus)).getQueryHandle();
+ }
+
+ private WiserMessage getMessage() throws InterruptedException {
+ List<WiserMessage> messages = new ArrayList<>();
+ for (int i = 0; i < NUM_ITERS; i++) {
+ messages = wiser.getMessages();
+ if (messages.size() >= 4) {
+ break;
+ }
+ Thread.sleep(2000);
+ }
+ assertEquals(messages.size(), 4);
+ return messages.get(0);
+ }
- Assert.assertNotNull(handle);
- LensQuery ctx = target.path(handle.toString()).queryParam("sessionid", lensSessionId).request()
- .get(LensQuery.class);
- QueryStatus stat = ctx.getStatus();
- while (!stat.finished()) {
- ctx = target.path(handle.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- stat = ctx.getStatus();
- Thread.sleep(1000);
+ private void assertKeywordsContains(Object o, Object... keywords) {
+ String string = o.toString();
+ for (Object keyword : keywords) {
+ // Mail message converts \n to \r\n
+ assertTrue(string.contains(keyword.toString().replaceAll("\\n", "\r\n")), o + " doesn't contain " + keyword);
}
- Assert.assertEquals(ctx.getStatus().getStatus(), expectedStatus);
- return handle;
}
/**
@@ -201,91 +200,70 @@ public class TestQueryEndEmailNotifier extends LensJerseyTest {
* @throws InterruptedException the interrupted exception
*/
@Test
- public void testEmailNotification() throws InterruptedException {
- wiser.start();
-
- LensConf conf = new LensConf();
-
+ public void testLaunchFailure() throws InterruptedException {
// launch failure
final Response response = execute(target(), Optional.of(lensSessionId), Optional.of("select fail from non_exist"));
assertEquals(response.getStatus(), Response.Status.OK.getStatusCode());
QueryHandle handle = response.readEntity(new GenericType<LensAPIResult<QueryHandle>>() {}).getData();
- LensQuery lensQuery = waitForQueryToFinish(target(), lensSessionId, handle);
-
- List<WiserMessage> messages = new ArrayList<WiserMessage>();
- for (int i = 0; i < NUM_ITERS; i++) {
- messages = wiser.getMessages();
- if (messages.size() >= 4) {
- break;
- }
- Thread.sleep(2000);
- }
-
- Assert.assertEquals(messages.size(), 4);
- Assert.assertTrue(messages.get(0).toString().contains(handle.toString()));
- Assert.assertTrue(messages.get(0).toString().contains("Launching query failed"));
- Assert.assertTrue(messages.get(0).toString().contains("Reason"));
+ assertKeywordsContains(getMessage(), handle, "Launching query failed", "Reason");
+ }
+ @Test
+ public void testFormattingFailure() throws InterruptedException {
// formatting failure
- conf = new LensConf();
- conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_SET, "true");
- conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "false");
- conf.addProperty(LensConfConstants.QUERY_OUTPUT_SERDE, "NonexistentSerde.class");
- handle = launchAndWaitForQuery(conf, "select ID, IDSTR from " + TEST_TABLE,
- QueryStatus.Status.FAILED);
-
- messages = new ArrayList<WiserMessage>();
- for (int i = 0; i < NUM_ITERS; i++) {
- messages = wiser.getMessages();
- if (messages.size() >= 8) {
- break;
- }
- Thread.sleep(2000);
- }
-
- Assert.assertEquals(messages.size(), 8);
- Assert.assertTrue(messages.get(4).toString().contains(handle.toString()));
- Assert.assertTrue(messages.get(4).toString().contains("Result formatting failed!"));
- Assert.assertTrue(messages.get(4).toString().contains("Reason"));
+ LensConf conf = getLensConf(
+ LensConfConstants.QUERY_PERSISTENT_RESULT_SET, "true",
+ LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "false",
+ LensConfConstants.QUERY_OUTPUT_SERDE, "NonexistentSerde.class");
+ QueryHandle handle = launchAndWaitForQuery(conf, "select ID, IDSTR from " + TEST_TABLE, Status.FAILED);
+ assertKeywordsContains(getMessage(), handle, "Result formatting failed!", "Reason");
+ }
+ @Test
+ public void testExecutionFailure() throws InterruptedException {
// execution failure
- conf = new LensConf();
- conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "true");
- conf.addProperty(HiveConf.ConfVars.COMPRESSRESULT.name(), "true");
- conf.addProperty("mapred.compress.map.output", "true");
- conf.addProperty("mapred.map.output.compression.codec", "nonexisting");
- handle = launchAndWaitForQuery(conf, "select count(ID) from " + TEST_TABLE, QueryStatus.Status.FAILED);
- messages = new ArrayList<WiserMessage>();
- for (int i = 0; i < NUM_ITERS; i++) {
- messages = wiser.getMessages();
- if (messages.size() >= 8) {
- break;
- }
- Thread.sleep(2000);
- }
+ LensConf conf = getLensConf(
+ LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "true",
+ HiveConf.ConfVars.COMPRESSRESULT.name(), "true",
+ "mapred.compress.map.output", "true",
+ "mapred.map.output.compression.codec", "nonexisting");
+ QueryHandle handle = launchAndWaitForQuery(conf, "select count(ID) from " + TEST_TABLE, Status.FAILED);
+ assertKeywordsContains(getMessage(), handle, "Query execution failed!", "Reason");
+ }
- Assert.assertEquals(messages.size(), 12);
- Assert.assertTrue(messages.get(8).toString().contains(handle.toString()));
- Assert.assertTrue(messages.get(8).toString().contains("Query execution failed!"));
- Assert.assertTrue(messages.get(8).toString().contains("Reason"));
+ @DataProvider(name = "success-tests")
+ public Object[][] persistenceConfigDataProvider() {
+ return new Object[][]{
+ {false, false, },
+ {true, false, },
+ {false, true, },
+ {true, true, },
+ };
+ }
+ @Test(dataProvider = "success-tests")
+ public void testSuccessfulQuery(Boolean lensPersistence, Boolean driverPersistence) throws InterruptedException {
// successful query
- conf = new LensConf();
- conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_SET, "true");
- conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "true");
- handle = launchAndWaitForQuery(conf, "select ID, IDSTR from " + TEST_TABLE, QueryStatus.Status.SUCCESSFUL);
- messages = new ArrayList<WiserMessage>();
- for (int i = 0; i < NUM_ITERS; i++) {
- messages = wiser.getMessages();
- if (messages.size() >= 16) {
- break;
- }
- Thread.sleep(2000);
+ LensConf conf = getLensConf(
+ LensConfConstants.QUERY_PERSISTENT_RESULT_SET, lensPersistence,
+ LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, driverPersistence);
+ QueryHandle handle = launchAndWaitForQuery(conf, "select ID, IDSTR from " + TEST_TABLE, Status.SUCCESSFUL);
+ String expectedKeywords;
+ if (lensPersistence || driverPersistence) {
+ QueryResult result = getLensQueryResult(target(), lensSessionId, handle);
+ expectedKeywords = result.toPrettyString();
+ } else {
+ expectedKeywords = InMemoryQueryResult.DECLARATION;
+ }
+ WiserMessage message = getMessage();
+ assertKeywordsContains(message, handle, "Query SUCCESSFUL", expectedKeywords);
+ if (lensPersistence) {
+ assertKeywordsContains(message, new String[]{"Downloadable from", "httpresultset"});
+ assertEquals(getLensQueryHttpResult(target(), lensSessionId, handle).getStatus(),
+ Response.Status.OK.getStatusCode());
+ } else {
+ assertEquals(getLensQueryHttpResult(target(), lensSessionId, handle).getStatus(),
+ Response.Status.NOT_FOUND.getStatusCode());
}
- Assert.assertEquals(messages.size(), 16);
- Assert.assertTrue(messages.get(12).toString().contains(handle.toString()));
- Assert.assertTrue(messages.get(12).toString().contains("Query SUCCESSFUL"));
- Assert.assertTrue(messages.get(12).toString().contains("Result available at"));
- wiser.stop();
}
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/java/org/apache/lens/server/query/TestQueryService.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/query/TestQueryService.java b/lens-server/src/test/java/org/apache/lens/server/query/TestQueryService.java
index fa8d6ee..ea75ffb 100644
--- a/lens-server/src/test/java/org/apache/lens/server/query/TestQueryService.java
+++ b/lens-server/src/test/java/org/apache/lens/server/query/TestQueryService.java
@@ -18,10 +18,9 @@
*/
package org.apache.lens.server.query;
-import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
+import static javax.ws.rs.core.Response.Status.*;
-import static org.apache.lens.server.common.RestAPITestUtil.execute;
-import static org.apache.lens.server.common.RestAPITestUtil.waitForQueryToFinish;
+import static org.apache.lens.server.common.RestAPITestUtil.*;
import static org.testng.Assert.*;
@@ -31,7 +30,10 @@ import java.util.*;
import javax.ws.rs.NotFoundException;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
-import javax.ws.rs.core.*;
+import javax.ws.rs.core.Application;
+import javax.ws.rs.core.GenericType;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
import org.apache.lens.api.APIResult;
import org.apache.lens.api.LensConf;
@@ -39,7 +41,9 @@ import org.apache.lens.api.LensSessionHandle;
import org.apache.lens.api.jaxb.LensJAXBContextResolver;
import org.apache.lens.api.query.*;
import org.apache.lens.api.query.QueryStatus.Status;
-import org.apache.lens.api.result.*;
+import org.apache.lens.api.result.LensAPIResult;
+import org.apache.lens.api.result.LensErrorTO;
+import org.apache.lens.api.result.QueryCostTO;
import org.apache.lens.cube.error.LensCubeErrorCode;
import org.apache.lens.driver.hive.HiveDriver;
import org.apache.lens.server.LensJerseyTest;
@@ -62,17 +66,24 @@ import org.apache.lens.server.session.HiveSessionService;
import org.apache.lens.server.session.LensSessionImpl;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.*;
+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.hadoop.io.IOUtils;
import org.glassfish.jersey.client.ClientConfig;
-import org.glassfish.jersey.media.multipart.*;
+import org.glassfish.jersey.media.multipart.FormDataBodyPart;
+import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
+import org.glassfish.jersey.media.multipart.FormDataMultiPart;
+import org.glassfish.jersey.media.multipart.MultiPartFeature;
import org.glassfish.jersey.test.TestProperties;
-import org.testng.annotations.*;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.AfterTest;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
import com.codahale.metrics.MetricRegistry;
-
import com.google.common.base.Optional;
import lombok.extern.slf4j.Slf4j;
@@ -102,6 +113,7 @@ public class TestQueryService extends LensJerseyTest {
return classes;
}
}
+
/*
* (non-Javadoc)
*
@@ -218,14 +230,14 @@ public class TestQueryService extends LensJerseyTest {
LensConf conf = new LensConf();
final FormDataMultiPart mp = new FormDataMultiPart();
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("sessionid").build(), lensSessionId,
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
mp.bodyPart(
- new FormDataBodyPart(FormDataContentDisposition.name("query").build(), "select ID from non_exist_table"));
+ new FormDataBodyPart(FormDataContentDisposition.name("query").build(), "select ID from non_exist_table"));
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("operation").build(), "execute"));
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("conf").fileName("conf").build(), conf,
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
final Response response = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
- assertEquals(response.getStatus(), Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
+ assertEquals(response.getStatus(), INTERNAL_SERVER_ERROR.getStatusCode());
}
/**
@@ -235,19 +247,13 @@ public class TestQueryService extends LensJerseyTest {
*/
@Test
public void testLaunchFail() throws InterruptedException {
-
- final Response response = execute(target(), Optional.of(lensSessionId), Optional.of("select fail from non_exist"));
- assertEquals(response.getStatus(), Response.Status.OK.getStatusCode());
-
- QueryHandle handle = response.readEntity(new GenericType<LensAPIResult<QueryHandle>>() {}).getData();
- LensQuery lensQuery = waitForQueryToFinish(target(), lensSessionId, handle);
-
+ LensQuery lensQuery = executeAndWaitForQueryToFinish(target(), lensSessionId, "select fail from non_exist",
+ Optional.<LensConf>absent(), Optional.of(Status.FAILED));
assertTrue(lensQuery.getSubmissionTime() > 0);
assertEquals(lensQuery.getLaunchTime(), 0);
assertEquals(lensQuery.getDriverStartTime(), 0);
assertEquals(lensQuery.getDriverFinishTime(), 0);
assertTrue(lensQuery.getFinishTime() > 0);
- assertEquals(lensQuery.getStatus().getStatus(), QueryStatus.Status.FAILED);
}
// test with execute async post, get all queries, get query context,
@@ -262,22 +268,13 @@ public class TestQueryService extends LensJerseyTest {
public void testQueriesAPI() throws InterruptedException {
// test post execute op
final WebTarget target = target().path("queryapi/queries");
- final FormDataMultiPart mp = new FormDataMultiPart();
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("sessionid").build(), lensSessionId,
- MediaType.APPLICATION_XML_TYPE));
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("query").build(), "select ID from " + TEST_TABLE));
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("operation").build(), "execute"));
- mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("conf").fileName("conf").build(), new LensConf(),
- MediaType.APPLICATION_XML_TYPE));
long queuedQueries = metricsSvc.getQueuedQueries();
long runningQueries = metricsSvc.getRunningQueries();
long finishedQueries = metricsSvc.getFinishedQueries();
- final QueryHandle handle = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE),
- new GenericType<LensAPIResult<QueryHandle>>() {
- }).getData();
- assertNotNull(handle);
+ QueryHandle handle = executeAndGetHandle(target(), Optional.of(lensSessionId), Optional.of("select ID from "
+ + TEST_TABLE), Optional.<LensConf>absent());
// Get all queries
// XML
@@ -285,7 +282,7 @@ public class TestQueryService extends LensJerseyTest {
.get(new GenericType<List<QueryHandle>>() {});
assertTrue(allQueriesXML.size() >= 1);
- List<QueryHandle> allQueries = (List<QueryHandle>) target.queryParam("sessionid", lensSessionId).request()
+ List<QueryHandle> allQueries = target.queryParam("sessionid", lensSessionId).request()
.get(new GenericType<List<QueryHandle>>() {});
assertTrue(allQueries.size() >= 1);
assertTrue(allQueries.contains(handle));
@@ -375,7 +372,7 @@ public class TestQueryService extends LensJerseyTest {
MediaType.APPLICATION_XML_TYPE));
final QueryPlan plan2 = ptarget.request().post(Entity.entity(mp2, MediaType.MULTIPART_FORM_DATA_TYPE),
- new GenericType<LensAPIResult<QueryPlan>>() {}).getData();
+ new GenericType<LensAPIResult<QueryPlan>>() {}).getData();
assertEquals(plan2.getTablesQueried().size(), 1);
assertTrue(plan2.getTablesQueried().get(0).endsWith(TEST_TABLE.toLowerCase()));
assertNotNull(plan2.getPrepareHandle());
@@ -395,33 +392,33 @@ public class TestQueryService extends LensJerseyTest {
final FormDataMultiPart mp = new FormDataMultiPart();
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("sessionid").build(), lensSessionId,
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("query").build(), "select NO_ID from "
- + TEST_TABLE));
+ + TEST_TABLE));
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("operation").build(), "explain"));
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("conf").fileName("conf").build(), new LensConf(),
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
final Response responseExplain = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
- assertEquals(responseExplain.getStatus(), Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
+ assertEquals(responseExplain.getStatus(), INTERNAL_SERVER_ERROR.getStatusCode());
// Test explain and prepare
final WebTarget ptarget = target().path("queryapi/preparedqueries");
final FormDataMultiPart mp2 = new FormDataMultiPart();
mp2.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("sessionid").build(), lensSessionId,
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
mp2.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("query").build(), "select NO_ID from "
- + TEST_TABLE));
+ + TEST_TABLE));
mp2.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("operation").build(), "explain_and_prepare"));
mp2.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("conf").fileName("conf").build(), new LensConf(),
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
final Response responseExplainAndPrepare = target.request().post(
- Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
+ Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
- assertEquals(responseExplainAndPrepare.getStatus(), Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
+ assertEquals(responseExplainAndPrepare.getStatus(), INTERNAL_SERVER_ERROR.getStatusCode());
}
// post to preparedqueries
@@ -451,7 +448,7 @@ public class TestQueryService extends LensJerseyTest {
MediaType.APPLICATION_XML_TYPE));
final QueryPrepareHandle pHandle = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE),
- new GenericType<LensAPIResult<QueryPrepareHandle>>() {}).getData();
+ new GenericType<LensAPIResult<QueryPrepareHandle>>() {}).getData();
// Get all prepared queries
List<QueryPrepareHandle> allQueries = (List<QueryPrepareHandle>) target.queryParam("sessionid", lensSessionId)
@@ -493,32 +490,11 @@ public class TestQueryService extends LensJerseyTest {
.post(Entity.entity(confpart, MediaType.MULTIPART_FORM_DATA_TYPE), QueryHandle.class);
assertNotEquals(handle1, handle2);
- LensQuery ctx1 = target().path("queryapi/queries").path(handle1.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
+ LensQuery ctx1 = waitForQueryToFinish(target(), lensSessionId, handle1, Status.SUCCESSFUL);
assertEquals(ctx1.getQueryName().toLowerCase(), "testquery1");
- // wait till the query finishes
- QueryStatus stat = ctx1.getStatus();
- while (!stat.finished()) {
- ctx1 = target().path("queryapi/queries").path(handle1.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- stat = ctx1.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx1.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
- LensQuery ctx2 = target().path("queryapi/queries").path(handle2.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- assertNotNull(ctx2);
+ LensQuery ctx2 = waitForQueryToFinish(target(), lensSessionId, handle2, Status.SUCCESSFUL);
assertEquals(ctx2.getQueryName().toLowerCase(), "testqueryname2");
- // wait till the query finishes
- stat = ctx2.getStatus();
- while (!stat.finished()) {
- ctx2 = target().path("queryapi/queries").path(handle1.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- stat = ctx2.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx1.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
// destroy prepared
APIResult result = target.path(pHandle.toString()).queryParam("sessionid", lensSessionId).request()
@@ -550,7 +526,7 @@ public class TestQueryService extends LensJerseyTest {
final QueryPlan plan = target.request()
.post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE),
- new GenericType<LensAPIResult<QueryPlan>>() {}).getData();
+ new GenericType<LensAPIResult<QueryPlan>>() {}).getData();
assertEquals(plan.getTablesQueried().size(), 1);
assertTrue(plan.getTablesQueried().get(0).endsWith(TEST_TABLE.toLowerCase()));
@@ -588,29 +564,8 @@ public class TestQueryService extends LensJerseyTest {
.post(Entity.entity(confpart, MediaType.MULTIPART_FORM_DATA_TYPE), QueryHandle.class);
assertNotEquals(handle1, handle2);
- LensQuery ctx1 = target().path("queryapi/queries").path(handle1.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- // wait till the query finishes
- QueryStatus stat = ctx1.getStatus();
- while (!stat.finished()) {
- ctx1 = target().path("queryapi/queries").path(handle1.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- stat = ctx1.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx1.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
-
- LensQuery ctx2 = target().path("queryapi/queries").path(handle2.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- // wait till the query finishes
- stat = ctx2.getStatus();
- while (!stat.finished()) {
- ctx2 = target().path("queryapi/queries").path(handle1.toString()).queryParam("sessionid", lensSessionId)
- .request().get(LensQuery.class);
- stat = ctx2.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx1.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
+ waitForQueryToFinish(target(), lensSessionId, handle1, Status.SUCCESSFUL);
+ waitForQueryToFinish(target(), lensSessionId, handle2, Status.SUCCESSFUL);
// destroy prepared
APIResult result = target.path(plan.getPrepareHandle().toString()).queryParam("sessionid", lensSessionId).request()
@@ -660,7 +615,7 @@ public class TestQueryService extends LensJerseyTest {
assertTrue(lensQuery.getStatus().getStatus().equals(Status.QUEUED)
|| lensQuery.getStatus().getStatus().equals(Status.LAUNCHED)
|| lensQuery.getStatus().getStatus().equals(Status.RUNNING)
- || lensQuery.getStatus().getStatus().equals(Status.SUCCESSFUL));
+ || lensQuery.getStatus().getStatus().equals(Status.SUCCESSFUL), lensQuery.getStatus().toString());
// wait till the query finishes
QueryStatus stat = lensQuery.getStatus();
@@ -729,14 +684,7 @@ public class TestQueryService extends LensJerseyTest {
new GenericType<LensAPIResult<QueryHandle>>() {}).getData();
// Get query
- lensQuery = target.path(handle3.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- // wait till the query finishes
- stat = lensQuery.getStatus();
- while (!stat.finished()) {
- lensQuery = target.path(handle3.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- stat = lensQuery.getStatus();
- Thread.sleep(1000);
- }
+ waitForQueryToFinish(target(), lensSessionId, handle3, Status.SUCCESSFUL);
validateHttpEndPoint(target(), null, handle3, null);
}
@@ -918,7 +866,7 @@ public class TestQueryService extends LensJerseyTest {
List<String> actualRows = Arrays.asList(result.split("\n"));
validatePersistentResult(actualRows);
} else {
- assertEquals(Response.Status.SEE_OTHER.getStatusCode(), response.getStatus());
+ assertEquals(SEE_OTHER.getStatusCode(), response.getStatus());
assertTrue(response.getHeaderString("Location").contains(redirectUrl));
}
}
@@ -934,10 +882,10 @@ public class TestQueryService extends LensJerseyTest {
try {
Response response = parent.path("queryapi/queries/" + handle.toString() + "/httpresultset")
.queryParam("sessionid", lensSessionId).request().get();
- if (Response.Status.NOT_FOUND.getStatusCode() != response.getStatus()) {
+ if (NOT_FOUND.getStatusCode() != response.getStatus()) {
fail("Expected not found excepiton, but got:" + response.getStatus());
}
- assertEquals(response.getStatus(), Response.Status.NOT_FOUND.getStatusCode());
+ assertEquals(response.getStatus(), NOT_FOUND.getStatusCode());
} catch (NotFoundException e) {
// expected
log.error("Resource not found.", e);
@@ -975,27 +923,13 @@ public class TestQueryService extends LensJerseyTest {
assertNotNull(handle);
// Get query
- LensQuery ctx = target.path(handle.toString()).queryParam("sessionid", lensSessionId).request()
- .get(LensQuery.class);
- assertTrue(ctx.getStatus().getStatus().equals(Status.QUEUED)
- || ctx.getStatus().getStatus().equals(Status.LAUNCHED) || ctx.getStatus().getStatus().equals(Status.RUNNING)
- || ctx.getStatus().getStatus().equals(Status.SUCCESSFUL));
-
- // wait till the query finishes
- QueryStatus stat = ctx.getStatus();
- while (!stat.finished()) {
- ctx = target.path(handle.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- stat = ctx.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
+ waitForQueryToFinish(target(), lensSessionId, handle, Status.SUCCESSFUL);
// fetch results
validateResultSetMetadata(handle, "",
new String[][]{{"ID", "INT"}, {"IDSTR", "STRING"}},
target(), lensSessionId);
-
InMemoryQueryResult resultset = target.path(handle.toString()).path("resultset")
.queryParam("sessionid", lensSessionId).request().get(InMemoryQueryResult.class);
validateInmemoryResult(resultset);
@@ -1030,20 +964,7 @@ public class TestQueryService extends LensJerseyTest {
assertNotNull(dropHandle);
// Get query
- LensQuery ctx = target.path(dropHandle.toString()).queryParam("sessionid", lensSessionId).request()
- .get(LensQuery.class);
- assertTrue(ctx.getStatus().getStatus().equals(Status.QUEUED)
- || ctx.getStatus().getStatus().equals(Status.LAUNCHED) || ctx.getStatus().getStatus().equals(Status.RUNNING)
- || ctx.getStatus().getStatus().equals(Status.SUCCESSFUL));
-
- // wait till the query finishes
- QueryStatus stat = ctx.getStatus();
- while (!stat.finished()) {
- ctx = target.path(dropHandle.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- stat = ctx.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
+ waitForQueryToFinish(target(), lensSessionId, dropHandle, Status.SUCCESSFUL);
final FormDataMultiPart mp = new FormDataMultiPart();
conf = new LensConf();
@@ -1061,19 +982,7 @@ public class TestQueryService extends LensJerseyTest {
assertNotNull(handle);
// Get query
- ctx = target.path(handle.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- assertTrue(ctx.getStatus().getStatus().equals(Status.QUEUED)
- || ctx.getStatus().getStatus().equals(Status.LAUNCHED) || ctx.getStatus().getStatus().equals(Status.RUNNING)
- || ctx.getStatus().getStatus().equals(Status.SUCCESSFUL));
-
- // wait till the query finishes
- stat = ctx.getStatus();
- while (!stat.finished()) {
- ctx = target.path(handle.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- stat = ctx.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
+ waitForQueryToFinish(target(), lensSessionId, handle, Status.SUCCESSFUL);
String select = "SELECT * FROM temp_output";
final FormDataMultiPart fetch = new FormDataMultiPart();
@@ -1089,16 +998,7 @@ public class TestQueryService extends LensJerseyTest {
assertNotNull(handle2);
// Get query
- ctx = target.path(handle2.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
-
- // wait till the query finishes
- stat = ctx.getStatus();
- while (!stat.finished()) {
- ctx = target.path(handle2.toString()).queryParam("sessionid", lensSessionId).request().get(LensQuery.class);
- stat = ctx.getStatus();
- Thread.sleep(1000);
- }
- assertEquals(ctx.getStatus().getStatus(), QueryStatus.Status.SUCCESSFUL);
+ waitForQueryToFinish(target(), lensSessionId, handle2, Status.SUCCESSFUL);
// fetch results
validateResultSetMetadata(handle2, "temp_output.", new String[][]{{"ID", "INT"}, {"IDSTR", "STRING"}},
@@ -1236,10 +1136,10 @@ public class TestQueryService extends LensJerseyTest {
// set a timeout value enough for tests
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("timeoutmillis").build(), "300000"));
mp.bodyPart(new FormDataBodyPart(FormDataContentDisposition.name("conf").fileName("conf").build(), new LensConf(),
- MediaType.APPLICATION_XML_TYPE));
+ MediaType.APPLICATION_XML_TYPE));
Response response = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
- assertEquals(response.getStatus(), Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
+ assertEquals(response.getStatus(), INTERNAL_SERVER_ERROR.getStatusCode());
}
/**
@@ -1423,7 +1323,7 @@ public class TestQueryService extends LensJerseyTest {
LensErrorTO expectedLensErrorTO = LensErrorTO.composedOf(
- LensCubeErrorCode.NEITHER_CUBE_NOR_DIMENSION.getLensErrorInfo().getErrorCode(),
+ LensCubeErrorCode.NEITHER_CUBE_NOR_DIMENSION.getLensErrorInfo().getErrorCode(),
"Neither cube nor dimensions accessed in the query", TestDataUtils.MOCK_STACK_TRACE);
ErrorResponseExpectedData expectedData = new ErrorResponseExpectedData(BAD_REQUEST, expectedLensErrorTO);
@@ -1499,6 +1399,7 @@ public class TestQueryService extends LensJerseyTest {
"lens.MethodMetricGauge.TestQueryService-testEstimateGauges-PARALLEL_ESTIMATE")),
reg.getGauges().keySet().toString());
}
+
@Test
public void testQueryRejection() throws InterruptedException, IOException {
final WebTarget target = target().path("queryapi/queries");
@@ -1515,4 +1416,40 @@ public class TestQueryService extends LensJerseyTest {
Response response = target.request().post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA_TYPE));
assertEquals(response.getStatus(), 400);
}
+
+ /**
+ * Test query purger
+ *
+ * @throws InterruptedException the interrupted exception
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ @Test
+ public void testQueryPurger() throws InterruptedException, IOException {
+ waitForPurge();
+ LensConf conf = getLensConf(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "false");
+ // test post execute op
+ LensQuery ctx1 = executeAndWaitForQueryToFinish(target(), lensSessionId,
+ "select ID, IDSTR from " + TEST_TABLE,
+ Optional.of(conf), Optional.of(Status.SUCCESSFUL));
+ LensQuery ctx2 = executeAndWaitForQueryToFinish(target(), lensSessionId,
+ "select ID, IDSTR from " + TEST_TABLE,
+ Optional.of(conf), Optional.of(Status.SUCCESSFUL));
+ LensQuery ctx3 = executeAndWaitForQueryToFinish(target(), lensSessionId,
+ "select ID, IDSTR from " + TEST_TABLE,
+ Optional.of(conf), Optional.of(Status.SUCCESSFUL));
+ waitForPurge(3, queryService.finishedQueries);
+ assertEquals(queryService.finishedQueries.size(), 3);
+ getLensQueryResult(target(), lensSessionId, ctx3.getQueryHandle());
+ waitForPurge(2, queryService.finishedQueries);
+ assertTrue(queryService.finishedQueries.size() == 2);
+ getLensQueryResult(target(), lensSessionId, ctx2.getQueryHandle());
+ waitForPurge(1, queryService.finishedQueries);
+ assertTrue(queryService.finishedQueries.size() == 1);
+ getLensQueryResult(target(), lensSessionId, ctx1.getQueryHandle());
+ }
+
+ @AfterMethod
+ private void waitForPurge() throws InterruptedException {
+ waitForPurge(0, queryService.finishedQueries);
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/java/org/apache/lens/server/query/TestResultFormatting.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/query/TestResultFormatting.java b/lens-server/src/test/java/org/apache/lens/server/query/TestResultFormatting.java
index 98a658e..987129c 100644
--- a/lens-server/src/test/java/org/apache/lens/server/query/TestResultFormatting.java
+++ b/lens-server/src/test/java/org/apache/lens/server/query/TestResultFormatting.java
@@ -130,9 +130,9 @@ public class TestResultFormatting extends LensJerseyTest {
conf.addProperty(LensConfConstants.QUERY_OUTPUT_SERDE, LazySimpleSerDe.class.getCanonicalName());
testResultFormatter(conf, QueryStatus.Status.SUCCESSFUL, false, null);
- conf.addProperty(LensConfConstants.RESULT_FS_READ_URL, "filereadurl://");
+ queryService.conf.set(LensConfConstants.RESULT_FS_READ_URL, "filereadurl://");
testResultFormatter(conf, QueryStatus.Status.SUCCESSFUL, false, "filereadurl://");
-
+ queryService.conf.unset(LensConfConstants.RESULT_FS_READ_URL);
}
// test with execute async post with result formatter, get query, get results
@@ -149,8 +149,9 @@ public class TestResultFormatting extends LensJerseyTest {
conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "true");
testResultFormatter(conf, QueryStatus.Status.SUCCESSFUL, false, null);
- conf.addProperty(LensConfConstants.RESULT_FS_READ_URL, "filereadurl://");
+ queryService.conf.set(LensConfConstants.RESULT_FS_READ_URL, "filereadurl://");
testResultFormatter(conf, QueryStatus.Status.SUCCESSFUL, false, "filereadurl://");
+ queryService.conf.unset(LensConfConstants.RESULT_FS_READ_URL);
}
/**
@@ -262,4 +263,8 @@ public class TestResultFormatting extends LensJerseyTest {
}
}
+ @AfterTest
+ public void waitForPurge() throws InterruptedException {
+ waitForPurge(0, queryService.finishedQueries);
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/lens-server/src/test/resources/lens-site.xml
----------------------------------------------------------------------
diff --git a/lens-server/src/test/resources/lens-site.xml b/lens-server/src/test/resources/lens-site.xml
index f728810..cc887ef 100644
--- a/lens-server/src/test/resources/lens-site.xml
+++ b/lens-server/src/test/resources/lens-site.xml
@@ -127,9 +127,9 @@
</property>
<property>
- <!-- Keep only the last query in memory -->
- <name>lens.server.max.finished.queries</name>
- <value>1</value>
+ <!-- run every second -->
+ <name>lens.server.purge.interval</name>
+ <value>1000</value>
</property>
<property>
http://git-wip-us.apache.org/repos/asf/lens/blob/669e8727/src/site/apt/admin/config.apt
----------------------------------------------------------------------
diff --git a/src/site/apt/admin/config.apt b/src/site/apt/admin/config.apt
index 3500441..3a0678f 100644
--- a/src/site/apt/admin/config.apt
+++ b/src/site/apt/admin/config.apt
@@ -101,43 +101,43 @@ Lens server configuration
*--+--+---+--+
|37|lens.server.mail.smtp.timeout|30000|Socket read timeout value in milliseconds. This timeout is implemented by java.net.Socket. Default is 30 seconds.|
*--+--+---+--+
-|38|lens.server.max.finished.queries|100|Maximum number of finished queries which lens server will keep in memory before purging.|
+|38|lens.server.metastore.service.impl|org.apache.lens.server.metastore.CubeMetastoreServiceImpl|Implementation class for metastore service|
*--+--+---+--+
-|39|lens.server.metastore.service.impl|org.apache.lens.server.metastore.CubeMetastoreServiceImpl|Implementation class for metastore service|
+|39|lens.server.metastore.ws.resource.impl|org.apache.lens.server.metastore.MetastoreResource|Implementation class for Metastore Resource|
*--+--+---+--+
-|40|lens.server.metastore.ws.resource.impl|org.apache.lens.server.metastore.MetastoreResource|Implementation class for Metastore Resource|
+|40|lens.server.metrics.csv.directory.path|metrics/|Path of the directory in which to report metrics as separate csv files.|
*--+--+---+--+
-|41|lens.server.metrics.csv.directory.path|metrics/|Path of the directory in which to report metrics as separate csv files.|
+|41|lens.server.metrics.ganglia.host| |The ganglia host name|
*--+--+---+--+
-|42|lens.server.metrics.ganglia.host| |The ganglia host name|
+|42|lens.server.metrics.ganglia.port| |The ganglia port|
*--+--+---+--+
-|43|lens.server.metrics.ganglia.port| |The ganglia port|
+|43|lens.server.metrics.graphite.host| |The graphite host name|
*--+--+---+--+
-|44|lens.server.metrics.graphite.host| |The graphite host name|
+|44|lens.server.metrics.graphite.port| |The graphite port|
*--+--+---+--+
-|45|lens.server.metrics.graphite.port| |The graphite port|
+|45|lens.server.metrics.reporting.period|10|The reporting period for metrics. The value is in seconds|
*--+--+---+--+
-|46|lens.server.metrics.reporting.period|10|The reporting period for metrics. The value is in seconds|
+|46|lens.server.mode|OPEN|The mode in which server should run. Allowed values are OPEN, READ_ONLY, METASTORE_READONLY, METASTORE_NODROP. OPEN mode will allow all requests. READ_ONLY mode will allow all requests on session resouce and only GET requests on all other resources. METASTORE_READONLY will allow GET on metastore and all other requests in other services. METASTORE_NODROP will not allow DELETE on metastore, will allow all other requests.|
*--+--+---+--+
-|47|lens.server.mode|OPEN|The mode in which server should run. Allowed values are OPEN, READ_ONLY, METASTORE_READONLY, METASTORE_NODROP. OPEN mode will allow all requests. READ_ONLY mode will allow all requests on session resouce and only GET requests on all other resources. METASTORE_READONLY will allow GET on metastore and all other requests in other services. METASTORE_NODROP will not allow DELETE on metastore, will allow all other requests.|
+|47|lens.server.multipart.ws.feature.impl|org.glassfish.jersey.media.multipart.MultiPartFeature|Implementation class for query scheduler resource|
*--+--+---+--+
-|48|lens.server.multipart.ws.feature.impl|org.glassfish.jersey.media.multipart.MultiPartFeature|Implementation class for query scheduler resource|
+|48|lens.server.persist.location|file:///tmp/lensserver|The directory in which lens server will persist its state when it is going down. The location be on any Hadoop compatible file system. Server will read from the location when it is restarted and recovery is enabled. So, Server should have both read and write permissions to the location|
*--+--+---+--+
-|49|lens.server.persist.location|file:///tmp/lensserver|The directory in which lens server will persist its state when it is going down. The location be on any Hadoop compatible file system. Server will read from the location when it is restarted and recovery is enabled. So, Server should have both read and write permissions to the location|
+|49|lens.server.query.acceptors| |Query Acceptors configured. Query acceptors are consulted first, before anything happens for the given query. They can either return null or return a messaging indicating why the given query shouldn't be accepted. These can be used to filter out queries at the earliest.|
*--+--+---+--+
-|50|lens.server.query.acceptors| |Query Acceptors configured. Query acceptors are consulted first, before anything happens for the given query. They can either return null or return a messaging indicating why the given query shouldn't be accepted. These can be used to filter out queries at the earliest.|
+|50|lens.server.query.launching.constraint.factories|org.apache.lens.server.query.constraint.TotalQueryCostCeilingConstraintFactory|Factories used to instantiate constraints enforced on queries by lens. Every Factory should be an implementation of org.apache.lens.server.api.common.ConfigBasedObjectCreationFactory and create an implementation of org.apache.lens.server.api.query.constraint.QueryLaunchingConstraint. A query will be launched only if all constraints pass.|
*--+--+---+--+
-|51|lens.server.query.launching.constraint.factories|org.apache.lens.server.query.constraint.TotalQueryCostCeilingConstraintFactory|Factories used to instantiate constraints enforced on queries by lens. Every Factory should be an implementation of org.apache.lens.server.api.common.ConfigBasedObjectCreationFactory and create an implementation of org.apache.lens.server.api.query.constraint.QueryLaunchingConstraint. A query will be launched only if all constraints pass.|
-*--+--+---+--+
-|52|lens.server.query.phase1.rewriters| |Query phase 1 rewriters. This is to convert user query to cube query. The resulting cube query will be passed for validation and rewriting to hql query.\ |
+|51|lens.server.query.phase1.rewriters| |Query phase 1 rewriters. This is to convert user query to cube query. The resulting cube query will be passed for validation and rewriting to hql query.\ |
| | | |Use cases will be to use extra intelligence to convert user query to optimized cube query. \ |
| | | |Or define shortcuts for certain frequently used queries :) |
*--+--+---+--+
-|53|lens.server.query.service.impl|org.apache.lens.server.query.QueryExecutionServiceImpl|Implementation class for query execution service|
+|52|lens.server.query.service.impl|org.apache.lens.server.query.QueryExecutionServiceImpl|Implementation class for query execution service|
+*--+--+---+--+
+|53|lens.server.query.state.logger.enabled|true|Disable or enable the query state logger with this config. The location for the logger can be specified in logback xml for the class org.apache.lens.server.query.QueryExecutionServiceImpl.QueryStatusLogger|
*--+--+---+--+
-|54|lens.server.query.state.logger.enabled|true|Disable or enable the query state logger with this config. The location for the logger can be specified in logback xml for the class org.apache.lens.server.query.QueryExecutionServiceImpl.QueryStatusLogger|
+|54|lens.server.query.ws.resource.impl|org.apache.lens.server.query.QueryServiceResource|Implementation class for Query Resource|
*--+--+---+--+
-|55|lens.server.query.ws.resource.impl|org.apache.lens.server.query.QueryServiceResource|Implementation class for Query Resource|
+|55|lens.server.querypurger.sleep.interval|10000|The interval(milliseconds) with which purger to run periodically. Default 10 sec.|
*--+--+---+--+
|56|lens.server.quota.service.impl|org.apache.lens.server.quota.QuotaServiceImpl|Implementation class for quota service|
*--+--+---+--+
@@ -167,7 +167,7 @@ Lens server configuration
*--+--+---+--+
|69|lens.server.service.provider.factory|org.apache.lens.server.ServiceProviderFactoryImpl|Service provider factory implementation class. This parameter is used to lookup the factory implementation class name that would provide an instance of ServiceProvider. Users should instantiate the class to obtain its instance. Example -- Class spfClass = conf.getClass("lens.server.service.provider.factory", null, ServiceProviderFactory.class); ServiceProviderFactory spf = spfClass.newInstance(); ServiceProvider serviceProvider = spf.getServiceProvider(); -- This is not supposed to be overridden by users.|
*--+--+---+--+
-|70|lens.server.servicenames|session,query,metastore,scheduler,quota,savedquery|These services would be started in the specified order when lens-server starts up|
+|70|lens.server.servicenames|session,query,metastore,scheduler,quota|These services would be started in the specified order when lens-server starts up|
*--+--+---+--+
|71|lens.server.session.expiry.service.interval.secs|3600|Interval at which lens session expiry service runs|
*--+--+---+--+
@@ -237,6 +237,6 @@ Lens server configuration
*--+--+---+--+
|104|lens.server.ws.listenernames|appevent|These listeners would be called in the specified order when lens-server starts up|
*--+--+---+--+
-|105|lens.server.ws.resourcenames|session,metastore,query,quota,scheduler,index,log,savedquery|These JAX-RS resources would be started in the specified order when lens-server starts up|
+|105|lens.server.ws.resourcenames|session,metastore,query,quota,scheduler,index,log|These JAX-RS resources would be started in the specified order when lens-server starts up|
*--+--+---+--+
The configuration parameters and their default values
[2/2] lens git commit: LENS-693 : Queries get purged as soon as they
are finished
Posted by am...@apache.org.
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/master
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);
+ }
+ }
}