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 2014/12/01 10:18:55 UTC
[1/2] incubator-lens git commit: LENS-20 : Sets Driver Specific
Configuration should be set in Query Context (Suma Shivaprasad via
amareshwari)
Repository: incubator-lens
Updated Branches:
refs/heads/master f5e5eccc7 -> fcbb44a14
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/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 a3144f0..8f8764d 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,97 +88,159 @@ import org.codehaus.jackson.map.module.SimpleModule;
*/
public class QueryExecutionServiceImpl extends LensService implements QueryExecutionService {
- /** The Constant LOG. */
+ /**
+ * The Constant LOG.
+ */
public static final Log LOG = LogFactory.getLog(QueryExecutionServiceImpl.class);
- /** The Constant PREPARED_QUERIES_COUNTER. */
+ /**
+ * The Constant PREPARED_QUERIES_COUNTER.
+ */
public static final String PREPARED_QUERIES_COUNTER = "prepared-queries";
- /** The Constant QUERY_SUBMITTER_COUNTER. */
+ /**
+ * The Constant QUERY_SUBMITTER_COUNTER.
+ */
public static final String QUERY_SUBMITTER_COUNTER = "query-submitter-errors";
- /** The Constant STATUS_UPDATE_COUNTER. */
+ /**
+ * The Constant STATUS_UPDATE_COUNTER.
+ */
public static final String STATUS_UPDATE_COUNTER = "status-update-errors";
- /** The Constant QUERY_PURGER_COUNTER. */
+ /**
+ * The Constant QUERY_PURGER_COUNTER.
+ */
public static final String QUERY_PURGER_COUNTER = "query-purger-errors";
- /** The Constant PREPARED_QUERY_PURGER_COUNTER. */
+ /**
+ * The Constant PREPARED_QUERY_PURGER_COUNTER.
+ */
public static final String PREPARED_QUERY_PURGER_COUNTER = "prepared-query-purger-errors";
- /** The millis in week. */
+ /**
+ * The millis in week.
+ */
private static long millisInWeek = 7 * 24 * 60 * 60 * 1000;
- /** The Constant NAME. */
+ /**
+ * The Constant NAME.
+ */
public static final String NAME = "query";
- /** The Constant mapper. */
+ /**
+ * The Constant mapper.
+ */
private static final ObjectMapper mapper = new ObjectMapper();
- /** The accepted queries. */
+ /**
+ * The accepted queries.
+ */
private PriorityBlockingQueue<QueryContext> acceptedQueries = new PriorityBlockingQueue<QueryContext>();
- /** The launched queries. */
+ /**
+ * The launched queries.
+ */
private List<QueryContext> launchedQueries = new ArrayList<QueryContext>();
- /** The finished queries. */
+ /**
+ * The finished queries.
+ */
private DelayQueue<FinishedQuery> finishedQueries = new DelayQueue<FinishedQuery>();
- /** The prepared query queue. */
+ /**
+ * The prepared query queue.
+ */
private DelayQueue<PreparedQueryContext> preparedQueryQueue = new DelayQueue<PreparedQueryContext>();
- /** The prepared queries. */
+ /**
+ * The prepared queries.
+ */
private Map<QueryPrepareHandle, PreparedQueryContext> preparedQueries = new HashMap<QueryPrepareHandle, PreparedQueryContext>();
- /** The all queries. */
+ /**
+ * The all queries.
+ */
private ConcurrentMap<QueryHandle, QueryContext> allQueries = new ConcurrentHashMap<QueryHandle, QueryContext>();
- /** The conf. */
+ /**
+ * The conf.
+ */
private Configuration conf;
- /** The query submitter runnable. */
+ /**
+ * The query submitter runnable.
+ */
private final QuerySubmitter querySubmitterRunnable = new QuerySubmitter();
- /** The query submitter. */
+ /**
+ * The query submitter.
+ */
protected final Thread querySubmitter = new Thread(querySubmitterRunnable, "QuerySubmitter");
- /** The status poller. */
+ /**
+ * The status poller.
+ */
private final Thread statusPoller = new Thread(new StatusPoller(), "StatusPoller");
- /** The query purger. */
+ /**
+ * The query purger.
+ */
private final Thread queryPurger = new Thread(new QueryPurger(), "QueryPurger");
- /** The prepare query purger. */
+ /**
+ * The prepare query purger.
+ */
private final Thread prepareQueryPurger = new Thread(new PreparedQueryPurger(), "PrepareQueryPurger");
- /** The query acceptors. */
+ /**
+ * The query acceptors.
+ */
private List<QueryAcceptor> queryAcceptors = new ArrayList<QueryAcceptor>();
- /** The drivers. */
+ /**
+ * The drivers.
+ */
private final Map<String, LensDriver> drivers = new HashMap<String, LensDriver>();
- /** The driver selector. */
+ /**
+ * The driver selector.
+ */
private DriverSelector driverSelector;
- /** The result sets. */
+ /**
+ * The result sets.
+ */
private Map<QueryHandle, LensResultSet> resultSets = new HashMap<QueryHandle, LensResultSet>();
- /** The event service. */
+ /**
+ * The event service.
+ */
private LensEventService eventService;
- /** The metrics service. */
+ /**
+ * The metrics service.
+ */
private MetricsService metricsService;
- /** The statistics service. */
+ /**
+ * The statistics service.
+ */
private StatisticsService statisticsService;
- /** The max finished queries. */
+ /**
+ * The max finished queries.
+ */
private int maxFinishedQueries;
- /** The lens server dao. */
+ /**
+ * The lens server dao.
+ */
LensServerDAO lensServerDao;
- /** The driver event listener. */
+ /**
+ * The driver event listener.
+ */
final LensEventListener<DriverEvent> driverEventListener = new LensEventListener<DriverEvent>() {
@Override
public void onEvent(DriverEvent event) {
@@ -193,10 +255,8 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Instantiates a new query execution service impl.
*
- * @param cliService
- * the cli service
- * @throws LensException
- * the lens exception
+ * @param cliService the cli service
+ * @throws LensException the lens exception
*/
public QueryExecutionServiceImpl(CLIService cliService) throws LensException {
super(NAME, cliService);
@@ -213,7 +273,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
// Add result formatter
getEventService().addListenerForType(new ResultFormatter(this), QueryExecuted.class);
getEventService().addListenerForType(new QueryExecutionStatisticsGenerator(this, getEventService()),
- QueryEnded.class);
+ QueryEnded.class);
getEventService().addListenerForType(new QueryEndNotifier(this, getCliService().getHiveConf()), QueryEnded.class);
LOG.info("Registered query result formatter");
}
@@ -221,8 +281,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Load drivers and selector.
*
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private void loadDriversAndSelector() throws LensException {
conf.get(LensConfConstants.DRIVER_CLASSES);
@@ -284,8 +343,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Incr counter.
*
- * @param counter
- * the counter
+ * @param counter the counter
*/
private void incrCounter(String counter) {
getMetrics().incrCounter(QueryExecutionService.class, counter);
@@ -294,8 +352,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Decr counter.
*
- * @param counter
- * the counter
+ * @param counter the counter
*/
private void decrCounter(String counter) {
getMetrics().decrCounter(QueryExecutionService.class, counter);
@@ -306,7 +363,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
public static class QueryStatusLogger implements LensEventListener<StatusChange> {
- /** The Constant STATUS_LOG. */
+ /**
+ * The Constant STATUS_LOG.
+ */
public static final Log STATUS_LOG = LogFactory.getLog(QueryStatusLogger.class);
/*
@@ -325,17 +384,20 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
private class FinishedQuery implements Delayed {
- /** The ctx. */
+ /**
+ * The ctx.
+ */
private final QueryContext ctx;
- /** The finish time. */
+ /**
+ * The finish time.
+ */
private final Date finishTime;
/**
* Instantiates a new finished query.
*
- * @param ctx
- * the ctx
+ * @param ctx the ctx
*/
FinishedQuery(QueryContext ctx) {
this.ctx = ctx;
@@ -388,7 +450,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
private class QuerySubmitter implements Runnable {
- /** The paused for test. */
+ /**
+ * The paused for test.
+ */
private boolean pausedForTest = false;
/*
@@ -404,7 +468,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
QueryContext ctx = acceptedQueries.take();
synchronized (ctx) {
if (ctx.getStatus().getStatus().equals(Status.QUEUED)) {
- LOG.info("Launching query:" + ctx.getDriverQuery());
+ LOG.info("Launching query:" + ctx.getSelectedDriverQuery());
try {
// acquire session before any query operation.
acquire(ctx.getLensSessionIdentifier());
@@ -440,6 +504,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
}
// used in tests
+
/**
* Pause query submitter.
*/
@@ -452,7 +517,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
private class StatusPoller implements Runnable {
- /** The poll interval. */
+ /**
+ * The poll interval.
+ */
long pollInterval = 1000;
/*
@@ -498,14 +565,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Sets the failed status.
*
- * @param ctx
- * the ctx
- * @param statusMsg
- * the status msg
- * @param reason
- * the reason
- * @throws LensException
- * the lens exception
+ * @param ctx the ctx
+ * @param statusMsg the status msg
+ * @param reason the reason
+ * @throws LensException the lens exception
*/
void setFailedStatus(QueryContext ctx, String statusMsg, String reason) throws LensException {
QueryStatus before = ctx.getStatus();
@@ -517,7 +580,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
private void setLaunchedStatus(QueryContext ctx) throws LensException {
QueryStatus before = ctx.getStatus();
ctx.setStatus(new QueryStatus(ctx.getStatus().getProgress(), QueryStatus.Status.LAUNCHED, "launched on the driver",
- false, null, null));
+ false, null, null));
launchedQueries.add(ctx);
ctx.setLaunchTime(System.currentTimeMillis());
fireStatusChangeEvent(ctx, ctx.getStatus(), before);
@@ -526,12 +589,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Sets the cancelled status.
*
- * @param ctx
- * the ctx
- * @param statusMsg
- * the status msg
- * @throws LensException
- * the lens exception
+ * @param ctx the ctx
+ * @param statusMsg the status msg
+ * @throws LensException the lens exception
*/
private void setCancelledStatus(QueryContext ctx, String statusMsg) throws LensException {
QueryStatus before = ctx.getStatus();
@@ -543,10 +603,8 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Update finished query.
*
- * @param ctx
- * the ctx
- * @param before
- * the before
+ * @param ctx the ctx
+ * @param before the before
*/
private void updateFinishedQuery(QueryContext ctx, QueryStatus before) {
// before would be null in case of server restart
@@ -563,7 +621,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
void setSuccessState(QueryContext ctx) throws LensException {
QueryStatus before = ctx.getStatus();
ctx.setStatus(new QueryStatus(1.0f, QueryStatus.Status.SUCCESSFUL, "Query is successful!", ctx
- .isResultAvailableInDriver(), null, null));
+ .isResultAvailableInDriver(), null, null));
updateFinishedQuery(ctx, before);
fireStatusChangeEvent(ctx, ctx.getStatus(), before);
}
@@ -571,10 +629,8 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Update status.
*
- * @param handle
- * the handle
- * @throws LensException
- * the lens exception
+ * @param handle the handle
+ * @throws LensException the lens exception
*/
private void updateStatus(final QueryHandle handle) throws LensException {
QueryContext ctx = allQueries.get(handle);
@@ -582,7 +638,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
synchronized (ctx) {
QueryStatus before = ctx.getStatus();
if (!ctx.getStatus().getStatus().equals(QueryStatus.Status.QUEUED) && !ctx.getDriverStatus().isFinished()
- && !ctx.getStatus().isFinished()) {
+ && !ctx.getStatus().isFinished()) {
LOG.info("Updating status for " + ctx.getQueryHandle());
try {
ctx.getSelectedDriver().updateStatus(ctx);
@@ -610,12 +666,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* New status change event.
*
- * @param ctx
- * the ctx
- * @param prevState
- * the prev state
- * @param currState
- * the curr state
+ * @param ctx the ctx
+ * @param prevState the prev state
+ * @param currState the curr state
* @return the status change
*/
private static StatusChange newStatusChangeEvent(QueryContext ctx, QueryStatus.Status prevState, QueryStatus.Status currState) {
@@ -634,7 +687,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
return new QueryQueued(ctx.getSubmissionTime(), prevState, currState, query, ctx.getSubmittedUser());
case RUNNING:
return new QueryRunning(System.currentTimeMillis() - ctx.getDriverStatus().getDriverStartTime(), prevState,
- currState, query);
+ currState, query);
case EXECUTED:
return new QueryExecuted(ctx.getDriverStatus().getDriverFinishTime(), prevState, currState, query);
case SUCCESSFUL:
@@ -648,12 +701,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* If query status has changed, fire a specific StatusChange event.
*
- * @param ctx
- * the ctx
- * @param current
- * the current
- * @param before
- * the before
+ * @param ctx the ctx
+ * @param current the current
+ * @param before the before
*/
private void fireStatusChangeEvent(QueryContext ctx, QueryStatus current, QueryStatus before) {
if (ctx == null || current == null) {
@@ -736,7 +786,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
resultSets.remove(finished.getCtx().getQueryHandle());
}
fireStatusChangeEvent(finished.getCtx(),
- new QueryStatus(1f, Status.CLOSED, "Query purged", false, null, null), finished.getCtx().getStatus());
+ new QueryStatus(1f, Status.CLOSED, "Query purged", false, null, null), finished.getCtx().getStatus());
LOG.info("Query purged: " + finished.getCtx().getQueryHandle());
} catch (LensException e) {
incrCounter(QUERY_PURGER_COUNTER);
@@ -798,7 +848,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
throw new IllegalStateException("Could not load drivers");
}
maxFinishedQueries = conf.getInt(LensConfConstants.MAX_NUMBER_OF_FINISHED_QUERY,
- LensConfConstants.DEFAULT_FINISHED_QUERIES);
+ LensConfConstants.DEFAULT_FINISHED_QUERIES);
initalizeFinishedQueryStore(conf);
LOG.info("Query execution service initialized");
}
@@ -806,8 +856,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Initalize finished query store.
*
- * @param conf
- * the conf
+ * @param conf the conf
*/
private void initalizeFinishedQueryStore(Configuration conf) {
this.lensServerDao = new LensServerDAO();
@@ -821,7 +870,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
module.addSerializer(ColumnDescriptor.class, new JsonSerializer<ColumnDescriptor>() {
@Override
public void serialize(ColumnDescriptor columnDescriptor, JsonGenerator jsonGenerator,
- SerializerProvider serializerProvider) throws IOException, JsonProcessingException {
+ SerializerProvider serializerProvider) throws IOException, JsonProcessingException {
jsonGenerator.writeStartObject();
jsonGenerator.writeStringField("name", columnDescriptor.getName());
jsonGenerator.writeStringField("comment", columnDescriptor.getComment());
@@ -833,12 +882,12 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
module.addDeserializer(ColumnDescriptor.class, new JsonDeserializer<ColumnDescriptor>() {
@Override
public ColumnDescriptor deserialize(JsonParser jsonParser, DeserializationContext deserializationContext)
- throws IOException, JsonProcessingException {
+ throws IOException, JsonProcessingException {
ObjectCodec oc = jsonParser.getCodec();
JsonNode node = oc.readTree(jsonParser);
org.apache.hive.service.cli.Type t = org.apache.hive.service.cli.Type.getType(node.get("type").asText());
return new ColumnDescriptor(node.get("name").asText(), node.get("comment").asText(), new TypeDescriptor(t),
- node.get("position").asInt());
+ node.get("position").asInt());
}
});
mapper.registerModule(module);
@@ -864,7 +913,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
public synchronized void stop() {
super.stop();
- for (Thread th : new Thread[] { querySubmitter, statusPoller, queryPurger, prepareQueryPurger }) {
+ for (Thread th : new Thread[]{querySubmitter, statusPoller, queryPurger, prepareQueryPurger}) {
try {
LOG.debug("Waiting for" + th.getName());
th.join();
@@ -887,12 +936,17 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
try {
if (sessionMap.containsKey(ctx.getLensSessionIdentifier())) {
// try setting configuration if the query session is still not closed
- ctx.setConf(getLensConf(getSessionHandle(ctx.getLensSessionIdentifier()), ctx.getQconf()));
+ ctx.setConf(getLensConf(getSessionHandle(ctx.getLensSessionIdentifier()), ctx.getLensConf()));
} else {
- ctx.setConf(getLensConf(ctx.getQconf()));
+ ctx.setConf(getLensConf(ctx.getLensConf()));
+ }
+ for(LensDriver driver : drivers.values()) {
+ if(ctx.getDriverContext() != null) {
+ ctx.getDriverContext().setDriverConf(driver, ctx.getConf());
+ }
}
} catch (LensException e) {
- LOG.error("Could not set query conf");
+ LOG.error("Could not set query conf " , e);
}
}
}
@@ -906,33 +960,25 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Rewrite and select.
*
- * @param ctx
- * the ctx
- * @throws LensException
- * the lens exception
+ * @param ctx the ctx
+ * @throws LensException the lens exception
*/
private void rewriteAndSelect(AbstractQueryContext ctx) throws LensException {
- ctx.setDriverQueriesAndPlans(RewriteUtil.rewriteQuery(ctx.getUserQuery(), drivers.values(),
- ctx.getConf()));
+ ctx.getDriverContext().setDriverQueriesAndPlans(RewriteUtil.rewriteQuery(ctx));
// 2. select driver to run the query
LensDriver driver = driverSelector.select(ctx, conf);
ctx.setSelectedDriver(driver);
- ctx.setDriverQuery(ctx.getDriverQueries().get(ctx.getSelectedDriver()));
}
/**
* Accept.
*
- * @param query
- * the query
- * @param conf
- * the conf
- * @param submitOp
- * the submit op
- * @throws LensException
- * the lens exception
+ * @param query the query
+ * @param conf the conf
+ * @param submitOp the submit op
+ * @throws LensException the lens exception
*/
private void accept(String query, Configuration conf, SubmitOp submitOp) throws LensException {
// run through all the query acceptors, and throw Exception if any of them
@@ -951,11 +997,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Gets the resultset from dao.
*
- * @param queryHandle
- * the query handle
+ * @param queryHandle the query handle
* @return the resultset from dao
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private LensResultSet getResultsetFromDAO(QueryHandle queryHandle) throws LensException {
FinishedLensQuery query = lensServerDao.getQuery(queryHandle.toString());
@@ -966,7 +1010,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
try {
Class<LensResultSetMetadata> mdKlass = (Class<LensResultSetMetadata>) Class.forName(query.getMetadataClass());
return new LensPersistentResult(mapper.readValue(query.getMetadata(), mdKlass), query.getResult(),
- query.getRows());
+ query.getRows());
} catch (Exception e) {
throw new LensException(e);
}
@@ -977,11 +1021,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Gets the resultset.
*
- * @param queryHandle
- * the query handle
+ * @param queryHandle the query handle
* @return the resultset
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private LensResultSet getResultset(QueryHandle queryHandle) throws LensException {
QueryContext ctx = allQueries.get(queryHandle);
@@ -996,11 +1038,12 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
if (resultSet == null) {
if (ctx.isPersistent() && ctx.getQueryOutputFormatter() != null) {
resultSets
- .put(queryHandle, new LensPersistentResult(ctx.getQueryOutputFormatter().getMetadata(), ctx
- .getQueryOutputFormatter().getFinalOutputPath().toString(), ctx.getQueryOutputFormatter()
- .getNumRows()));
+ .put(queryHandle, new LensPersistentResult(ctx.getQueryOutputFormatter().getMetadata(), ctx
+ .getQueryOutputFormatter().getFinalOutputPath().toString(), ctx.getQueryOutputFormatter()
+ .getNumRows()));
} else if (allQueries.get(queryHandle).isResultAvailableInDriver()) {
- resultSet = allQueries.get(queryHandle).getSelectedDriver().fetchResultSet(allQueries.get(queryHandle));
+ resultSet = allQueries.get(queryHandle).getSelectedDriver().fetchResultSet(allQueries
+ .get(queryHandle));
resultSets.put(queryHandle, resultSet);
} else {
throw new NotFoundException("Result set not available for query:" + queryHandle);
@@ -1014,14 +1057,13 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Gets the driver resultset.
*
- * @param queryHandle
- * the query handle
+ * @param queryHandle the query handle
* @return the driver resultset
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
LensResultSet getDriverResultset(QueryHandle queryHandle) throws LensException {
- return allQueries.get(queryHandle).getSelectedDriver().fetchResultSet(allQueries.get(queryHandle));
+ return allQueries.get(queryHandle).getSelectedDriver().fetchResultSet(allQueries.get
+ (queryHandle));
}
/*
@@ -1032,7 +1074,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryPrepareHandle prepare(LensSessionHandle sessionHandle, String query, LensConf lensConf, String queryName)
- throws LensException {
+ throws LensException {
try {
acquire(sessionHandle);
PreparedQueryContext prepared = prepareQuery(sessionHandle, query, lensConf, SubmitOp.PREPARE);
@@ -1047,24 +1089,19 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Prepare query.
*
- * @param sessionHandle
- * the session handle
- * @param query
- * the query
- * @param lensConf
- * the lens conf
- * @param op
- * the op
+ * @param sessionHandle the session handle
+ * @param query the query
+ * @param lensConf the lens conf
+ * @param op the op
* @return the prepared query context
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private PreparedQueryContext prepareQuery(LensSessionHandle sessionHandle, String query, LensConf lensConf,
- SubmitOp op) throws LensException {
+ SubmitOp op) throws LensException {
Configuration conf = getLensConf(sessionHandle, lensConf);
accept(query, conf, op);
PreparedQueryContext prepared = new PreparedQueryContext(query, getSession(sessionHandle).getLoggedInUser(), conf,
- lensConf);
+ lensConf, drivers.values());
rewriteAndSelect(prepared);
preparedQueries.put(prepared.getPrepareHandle(), prepared);
preparedQueryQueue.add(prepared);
@@ -1081,7 +1118,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryPlan explainAndPrepare(LensSessionHandle sessionHandle, String query, LensConf lensConf, String queryName)
- throws LensException {
+ throws LensException {
try {
LOG.info("ExlainAndPrepare: " + sessionHandle.toString() + " query: " + query);
acquire(sessionHandle);
@@ -1115,7 +1152,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryHandle executePrepareAsync(LensSessionHandle sessionHandle, QueryPrepareHandle prepareHandle,
- LensConf conf, String queryName) throws LensException {
+ LensConf conf, String queryName) throws LensException {
try {
LOG.info("ExecutePrepareAsync: " + sessionHandle.toString() + " query:" + prepareHandle.getPrepareHandleId());
acquire(sessionHandle);
@@ -1143,10 +1180,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryHandleWithResultSet executePrepare(LensSessionHandle sessionHandle, QueryPrepareHandle prepareHandle,
- long timeoutMillis, LensConf conf, String queryName) throws LensException {
+ long timeoutMillis, LensConf conf, String queryName) throws LensException {
try {
LOG.info("ExecutePrepare: " + sessionHandle.toString() + " query:" + prepareHandle.getPrepareHandleId()
- + " timeout:" + timeoutMillis);
+ + " timeout:" + timeoutMillis);
acquire(sessionHandle);
PreparedQueryContext pctx = getPreparedQueryContext(sessionHandle, prepareHandle);
Configuration qconf = getLensConf(sessionHandle, conf);
@@ -1171,7 +1208,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryHandle executeAsync(LensSessionHandle sessionHandle, String query, LensConf conf, String queryName)
- throws LensException {
+ throws LensException {
try {
LOG.info("ExecuteAsync: " + sessionHandle.toString() + " query: " + query);
acquire(sessionHandle);
@@ -1188,41 +1225,31 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Creates the context.
*
- * @param query
- * the query
- * @param userName
- * the user name
- * @param conf
- * the conf
- * @param qconf
- * the qconf
+ * @param query the query
+ * @param userName the user name
+ * @param conf the conf
+ * @param qconf the qconf
* @return the query context
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
protected QueryContext createContext(String query, String userName, LensConf conf, Configuration qconf)
- throws LensException {
- QueryContext ctx = new QueryContext(query, userName, conf, qconf);
+ throws LensException {
+ QueryContext ctx = new QueryContext(query, userName, conf, qconf, drivers.values());
return ctx;
}
/**
* Creates the context.
*
- * @param pctx
- * the pctx
- * @param userName
- * the user name
- * @param conf
- * the conf
- * @param qconf
- * the qconf
+ * @param pctx the pctx
+ * @param userName the user name
+ * @param conf the conf
+ * @param qconf the qconf
* @return the query context
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
protected QueryContext createContext(PreparedQueryContext pctx, String userName, LensConf conf, Configuration qconf)
- throws LensException {
+ throws LensException {
QueryContext ctx = new QueryContext(pctx, userName, conf, qconf);
return ctx;
}
@@ -1230,13 +1257,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Execute async internal.
*
- * @param sessionHandle
- * the session handle
- * @param ctx
- * the ctx
+ * @param sessionHandle the session handle
+ * @param ctx the ctx
* @return the query handle
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private QueryHandle executeAsyncInternal(LensSessionHandle sessionHandle, QueryContext ctx) throws LensException {
ctx.setLensSessionIdentifier(sessionHandle.getPublicId().toString());
@@ -1257,7 +1281,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public boolean updateQueryConf(LensSessionHandle sessionHandle, QueryHandle queryHandle, LensConf newconf)
- throws LensException {
+ throws LensException {
try {
LOG.info("UpdateQueryConf:" + sessionHandle.toString() + " query: " + queryHandle);
acquire(sessionHandle);
@@ -1282,7 +1306,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public boolean updateQueryConf(LensSessionHandle sessionHandle, QueryPrepareHandle prepareHandle, LensConf newconf)
- throws LensException {
+ throws LensException {
try {
LOG.info("UpdatePreparedQueryConf:" + sessionHandle.toString() + " query: " + prepareHandle);
acquire(sessionHandle);
@@ -1297,13 +1321,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Gets the query context.
*
- * @param sessionHandle
- * the session handle
- * @param queryHandle
- * the query handle
+ * @param sessionHandle the session handle
+ * @param queryHandle the query handle
* @return the query context
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private QueryContext getQueryContext(LensSessionHandle sessionHandle, QueryHandle queryHandle) throws LensException {
try {
@@ -1315,11 +1336,11 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
throw new NotFoundException("Query not found " + queryHandle);
}
QueryContext finishedCtx = new QueryContext(query.getUserQuery(), query.getSubmitter(), conf,
- query.getSubmissionTime());
+ drivers.values(), query.getSubmissionTime());
finishedCtx.setQueryHandle(queryHandle);
finishedCtx.setEndTime(query.getEndTime());
finishedCtx.setStatusSkippingTransitionTest(new QueryStatus(0.0, QueryStatus.Status.valueOf(query.getStatus()),
- query.getErrorMessage() == null ? "" : query.getErrorMessage(), query.getResult() != null, null, null));
+ query.getErrorMessage() == null ? "" : query.getErrorMessage(), query.getResult() != null, null, null));
finishedCtx.getDriverStatus().setDriverStartTime(query.getDriverStartTime());
finishedCtx.getDriverStatus().setDriverFinishTime(query.getDriverEndTime());
finishedCtx.setResultSetPath(query.getResult());
@@ -1336,8 +1357,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Gets the query context.
*
- * @param queryHandle
- * the query handle
+ * @param queryHandle the query handle
* @return the query context
*/
QueryContext getQueryContext(QueryHandle queryHandle) {
@@ -1358,16 +1378,13 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Gets the prepared query context.
*
- * @param sessionHandle
- * the session handle
- * @param prepareHandle
- * the prepare handle
+ * @param sessionHandle the session handle
+ * @param prepareHandle the prepare handle
* @return the prepared query context
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private PreparedQueryContext getPreparedQueryContext(LensSessionHandle sessionHandle, QueryPrepareHandle prepareHandle)
- throws LensException {
+ throws LensException {
try {
acquire(sessionHandle);
PreparedQueryContext ctx = preparedQueries.get(prepareHandle);
@@ -1388,7 +1405,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public LensPreparedQuery getPreparedQuery(LensSessionHandle sessionHandle, QueryPrepareHandle prepareHandle)
- throws LensException {
+ throws LensException {
return getPreparedQueryContext(sessionHandle, prepareHandle).toPreparedQuery();
}
@@ -1400,7 +1417,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryHandleWithResultSet execute(LensSessionHandle sessionHandle, String query, long timeoutMillis,
- LensConf conf, String queryName) throws LensException {
+ LensConf conf, String queryName) throws LensException {
try {
LOG.info("Blocking execute " + sessionHandle.toString() + " query: " + query + " timeout: " + timeoutMillis);
acquire(sessionHandle);
@@ -1417,20 +1434,15 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Execute timeout internal.
*
- * @param sessionHandle
- * the session handle
- * @param ctx
- * the ctx
- * @param timeoutMillis
- * the timeout millis
- * @param conf
- * the conf
+ * @param sessionHandle the session handle
+ * @param ctx the ctx
+ * @param timeoutMillis the timeout millis
+ * @param conf the conf
* @return the query handle with result set
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private QueryHandleWithResultSet executeTimeoutInternal(LensSessionHandle sessionHandle, QueryContext ctx,
- long timeoutMillis, Configuration conf) throws LensException {
+ long timeoutMillis, Configuration conf) throws LensException {
QueryHandle handle = executeAsyncInternal(sessionHandle, ctx);
QueryHandleWithResultSet result = new QueryHandleWithResultSet(handle);
// getQueryContext calls updateStatus, which fires query events if there's a change in status
@@ -1442,8 +1454,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
}
}
QueryCompletionListener listener = new QueryCompletionListenerImpl(handle);
- getQueryContext(sessionHandle, handle).getSelectedDriver().registerForCompletionNotification(handle, timeoutMillis,
- listener);
+ getQueryContext(sessionHandle, handle).getSelectedDriver().registerForCompletionNotification
+ (handle,
+ timeoutMillis,
+ listener);
try {
synchronized (listener) {
listener.wait(timeoutMillis);
@@ -1463,17 +1477,20 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
class QueryCompletionListenerImpl implements QueryCompletionListener {
- /** The succeeded. */
+ /**
+ * The succeeded.
+ */
boolean succeeded = false;
- /** The handle. */
+ /**
+ * The handle.
+ */
QueryHandle handle;
/**
* Instantiates a new query completion listener impl.
*
- * @param handle
- * the handle
+ * @param handle the handle
*/
QueryCompletionListenerImpl(QueryHandle handle) {
this.handle = handle;
@@ -1519,7 +1536,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryResultSetMetadata getResultSetMetadata(LensSessionHandle sessionHandle, QueryHandle queryHandle)
- throws LensException {
+ throws LensException {
try {
LOG.info("GetResultSetMetadata: " + sessionHandle.toString() + " query: " + queryHandle);
acquire(sessionHandle);
@@ -1528,7 +1545,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
return resultSet.getMetadata().toQueryResultSetMetadata();
} else {
throw new NotFoundException("Resultset metadata not found for query: (" + sessionHandle + ", " + queryHandle
- + ")");
+ + ")");
}
} finally {
release(sessionHandle);
@@ -1543,7 +1560,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public QueryResult fetchResultSet(LensSessionHandle sessionHandle, QueryHandle queryHandle, long startIndex,
- int fetchSize) throws LensException {
+ int fetchSize) throws LensException {
try {
LOG.info("FetchResultSet:" + sessionHandle.toString() + " query:" + queryHandle);
acquire(sessionHandle);
@@ -1589,7 +1606,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
}
synchronized (ctx) {
if (ctx.getStatus().getStatus().equals(QueryStatus.Status.LAUNCHED)
- || ctx.getStatus().getStatus().equals(QueryStatus.Status.RUNNING)) {
+ || ctx.getStatus().getStatus().equals(QueryStatus.Status.RUNNING)) {
boolean ret = ctx.getSelectedDriver().cancelQuery(queryHandle);
if (!ret) {
return false;
@@ -1612,7 +1629,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public List<QueryHandle> getAllQueries(LensSessionHandle sessionHandle, String state, String userName,
- String queryName, long fromDate, long toDate) throws LensException {
+ String queryName, long fromDate, long toDate) throws LensException {
validateTimeRange(fromDate, toDate);
userName = UtilityMethods.removeDomain(userName);
try {
@@ -1638,9 +1655,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
QueryContext context = allQueries.get(q);
long querySubmitTime = context.getSubmissionTime();
if ((filterByStatus && status != context.getStatus().getStatus())
- || (filterByQueryName && !context.getQueryName().toLowerCase().contains(queryName))
- || (!"all".equalsIgnoreCase(userName) && !userName.equalsIgnoreCase(context.getSubmittedUser()))
- || (!(fromDate <= querySubmitTime && querySubmitTime <= toDate))) {
+ || (filterByQueryName && !context.getQueryName().toLowerCase().contains(queryName))
+ || (!"all".equalsIgnoreCase(userName) && !userName.equalsIgnoreCase(context.getSubmittedUser()))
+ || (!(fromDate <= querySubmitTime && querySubmitTime <= toDate))) {
itr.remove();
}
}
@@ -1651,7 +1668,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
userName = null;
}
List<QueryHandle> persistedQueries = lensServerDao.findFinishedQueries(state, userName, queryName, fromDate,
- toDate);
+ toDate);
if (persistedQueries != null && !persistedQueries.isEmpty()) {
LOG.info("Adding persisted queries " + persistedQueries.size());
all.addAll(persistedQueries);
@@ -1673,7 +1690,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
*/
@Override
public List<QueryPrepareHandle> getAllPreparedQueries(LensSessionHandle sessionHandle, String user, String queryName,
- long fromDate, long toDate) throws LensException {
+ long fromDate, long toDate) throws LensException {
validateTimeRange(fromDate, toDate);
user = UtilityMethods.removeDomain(user);
try {
@@ -1713,10 +1730,8 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Validate time range.
*
- * @param fromDate
- * the from date
- * @param toDate
- * the to date
+ * @param fromDate the from date
+ * @param toDate the to date
*/
private void validateTimeRange(long fromDate, long toDate) {
if (fromDate >= toDate) {
@@ -1745,10 +1760,8 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Destroy prepared query.
*
- * @param ctx
- * the ctx
- * @throws LensException
- * the lens exception
+ * @param ctx the ctx
+ * @throws LensException the lens exception
*/
private void destroyPreparedQuery(PreparedQueryContext ctx) throws LensException {
ctx.getSelectedDriver().closePreparedQuery(ctx.getPrepareHandle());
@@ -1769,9 +1782,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
LOG.info("Explain: " + sessionHandle.toString() + " query:" + query);
acquire(sessionHandle);
Configuration qconf = getLensConf(sessionHandle, lensConf);
- ExplainQueryContext explainQueryContext = new ExplainQueryContext(query, qconf);
+ ExplainQueryContext explainQueryContext = new ExplainQueryContext(query, lensConf, qconf, drivers.values());
+
accept(query, qconf, SubmitOp.EXPLAIN);
- explainQueryContext.setDriverQueriesAndPlans(RewriteUtil.rewriteQuery(query, drivers.values(), qconf));
+ explainQueryContext.getDriverContext().setDriverQueriesAndPlans(RewriteUtil.rewriteQuery(explainQueryContext));
// select driver to run the query
explainQueryContext.setSelectedDriver(driverSelector.select(explainQueryContext, qconf));
return explainQueryContext.getSelectedDriverQueryPlan().toQueryPlan();
@@ -1812,23 +1826,19 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Creates the add resource query.
*
- * @param sessionHandle
- * the session handle
- * @param type
- * the type
- * @param path
- * the path
+ * @param sessionHandle the session handle
+ * @param type the type
+ * @param path the path
* @return the query context
- * @throws LensException
- * the lens exception
+ * @throws LensException the lens exception
*/
private QueryContext createAddResourceQuery(LensSessionHandle sessionHandle, String type, String path)
- throws LensException {
+ throws LensException {
String command = "add " + type.toLowerCase() + " " + path;
LensConf conf = new LensConf();
conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "false");
- QueryContext addQuery = new QueryContext(command, getSession(sessionHandle).getLoggedInUser(), getLensConf(
- sessionHandle, conf));
+ QueryContext addQuery = new QueryContext(command, getSession(sessionHandle).getLoggedInUser(), conf, getLensConf(
+ sessionHandle, conf), drivers.values());
addQuery.setLensSessionIdentifier(sessionHandle.getPublicId().toString());
return addQuery;
}
@@ -1848,7 +1858,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
LensConf conf = new LensConf();
conf.addProperty(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, "false");
QueryContext addQuery = new QueryContext(command, getSession(sessionHandle).getLoggedInUser(), getLensConf(
- sessionHandle, conf));
+ sessionHandle, conf), drivers.values());
addQuery.setLensSessionIdentifier(sessionHandle.getPublicId().toString());
driver.execute(addQuery);
}
@@ -1895,10 +1905,15 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
for (int i = 0; i < numQueries; i++) {
QueryContext ctx = (QueryContext) in.readObject();
+
+ //Create Driver Selector Context with driver conf for now and reset it in start()
+ DriverSelectorQueryContext driverCtx = new DriverSelectorQueryContext(ctx.getDriverQuery(), new Configuration(),
+ drivers.values());
+ ctx.setDriverContext(driverCtx);
boolean driverAvailable = in.readBoolean();
if (driverAvailable) {
String clsName = in.readUTF();
- ctx.setSelectedDriver(drivers.get(clsName));
+ ctx.getDriverContext().setSelectedDriver(drivers.get(clsName));
}
allQueries.put(ctx.getQueryHandle(), ctx);
}
@@ -1961,12 +1976,9 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Pipe.
*
- * @param is
- * the is
- * @param os
- * the os
- * @throws IOException
- * Signals that an I/O exception has occurred.
+ * @param is the is
+ * @param os the os
+ * @throws IOException Signals that an I/O exception has occurred.
*/
private void pipe(InputStream is, OutputStream os) throws IOException {
int n;
@@ -2002,10 +2014,10 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
if (resultFSReadUrl != null) {
try {
URI resultReadPath = new URI(resultFSReadUrl + resultPath.toUri().getPath() + "?op=OPEN&user.name="
- + getSession(sessionHandle).getClusterUser());
+ + getSession(sessionHandle).getClusterUser());
return Response.seeOther(resultReadPath)
- .header("content-disposition", "attachment; filename = " + resultPath.getName())
- .type(MediaType.APPLICATION_OCTET_STREAM).build();
+ .header("content-disposition", "attachment; filename = " + resultPath.getName())
+ .type(MediaType.APPLICATION_OCTET_STREAM).build();
} catch (URISyntaxException e) {
throw new LensException(e);
}
@@ -2027,7 +2039,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
}
};
return Response.ok(stream).header("content-disposition", "attachment; filename = " + resultPath.getName())
- .type(MediaType.APPLICATION_OCTET_STREAM).build();
+ .type(MediaType.APPLICATION_OCTET_STREAM).build();
}
} else {
throw new NotFoundException("Http result not available for query:" + queryHandle.toString());
@@ -2037,8 +2049,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Allow drivers to release resources acquired for a session if any.
*
- * @param sessionHandle
- * the session handle
+ * @param sessionHandle the session handle
*/
public void closeDriverSessions(LensSessionHandle sessionHandle) {
for (LensDriver driver : drivers.values()) {
@@ -2082,8 +2093,7 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
/**
* Handle driver session start.
*
- * @param event
- * the event
+ * @param event the event
*/
protected void handleDriverSessionStart(DriverEvent event) {
DriverSessionStarted sessionStarted = (DriverSessionStarted) event;
@@ -2121,8 +2131,8 @@ public class QueryExecutionServiceImpl extends LensService implements QueryExecu
}
} catch (Exception e) {
LOG.warn(
- "Lens session went away! " + lensSession + " driver session: "
- + ((DriverSessionStarted) event).getDriverSessionID(), e);
+ "Lens session went away! " + lensSession + " driver session: "
+ + ((DriverSessionStarted) event).getDriverSessionID(), e);
} finally {
release(sessionHandle);
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/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 bb22df0..2ec2ba7 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
@@ -87,7 +87,7 @@ public class QueryExecutionStatisticsGenerator extends AsyncEventListener<QueryE
event.setClusterUser(ctx.getClusterUser());
event.setSubmissionTime(ctx.getSubmissionTime());
QueryDriverStatistics driverStats = new QueryDriverStatistics();
- driverStats.setDriverQuery(ctx.getDriverQuery());
+ driverStats.setDriverQuery(ctx.getSelectedDriverQuery());
driverStats.setStartTime(ctx.getDriverStatus().getDriverStartTime());
driverStats.setEndTime(ctx.getDriverStatus().getDriverStartTime());
event.setDriverStats(driverStats);
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/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 da1c6b7..b0ff421 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
@@ -44,6 +44,7 @@ import org.apache.hadoop.hive.ql.HiveDriverRunHook;
import org.apache.hadoop.hive.ql.HiveDriverRunHookContext;
import org.apache.hadoop.io.IOUtils;
import org.apache.lens.driver.hive.TestHiveDriver;
+import org.apache.lens.server.api.query.QueryContext;
import org.glassfish.jersey.client.ClientConfig;
import org.glassfish.jersey.media.multipart.FormDataBodyPart;
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
@@ -1317,6 +1318,18 @@ public class TestQueryService extends LensJerseyTest {
// Test server config. Hive configs overriden should be set
Assert.assertFalse(Boolean.parseBoolean(queryService.getHiveConf().get("hive.server2.log.redirection.enabled")));
Assert.assertEquals(queryService.getHiveConf().get("hive.server2.query.log.dir"), "target/query-logs");
+
+ final String query = "test query";
+ QueryContext ctx = new QueryContext(query, null, queryConf, conf, queryService.getDrivers());
+ Map<LensDriver, String> driverQueries = new HashMap<LensDriver, String>() {{ put(queryService.getDrivers
+ ().iterator().next(), query); }};
+ ctx.getDriverContext().setDriverQueriesAndPlans(driverQueries);
+
+ Assert.assertEquals(queryService.getSession(lensSessionId).getHiveConf().getClassLoader() , ctx.getConf()
+ .getClassLoader());
+ Assert.assertEquals(queryService.getSession(lensSessionId).getHiveConf().getClassLoader(), ctx.getDriverContext().getDriverConf(queryService.getDrivers
+ ().iterator().next()).getClassLoader());
+
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/tools/conf/server/hivedriver-site.xml
----------------------------------------------------------------------
diff --git a/tools/conf/server/hivedriver-site.xml b/tools/conf/server/hivedriver-site.xml
index f179ecd..2e8e7fa 100644
--- a/tools/conf/server/hivedriver-site.xml
+++ b/tools/conf/server/hivedriver-site.xml
@@ -36,5 +36,6 @@
<property>
<name>lens.cube.query.driver.supported.storages</name>
<value>local,cluster</value>
+ <final>true</final>
</property>
</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/tools/conf/server/jdbcdriver-site.xml
----------------------------------------------------------------------
diff --git a/tools/conf/server/jdbcdriver-site.xml b/tools/conf/server/jdbcdriver-site.xml
index 235187a..116fddf 100644
--- a/tools/conf/server/jdbcdriver-site.xml
+++ b/tools/conf/server/jdbcdriver-site.xml
@@ -37,6 +37,7 @@
<property>
<name>lens.cube.query.driver.supported.storages</name>
<value>mydb</value>
+ <final>true</final>
</property>
<property>
<name>lens.driver.jdbc.query.rewriter</name>
[2/2] incubator-lens git commit: LENS-20 : Sets Driver Specific
Configuration should be set in Query Context (Suma Shivaprasad via
amareshwari)
Posted by am...@apache.org.
LENS-20 : Sets Driver Specific Configuration should be set in Query Context (Suma Shivaprasad via amareshwari)
Project: http://git-wip-us.apache.org/repos/asf/incubator-lens/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-lens/commit/fcbb44a1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-lens/tree/fcbb44a1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-lens/diff/fcbb44a1
Branch: refs/heads/master
Commit: fcbb44a141d4a2b316bb0b6fc4f0716fc6a715be
Parents: f5e5ecc
Author: Amareshwari Sriramdasu <am...@inmobi.com>
Authored: Mon Dec 1 14:44:33 2014 +0530
Committer: Amareshwari Sriramdasu <am...@inmobi.com>
Committed: Mon Dec 1 14:44:33 2014 +0530
----------------------------------------------------------------------
.../apache/lens/driver/cube/RewriteUtil.java | 51 +-
.../lens/driver/cube/TestMinCostSelector.java | 42 +-
.../apache/lens/driver/cube/TestRewriting.java | 104 ++--
.../org/apache/lens/driver/hive/HiveDriver.java | 24 +-
.../DurationBasedQueryPriorityDecider.java | 8 +-
.../apache/lens/driver/hive/TestHiveDriver.java | 43 +-
.../lens/driver/hive/TestRemoteHiveDriver.java | 2 +
.../org/apache/lens/driver/jdbc/JDBCDriver.java | 19 +-
.../apache/lens/driver/jdbc/TestJDBCFinal.java | 24 +-
.../apache/lens/driver/jdbc/TestJdbcDriver.java | 52 +-
.../lib/query/TestAbstractFileFormatter.java | 23 +-
.../server/api/driver/MinQueryCostSelector.java | 5 +-
.../lens/server/api/driver/MockDriver.java | 6 +-
.../server/api/driver/MockQueryContext.java | 8 +-
.../server/api/query/AbstractQueryContext.java | 102 +++-
.../api/query/DriverSelectorQueryContext.java | 158 +++++-
.../server/api/query/ExplainQueryContext.java | 9 +-
.../server/api/query/PreparedQueryContext.java | 20 +-
.../lens/server/api/query/QueryContext.java | 93 ++--
.../server/query/QueryExecutionServiceImpl.java | 542 ++++++++++---------
.../QueryExecutionStatisticsGenerator.java | 2 +-
.../lens/server/query/TestQueryService.java | 13 +
tools/conf/server/hivedriver-site.xml | 1 +
tools/conf/server/jdbcdriver-site.xml | 1 +
24 files changed, 865 insertions(+), 487 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-cube/src/main/java/org/apache/lens/driver/cube/RewriteUtil.java
----------------------------------------------------------------------
diff --git a/lens-cube/src/main/java/org/apache/lens/driver/cube/RewriteUtil.java b/lens-cube/src/main/java/org/apache/lens/driver/cube/RewriteUtil.java
index 9c76271..e8cad1d 100644
--- a/lens-cube/src/main/java/org/apache/lens/driver/cube/RewriteUtil.java
+++ b/lens-cube/src/main/java/org/apache/lens/driver/cube/RewriteUtil.java
@@ -32,10 +32,11 @@ import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.ParseException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.lens.api.LensException;
-import org.apache.lens.cube.parse.CubeQueryConfUtil;
import org.apache.lens.cube.parse.CubeQueryRewriter;
import org.apache.lens.cube.parse.HQLParser;
import org.apache.lens.server.api.driver.LensDriver;
+
+import org.apache.lens.server.api.query.AbstractQueryContext;
import org.apache.log4j.Logger;
/**
@@ -187,41 +188,16 @@ public class RewriteUtil {
}
/**
- * Gets the final query conf.
- *
- * @param driver
- * the driver
- * @param queryConf
- * the query conf
- * @return the final query conf
- */
- public static Configuration getFinalQueryConf(LensDriver driver, Configuration queryConf) {
- Configuration conf = new Configuration(driver.getConf());
- for (Map.Entry<String, String> entry : queryConf) {
- if (entry.getKey().equals(CubeQueryConfUtil.DRIVER_SUPPORTED_STORAGES)) {
- LOG.warn(CubeQueryConfUtil.DRIVER_SUPPORTED_STORAGES + " value : " + entry.getValue()
- + " from query conf ignored/");
- continue;
- }
- conf.set(entry.getKey(), entry.getValue());
- }
- conf.setClassLoader(queryConf.getClassLoader());
- return conf;
- }
-
- /**
* Gets the rewriter.
*
- * @param driver
- * the driver
* @param queryConf
* the query conf
* @return the rewriter
* @throws SemanticException
* the semantic exception
*/
- static CubeQueryRewriter getRewriter(LensDriver driver, Configuration queryConf) throws SemanticException {
- return new CubeQueryRewriter(getFinalQueryConf(driver, queryConf));
+ static CubeQueryRewriter getCubeRewriter(Configuration queryConf) throws SemanticException {
+ return new CubeQueryRewriter(queryConf);
}
/**
@@ -240,33 +216,28 @@ public class RewriteUtil {
/**
* Rewrite query.
*
- * @param query
- * the query
- * @param drivers
- * the drivers
- * @param queryconf
- * the queryconf
+ * @param ctx
+ * the query context
* @return the map
* @throws LensException
* the lens exception
*/
- public static Map<LensDriver, String> rewriteQuery(final String query, Collection<LensDriver> drivers,
- Configuration queryconf) throws LensException {
+ public static Map<LensDriver, String> rewriteQuery(AbstractQueryContext ctx) throws LensException {
try {
- String replacedQuery = getReplacedQuery(query);
+ String replacedQuery = getReplacedQuery(ctx.getUserQuery());
String lowerCaseQuery = replacedQuery.toLowerCase();
Map<LensDriver, String> driverQueries = new HashMap<LensDriver, String>();
StringBuilder rewriteFailure = new StringBuilder();
String failureCause = null;
boolean useBuilder = false;
if (lowerCaseQuery.startsWith("add") || lowerCaseQuery.startsWith("set")) {
- for (LensDriver driver : drivers) {
+ for (LensDriver driver : ctx.getDriverContext().getDrivers()) {
driverQueries.put(driver, replacedQuery);
}
} else {
List<RewriteUtil.CubeQueryInfo> cubeQueries = findCubePositions(replacedQuery);
- for (LensDriver driver : drivers) {
- CubeQueryRewriter rewriter = getRewriter(driver, queryconf);
+ for (LensDriver driver : ctx.getDriverContext().getDrivers()) {
+ CubeQueryRewriter rewriter = getCubeRewriter(ctx.getDriverContext().getDriverConf(driver));
StringBuilder builder = new StringBuilder();
int start = 0;
try {
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-cube/src/test/java/org/apache/lens/driver/cube/TestMinCostSelector.java
----------------------------------------------------------------------
diff --git a/lens-cube/src/test/java/org/apache/lens/driver/cube/TestMinCostSelector.java b/lens-cube/src/test/java/org/apache/lens/driver/cube/TestMinCostSelector.java
index a3df316..1981757 100644
--- a/lens-cube/src/test/java/org/apache/lens/driver/cube/TestMinCostSelector.java
+++ b/lens-cube/src/test/java/org/apache/lens/driver/cube/TestMinCostSelector.java
@@ -19,11 +19,13 @@
package org.apache.lens.driver.cube;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensException;
import org.apache.lens.server.api.driver.*;
import org.testng.Assert;
@@ -33,36 +35,60 @@ import org.testng.annotations.Test;
* The Class TestMinCostSelector.
*/
public class TestMinCostSelector {
+
+ private MockQueryContext createMockContext(String query, Configuration conf, LensConf lensConf,
+ Map<LensDriver, String> driverQueries) throws LensException {
+ MockQueryContext ctx = new MockQueryContext(query, lensConf, conf, driverQueries.keySet());
+ ctx.getDriverContext().setDriverQueriesAndPlans(driverQueries);
+ return ctx;
+ }
+
@Test
public void testMinCostSelector() throws LensException {
MinQueryCostSelector selector = new MinQueryCostSelector();
List<LensDriver> drivers = new ArrayList<LensDriver>();
Map<LensDriver, String> driverQueries = new HashMap<LensDriver, String>();
Configuration conf = new Configuration();
+ LensConf qconf = new LensConf();
MockDriver d1 = new MockDriver();
+ d1.configure(conf);
MockDriver d2 = new MockDriver();
+ d2.configure(conf);
MockFailDriver fd1 = new MockFailDriver();
+ fd1.configure(conf);
MockFailDriver fd2 = new MockFailDriver();
+ fd2.configure(conf);
drivers.add(d1);
drivers.add(d2);
- driverQueries.put(d1, "test query");
- LensDriver selected = selector.select(new MockQueryContext(driverQueries), conf);
+ String query = "test query";
+ driverQueries.put(d1, query);
+
+ MockQueryContext ctx = createMockContext(query, conf, qconf, driverQueries);
+ LensDriver selected = selector.select(ctx, conf);
+
Assert.assertEquals(d1, selected);
- driverQueries.put(d2, "test query");
+ driverQueries.put(d2, query);
driverQueries.remove(d1);
- selected = selector.select(new MockQueryContext(driverQueries), conf);
+ ctx = createMockContext(query, conf, qconf, driverQueries);
+
+ selected = selector.select(ctx, conf);
Assert.assertEquals(d2, selected);
drivers.add(fd1);
- driverQueries.put(fd1, "test query");
- selected = selector.select(new MockQueryContext(driverQueries), conf);
+ driverQueries.put(fd1, query);
+
+ ctx = createMockContext(query, conf, qconf, driverQueries);
+ selected = selector.select(ctx, conf);
Assert.assertEquals(d2, selected);
+
drivers.add(fd2);
- driverQueries.put(fd2, "test query");
- selected = selector.select(new MockQueryContext(driverQueries), conf);
+ driverQueries.put(fd2, query);
+ ctx = createMockContext(query, conf, qconf, driverQueries);
+
+ selected = selector.select(ctx, conf);
Assert.assertEquals(d2, selected);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-cube/src/test/java/org/apache/lens/driver/cube/TestRewriting.java
----------------------------------------------------------------------
diff --git a/lens-cube/src/test/java/org/apache/lens/driver/cube/TestRewriting.java b/lens-cube/src/test/java/org/apache/lens/driver/cube/TestRewriting.java
index 538d27d..796a582 100644
--- a/lens-cube/src/test/java/org/apache/lens/driver/cube/TestRewriting.java
+++ b/lens-cube/src/test/java/org/apache/lens/driver/cube/TestRewriting.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.ParseException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensException;
import org.apache.lens.cube.parse.CubeQueryContext;
import org.apache.lens.cube.parse.CubeQueryRewriter;
@@ -36,6 +37,7 @@ import org.apache.lens.server.api.driver.LensDriver;
import static org.mockito.Matchers.any;
import org.apache.lens.server.api.driver.MockDriver;
+import org.apache.lens.server.api.query.QueryContext;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest;
@@ -146,80 +148,91 @@ public class TestRewriting {
public void testCubeQuery() throws ParseException, SemanticException, LensException {
List<LensDriver> drivers = new ArrayList<LensDriver>();
MockDriver driver = new MockDriver();
+ LensConf lensConf = new LensConf();
Configuration conf = new Configuration();
driver.configure(conf);
drivers.add(driver);
CubeQueryRewriter mockWriter = getMockedRewriter();
- PowerMockito.stub(PowerMockito.method(RewriteUtil.class, "getRewriter")).toReturn(mockWriter);
+ PowerMockito.stub(PowerMockito.method(RewriteUtil.class, "getCubeRewriter")).toReturn(mockWriter);
String q1 = "select name from table";
Assert.assertFalse(RewriteUtil.isCubeQuery(q1));
List<RewriteUtil.CubeQueryInfo> cubeQueries = RewriteUtil.findCubePositions(q1);
Assert.assertEquals(cubeQueries.size(), 0);
- RewriteUtil.rewriteQuery(q1, drivers, conf);
+ QueryContext ctx = new QueryContext(q1, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
String q2 = "cube select name from table";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "insert overwrite directory '/tmp/rewrite' cube select name from table";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "insert overwrite local directory '/tmp/rewrite' cube select name from table";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "insert overwrite local directory '/tmp/example-output' cube select id,name from dim_table";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select id,name from dim_table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "explain cube select name from table";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table) a";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "insert overwrite directory '/tmp/rewrite' select * from (cube select name from table) a";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table)a";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from ( cube select name from table ) a";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from ( cube select name from table where"
+ " (name = 'ABC'||name = 'XYZ')&&(key=100) ) a";
@@ -228,7 +241,8 @@ public class TestRewriting {
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from"
+ " table where (name = 'ABC' OR name = 'XYZ') AND (key=100)");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table) a join (cube select" + " name2 from table2) b";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
@@ -236,7 +250,8 @@ public class TestRewriting {
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table) a full outer join"
+ " (cube select name2 from table2) b on a.name=b.name2";
@@ -245,68 +260,80 @@ public class TestRewriting {
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table) a join (select name2 from table2) b";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table union all cube select name2 from table2) u";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "insert overwrite directory '/tmp/rewrite' select * from (cube select name from table union all cube select name2 from table2) u";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select u.* from (select name from table union all cube select name2 from table2) u";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select u.* from (select name from table union all cube select name2 from table2)u";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table union all cube select name2"
+ " from table2 union all cube select name3 from table3) u";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 3);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
Assert.assertEquals(cubeQueries.get(2).query, "cube select name3 from table3");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from ( cube select name from table union all cube"
+ " select name2 from table2 union all cube select name3 from table3 ) u";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 3);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
Assert.assertEquals(cubeQueries.get(2).query, "cube select name3 from table3");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table union all cube select" + " name2 from table2) u group by u.name";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
@@ -314,43 +341,52 @@ public class TestRewriting {
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "select * from (cube select name from table union all cube select" + " name2 from table2) u group by u.name";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "create table temp1 as cube select name from table";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "create table temp1 as select * from (cube select name from table union all cube select"
+ " name2 from table2) u group by u.name";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 2);
Assert.assertEquals(cubeQueries.get(0).query, "cube select name from table");
Assert.assertEquals(cubeQueries.get(1).query, "cube select name2 from table2");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
q2 = "create table temp1 as cube select name from table where time_range_in('dt', '2014-06-24-23', '2014-06-25-00')";
Assert.assertTrue(RewriteUtil.isCubeQuery(q2));
cubeQueries = RewriteUtil.findCubePositions(q2);
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
Assert.assertEquals(cubeQueries.size(), 1);
Assert.assertEquals(cubeQueries.get(0).query,
"cube select name from table where time_range_in('dt', '2014-06-24-23', '2014-06-25-00')");
- RewriteUtil.rewriteQuery(q2, drivers, conf);
+ ctx = new QueryContext(q2, null, lensConf, conf, drivers);
+ RewriteUtil.rewriteQuery(ctx);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/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 65d6988..cdfa828 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
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.ArrayList;
+import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -33,6 +34,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
@@ -49,6 +51,7 @@ import org.apache.hive.service.cli.SessionHandle;
import org.apache.hive.service.cli.thrift.TProtocolVersion;
import org.apache.hive.service.cli.thrift.TSessionHandle;
import org.apache.hive.service.cli.thrift.TOperationHandle;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensException;
import org.apache.lens.api.LensSessionHandle;
import org.apache.lens.api.query.QueryHandle;
@@ -313,9 +316,11 @@ public class HiveDriver implements LensDriver {
LOG.info("Explain: " + query);
Configuration explainConf = new Configuration(conf);
explainConf.setBoolean(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, false);
- String explainQuery = "EXPLAIN EXTENDED " + query;
- QueryContext explainQueryCtx = new QueryContext(explainQuery, SessionState.get().getUserName(), explainConf);
- // Get result set of explain
+ final String explainQuery = "EXPLAIN EXTENDED " + query;
+ QueryContext explainQueryCtx = new QueryContext(explainQuery, SessionState.get().getUserName(), new LensConf(),
+ explainConf,
+ Lists.newArrayList((LensDriver) this), (LensDriver) this, new Date().getTime());
+ // Get result set of explain
HiveInMemoryResultSet inMemoryResultSet = (HiveInMemoryResultSet) execute(explainQueryCtx);
List<String> explainOutput = new ArrayList<String>();
while (inMemoryResultSet.hasNext()) {
@@ -343,7 +348,7 @@ public class HiveDriver implements LensDriver {
*/
@Override
public DriverQueryPlan explainAndPrepare(PreparedQueryContext pContext) throws LensException {
- DriverQueryPlan plan = explain(pContext.getDriverQuery(), pContext.getConf());
+ DriverQueryPlan plan = explain(pContext.getSelectedDriverQuery(), pContext.getSelectedDriverConf());
plan.setPrepareHandle(pContext.getPrepareHandle());
return plan;
}
@@ -379,7 +384,7 @@ public class HiveDriver implements LensDriver {
try {
addPersistentPath(ctx);
ctx.getConf().set("mapred.job.name", ctx.getQueryHandle().toString());
- OperationHandle op = getClient().executeStatement(getSession(ctx), ctx.getDriverQuery(),
+ OperationHandle op = getClient().executeStatement(getSession(ctx), ctx.getSelectedDriverQuery(),
ctx.getConf().getValByRegex(".*"));
LOG.info("The hive operation handle: " + op);
ctx.setDriverOpHandle(op.toString());
@@ -425,7 +430,8 @@ public class HiveDriver implements LensDriver {
LOG.error("could not set priority for lens session id:" + ctx.getLensSessionIdentifier(), e);
}
}
- OperationHandle op = getClient().executeStatementAsync(getSession(ctx), ctx.getDriverQuery(),
+ OperationHandle op = getClient().executeStatementAsync(getSession(ctx), ctx.
+ getSelectedDriverQuery(),
ctx.getConf().getValByRegex(".*"));
ctx.setDriverOpHandle(op.toString());
LOG.info("QueryHandle: " + ctx.getQueryHandle() + " HiveHandle:" + op);
@@ -767,13 +773,13 @@ public class HiveDriver implements LensDriver {
if (outputDirFormat != null) {
builder.append(outputDirFormat);
}
- builder.append(' ').append(context.getDriverQuery()).append(' ');
+ builder.append(' ').append(context.getSelectedDriverQuery()).append(' ');
hiveQuery = builder.toString();
} else {
- hiveQuery = context.getDriverQuery();
+ hiveQuery = context.getSelectedDriverQuery();
}
LOG.info("Hive driver query:" + hiveQuery);
- context.setDriverQuery(hiveQuery);
+ context.setSelectedDriverQuery(hiveQuery);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/priority/DurationBasedQueryPriorityDecider.java
----------------------------------------------------------------------
diff --git a/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/priority/DurationBasedQueryPriorityDecider.java b/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/priority/DurationBasedQueryPriorityDecider.java
index c9ab9bd..b87ee11 100644
--- a/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/priority/DurationBasedQueryPriorityDecider.java
+++ b/lens-driver-hive/src/main/java/org/apache/lens/driver/hive/priority/DurationBasedQueryPriorityDecider.java
@@ -85,7 +85,9 @@ public class DurationBasedQueryPriorityDecider implements QueryPriorityDecider {
*/
protected Map<String,List<String>> extractPartitions(AbstractQueryContext queryContext) throws LensException{
Map<String, List<String>> partitions = new HashMap<String, List<String>>();
- for(Map.Entry<String, List<String>> entry: queryContext.getSelectedDriverQueryPlan().getPartitions().entrySet()) {
+ for(Map.Entry<String, List<String>> entry: queryContext.getDriverContext().getSelectedDriverQueryPlan()
+ .getPartitions().entrySet
+ ()) {
partitions.put(entry.getKey(), new ArrayList<String>());
for(String s: entry.getValue()) {
String[] splits = s.split("\\s+");
@@ -109,7 +111,9 @@ public class DurationBasedQueryPriorityDecider implements QueryPriorityDecider {
for(String partition: partitions.get(table)) {
if(!partition.equals("latest")) {
cost +=
- queryContext.getSelectedDriverQueryPlan().getTableWeight(table) * getNormalizedPartitionCost(partition);
+ queryContext.getDriverContext().getSelectedDriverQueryPlan().getTableWeight(table) *
+ getNormalizedPartitionCost
+ (partition);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/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 dfbdb79..94953f7 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
@@ -37,6 +37,7 @@ import org.apache.hadoop.hive.ql.HiveDriverRunHookContext;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hive.service.cli.ColumnDescriptor;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensException;
import org.apache.lens.api.Priority;
import org.apache.lens.api.query.QueryHandle;
@@ -68,6 +69,9 @@ public class TestHiveDriver {
/** The driver. */
protected HiveDriver driver;
+ /** Driver list **/
+ protected Collection<LensDriver> drivers;
+
/** The data base. */
public String DATA_BASE = this.getClass().getSimpleName().toLowerCase();
@@ -112,6 +116,8 @@ public class TestHiveDriver {
conf.setBoolean(HiveDriver.HS2_CALCULATE_PRIORITY, false);
driver = new HiveDriver();
driver.configure(conf);
+ drivers = new ArrayList<LensDriver>() {{ add
+ (driver);}};
System.out.println("TestHiveDriver created");
}
@@ -120,8 +126,10 @@ public class TestHiveDriver {
SessionState.get().setCurrentDatabase(DATA_BASE);
}
- protected QueryContext createContext(String query, Configuration conf) {
- QueryContext context = new QueryContext(query, "testuser", conf);
+ protected QueryContext createContext(final String query, Configuration conf) throws LensException {
+ QueryContext context = new QueryContext(query, "testuser", conf, drivers);
+ context.getDriverContext().setDriverQueriesAndPlans(new HashMap<LensDriver, String>() {{ put(driver, query); }} );
+ context.setSelectedDriver(driver);
context.setLensSessionIdentifier(sessionid);
return context;
}
@@ -212,8 +220,8 @@ public class TestHiveDriver {
QueryContext context = createContext(query, conf);
driver.addPersistentPath(context);
assertEquals(context.getUserQuery(), query);
- assertNotNull(context.getDriverQuery());
- assertEquals(context.getDriverQuery(), context.getUserQuery());
+ assertNotNull(context.getDriverContext().getDriverQuery(driver));
+ assertEquals(context.getDriverContext().getDriverQuery(driver), context.getUserQuery());
}
/**
@@ -651,7 +659,9 @@ public class TestHiveDriver {
Assert.assertEquals(0, driver.getHiveHandleSize());
// test execute prepare
- PreparedQueryContext pctx = new PreparedQueryContext("SELECT ID FROM test_explain", null, conf);
+ PreparedQueryContext pctx = new PreparedQueryContext("SELECT ID FROM test_explain", null, conf, drivers);
+ pctx.setSelectedDriver(driver);
+
SessionState.setCurrentSessionState(ss);
plan = driver.explainAndPrepare(pctx);
QueryContext qctx = createContext(pctx, conf);
@@ -752,7 +762,8 @@ public class TestHiveDriver {
conf.setBoolean(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, true);
SessionState.setCurrentSessionState(ss);
String query2 = "SELECT DISTINCT ID FROM explain_test_1";
- PreparedQueryContext pctx = new PreparedQueryContext(query2, null, conf);
+ PreparedQueryContext pctx = new PreparedQueryContext(query2, null, conf, drivers);
+ pctx.setSelectedDriver(driver);
DriverQueryPlan plan2 = driver.explainAndPrepare(pctx);
// assertNotNull(plan2.getResultDestination());
Assert.assertEquals(0, driver.getHiveHandleSize());
@@ -776,7 +787,10 @@ public class TestHiveDriver {
*/
@Test
public void testPriority() throws IOException, LensException {
+ Configuration conf = new Configuration();
final MockDriver mockDriver = new MockDriver();
+ mockDriver.configure(conf);
+
BufferedReader br = new BufferedReader(new InputStreamReader(TestHiveDriver.class.getResourceAsStream("/priority_tests.txt")));
String line;
while((line = br.readLine()) != null) {
@@ -784,15 +798,22 @@ public class TestHiveDriver {
final List<String> partitions = Arrays.asList(kv[0].trim().split("\\s*,\\s*"));
final Priority expected = Priority.valueOf(kv[1]);
- AbstractQueryContext ctx = new MockQueryContext(new HashMap<LensDriver, String>(){
+ final HashMap<LensDriver, String> driverQuery1 = new HashMap<LensDriver, String>() {
{
put(mockDriver, "driverQuery1");
}
- });
+ };
+ AbstractQueryContext ctx = new MockQueryContext("driverQuery1", new LensConf(), conf,
+ driverQuery1.keySet());
+ ctx.getDriverContext().setDriverQueriesAndPlans(driverQuery1);
ctx.setSelectedDriver(mockDriver);
- ((MockDriver.MockQueryPlan)ctx.getDriverQueryPlans().get(mockDriver)).setPartitions(new HashMap<String, List<String>>() {
- {
- put("table1", partitions);
+
+ ((MockDriver.MockQueryPlan)ctx.getDriverContext().getDriverQueryPlan(mockDriver)).setPartitions
+ (new HashMap<String,
+ List<String>>
+ () {
+ {
+ put("table1", partitions);
}});
Assert.assertEquals(expected, driver.queryPriorityDecider.decidePriority(ctx));
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestRemoteHiveDriver.java
----------------------------------------------------------------------
diff --git a/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestRemoteHiveDriver.java b/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestRemoteHiveDriver.java
index c213709..b2f980e 100644
--- a/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestRemoteHiveDriver.java
+++ b/lens-driver-hive/src/test/java/org/apache/lens/driver/hive/TestRemoteHiveDriver.java
@@ -131,6 +131,8 @@ public class TestRemoteHiveDriver extends TestHiveDriver {
driver = new HiveDriver();
conf.setBoolean(HiveDriver.HS2_CALCULATE_PRIORITY, false);
driver.configure(conf);
+ drivers = new ArrayList<LensDriver>() {{ add
+ (driver);}};
System.out.println("TestRemoteHiveDriver created");
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-driver-jdbc/src/main/java/org/apache/lens/driver/jdbc/JDBCDriver.java
----------------------------------------------------------------------
diff --git a/lens-driver-jdbc/src/main/java/org/apache/lens/driver/jdbc/JDBCDriver.java b/lens-driver-jdbc/src/main/java/org/apache/lens/driver/jdbc/JDBCDriver.java
index f073203..09f88d8 100644
--- a/lens-driver-jdbc/src/main/java/org/apache/lens/driver/jdbc/JDBCDriver.java
+++ b/lens-driver-jdbc/src/main/java/org/apache/lens/driver/jdbc/JDBCDriver.java
@@ -27,11 +27,11 @@ import org.apache.lens.api.query.QueryCost;
import org.apache.lens.api.query.QueryHandle;
import org.apache.lens.api.query.QueryPrepareHandle;
import org.apache.lens.cube.parse.HQLParser;
-import org.apache.lens.driver.cube.RewriteUtil;
import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.driver.*;
import org.apache.lens.server.api.driver.DriverQueryStatus.DriverQueryState;
import org.apache.lens.server.api.events.LensEventListener;
+import static org.apache.lens.server.api.query.DriverSelectorQueryContext.DriverQueryContext;
import org.apache.lens.server.api.query.PreparedQueryContext;
import org.apache.lens.server.api.query.QueryContext;
import org.apache.lens.server.api.query.QueryRewriter;
@@ -549,7 +549,6 @@ public class JDBCDriver implements LensDriver {
throws LensException {
checkConfigured();
String explainQuery;
- conf = RewriteUtil.getFinalQueryConf(this, conf);
String rewrittenQuery = rewriteQuery(query, conf);
Configuration explainConf = new Configuration(conf);
explainConf.setBoolean(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER,
@@ -565,8 +564,9 @@ public class JDBCDriver implements LensDriver {
explainQuery = rewrittenQuery.replaceAll("select ", "select "
+ explainKeyword + " ");
LOG.info("Explain Query : " + explainQuery);
+ final LensDriver driver = (LensDriver) this;
QueryContext explainQueryCtx = new QueryContext(explainQuery, null,
- explainConf);
+ explainConf, new ArrayList<LensDriver>() {{ add(driver); }});
QueryResult result = null;
try {
@@ -595,7 +595,7 @@ public class JDBCDriver implements LensDriver {
public void prepare(PreparedQueryContext pContext) throws LensException {
checkConfigured();
// Only create a prepared statement and then close it
- String rewrittenQuery = rewriteQuery(pContext.getDriverQuery(), pContext.getConf());
+ String rewrittenQuery = rewriteQuery(pContext.getSelectedDriverQuery(), pContext.getSelectedDriverConf());
Connection conn = null;
PreparedStatement stmt = null;
try {
@@ -635,7 +635,7 @@ public class JDBCDriver implements LensDriver {
@Override
public DriverQueryPlan explainAndPrepare(PreparedQueryContext pContext) throws LensException {
checkConfigured();
- String rewritten = rewriteQuery(pContext.getDriverQuery(), conf);
+ String rewritten = rewriteQuery(pContext.getSelectedDriverQuery(), conf);
prepare(pContext);
return new JDBCQueryPlan();
}
@@ -666,9 +666,9 @@ public class JDBCDriver implements LensDriver {
@Override
public LensResultSet execute(QueryContext context) throws LensException {
checkConfigured();
- // Always use the driver rewritten query not user query. Since the
- // conf we are passing here is query context conf, we need to add jdbc xml in resource path
- String rewrittenQuery = rewriteQuery(context.getDriverQuery(), RewriteUtil.getFinalQueryConf(this, conf));
+
+ String rewrittenQuery = rewriteQuery(context.getSelectedDriverQuery(), context
+ .getSelectedDriverConf());
LOG.info("Execute " + context.getQueryHandle());
QueryResult result = executeInternal(context, rewrittenQuery);
return result.getLensResultSet(true);
@@ -709,7 +709,8 @@ public class JDBCDriver implements LensDriver {
checkConfigured();
// Always use the driver rewritten query not user query. Since the
// conf we are passing here is query context conf, we need to add jdbc xml in resource path
- String rewrittenQuery = rewriteQuery(context.getDriverQuery(), RewriteUtil.getFinalQueryConf(this, conf));
+ String rewrittenQuery = rewriteQuery(context.getSelectedDriverQuery(), context.getDriverContext()
+ .getSelectedDriverConf());
JdbcQueryContext jdbcCtx = new JdbcQueryContext(context);
jdbcCtx.setRewrittenQuery(rewrittenQuery);
try {
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJDBCFinal.java
----------------------------------------------------------------------
diff --git a/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJDBCFinal.java b/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJDBCFinal.java
index fafa1a5..115171d 100644
--- a/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJDBCFinal.java
+++ b/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJDBCFinal.java
@@ -24,6 +24,9 @@ import static org.testng.Assert.assertTrue;
import java.sql.Connection;
import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
@@ -35,6 +38,7 @@ import org.apache.lens.driver.jdbc.ColumnarSQLRewriter;
import org.apache.lens.driver.jdbc.JDBCDriver;
import org.apache.lens.driver.jdbc.JDBCDriverConfConstants;
import org.apache.lens.driver.jdbc.JDBCResultSet;
+import org.apache.lens.server.api.driver.LensDriver;
import org.apache.lens.server.api.driver.LensResultSet;
import org.apache.lens.server.api.driver.LensResultSetMetadata;
import org.apache.lens.server.api.driver.InMemoryResultSet;
@@ -55,6 +59,11 @@ public class TestJDBCFinal {
JDBCDriver driver;
/**
+ * Collection of drivers
+ */
+ Collection<LensDriver> drivers;
+
+ /**
* Test create jdbc driver.
*
* @throws Exception
@@ -76,6 +85,8 @@ public class TestJDBCFinal {
System.out.println("Driver configured!");
SessionState.start(new HiveConf(ColumnarSQLRewriter.class));
+ drivers = new ArrayList<LensDriver>() {{
+ add(driver); }};
}
/**
@@ -170,14 +181,17 @@ public class TestJDBCFinal {
@Test
public void testExecute1() throws Exception {
testCreateJdbcDriver();
- String query =
+ final String query =
"select fact.time_key,time_dim.day_of_week,time_dim.day," + "sum(fact.dollars_sold) dollars_sold "
+ "from sales_fact fact " + "inner join time_dim time_dim on fact.time_key = time_dim.time_key "
+ "where time_dim.day between '1900-01-01' and '1900-01-03' "
+ "group by fact.time_key,time_dim.day_of_week,time_dim.day " + "order by dollars_sold desc";
- QueryContext context = new QueryContext(query, "SA", baseConf);
+ QueryContext context = new QueryContext(query, "SA", baseConf, drivers);
+ context.getDriverContext().setDriverQueriesAndPlans(new HashMap<LensDriver, String>() {{ put(driver, query); }} );
+ context.setSelectedDriver(driver);
+
LensResultSet resultSet = driver.execute(context);
assertNotNull(resultSet);
@@ -223,7 +237,7 @@ public class TestJDBCFinal {
@Test
public void testExecute2() throws Exception {
testCreateJdbcDriver();
- String query =
+ final String query =
"select fact.time_key,time_dim.day_of_week,time_dim.day, " + "sum(fact.dollars_sold) dollars_sold "
+ "from sales_fact fact " + "inner join time_dim time_dim on fact.time_key = time_dim.time_key "
@@ -233,7 +247,9 @@ public class TestJDBCFinal {
+ "where time_dim.day between '1900-01-01' and '1900-01-04' " + "and location_dim.location_name = 'loc2' "
+ "group by fact.time_key,time_dim.day_of_week,time_dim.day " + "order by dollars_sold desc ";
- QueryContext context = new QueryContext(query, "SA", baseConf);
+ QueryContext context = new QueryContext(query, "SA", baseConf, drivers);
+ context.getDriverContext().setDriverQueriesAndPlans(new HashMap<LensDriver, String>() {{ put(driver, query); }} );
+ context.setSelectedDriver(driver);
LensResultSet resultSet = driver.execute(context);
assertNotNull(resultSet);
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJdbcDriver.java
----------------------------------------------------------------------
diff --git a/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJdbcDriver.java b/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJdbcDriver.java
index c2d38fc..30b7935 100644
--- a/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJdbcDriver.java
+++ b/lens-driver-jdbc/src/test/java/org/apache/lens/driver/jdbc/TestJdbcDriver.java
@@ -21,7 +21,11 @@ package org.apache.lens.driver.jdbc;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -33,6 +37,7 @@ import org.apache.lens.api.query.ResultRow;
import org.apache.lens.driver.jdbc.JDBCDriver;
import org.apache.lens.driver.jdbc.JDBCDriverConfConstants;
import org.apache.lens.driver.jdbc.JDBCResultSet;
+import org.apache.lens.server.api.driver.LensDriver;
import org.apache.lens.server.api.driver.LensResultSet;
import org.apache.lens.server.api.driver.LensResultSetMetadata;
import org.apache.lens.server.api.driver.InMemoryResultSet;
@@ -58,6 +63,8 @@ public class TestJdbcDriver {
/** The driver. */
JDBCDriver driver;
+ Collection<LensDriver> drivers;
+
/**
* Test create jdbc driver.
*
@@ -77,6 +84,9 @@ public class TestJdbcDriver {
driver.configure(baseConf);
assertNotNull(driver);
assertTrue(driver.configured);
+
+ drivers = new ArrayList<LensDriver>() {{
+ add(driver); }};
}
/**
@@ -90,6 +100,14 @@ public class TestJdbcDriver {
driver.close();
}
+
+ private QueryContext createQueryContext(final String query) throws LensException {
+ QueryContext context = new QueryContext(query, "SA", baseConf, drivers);
+ context.getDriverContext().setDriverQueriesAndPlans(new HashMap<LensDriver, String>() {{ put(driver, query); }} );
+ context.setSelectedDriver(driver);
+ return context;
+ }
+
/**
* Creates the table.
*
@@ -234,9 +252,9 @@ public class TestJdbcDriver {
insertData("execute_test");
// Query
- String query = "SELECT * FROM execute_test";
+ final String query = "SELECT * FROM execute_test";
- QueryContext context = new QueryContext(query, "SA", baseConf);
+ QueryContext context = createQueryContext(query);
LensResultSet resultSet = driver.execute(context);
assertNotNull(resultSet);
if (resultSet instanceof InMemoryResultSet) {
@@ -270,8 +288,10 @@ public class TestJdbcDriver {
createTable("prepare_test");
insertData("prepare_test");
- String query = "SELECT * from prepare_test";
- PreparedQueryContext pContext = new PreparedQueryContext(query, "SA", baseConf);
+ final String query = "SELECT * from prepare_test";
+ PreparedQueryContext pContext = new PreparedQueryContext(query, "SA", baseConf, drivers );
+ pContext.getDriverContext().setDriverQueriesAndPlans(new HashMap<LensDriver, String>() {{ put(driver, query); }});
+ pContext.setSelectedDriver(driver);
driver.prepare(pContext);
}
@@ -285,8 +305,8 @@ public class TestJdbcDriver {
public void testExecuteAsync() throws Exception {
createTable("execute_async_test");
insertData("execute_async_test");
- String query = "SELECT * FROM execute_async_test";
- QueryContext context = new QueryContext(query, "SA", baseConf);
+ final String query = "SELECT * FROM execute_async_test";
+ QueryContext context = createQueryContext(query);
System.out.println("@@@ Test_execute_async:" + context.getQueryHandle());
final CountDownLatch listenerNotificationLatch = new CountDownLatch(1);
@@ -377,8 +397,8 @@ public class TestJdbcDriver {
createTable("invalid_conn_close");
insertData("invalid_conn_close");
- String query = "SELECT * from invalid_conn_close2";
- QueryContext ctx = new QueryContext(query, "SA", baseConf);
+ final String query = "SELECT * from invalid_conn_close2";
+ QueryContext ctx = createQueryContext(query);
for (int i = 0; i < JDBCDriverConfConstants.JDBC_POOL_MAX_SIZE_DEFAULT; i++) {
driver.executeAsync(ctx);
@@ -387,7 +407,7 @@ public class TestJdbcDriver {
}
String validQuery = "SELECT * FROM invalid_conn_close";
- QueryContext validCtx = new QueryContext(validQuery, "SA", baseConf);
+ QueryContext validCtx = createQueryContext(validQuery);
System.out.println("@@@ Submitting valid query");
driver.executeAsync(validCtx);
@@ -414,8 +434,8 @@ public class TestJdbcDriver {
createTable("valid_conn_close");
insertData("valid_conn_close");
- String query = "SELECT * from valid_conn_close";
- QueryContext ctx = new QueryContext(query, "SA", baseConf);
+ final String query = "SELECT * from valid_conn_close";
+ QueryContext ctx = createQueryContext(query);
for (int i = 0; i < JDBCDriverConfConstants.JDBC_POOL_MAX_SIZE_DEFAULT; i++) {
LensResultSet resultSet = driver.execute(ctx);
@@ -438,7 +458,7 @@ public class TestJdbcDriver {
}
String validQuery = "SELECT * FROM valid_conn_close";
- QueryContext validCtx = new QueryContext(validQuery, "SA", baseConf);
+ QueryContext validCtx = createQueryContext(validQuery);
System.out.println("@@@ Submitting query after pool quota used");
driver.execute(validCtx);
}
@@ -453,8 +473,8 @@ public class TestJdbcDriver {
public void testCancelQuery() throws Exception {
createTable("cancel_query_test");
insertData("cancel_query_test");
- String query = "SELECT * FROM cancel_query_test";
- QueryContext context = new QueryContext(query, "SA", baseConf);
+ final String query = "SELECT * FROM cancel_query_test";
+ QueryContext context = createQueryContext(query);
System.out.println("@@@ test_cancel:" + context.getQueryHandle());
driver.executeAsync(context);
QueryHandle handle = context.getQueryHandle();
@@ -474,8 +494,8 @@ public class TestJdbcDriver {
*/
@Test
public void testInvalidQuery() throws Exception {
- String query = "SELECT * FROM invalid_table";
- QueryContext ctx = new QueryContext(query, "SA", baseConf);
+ final String query = "SELECT * FROM invalid_table";
+ QueryContext ctx = createQueryContext(query);
try {
LensResultSet rs = driver.execute(ctx);
fail("Should have thrown exception");
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-query-lib/src/test/java/org/apache/lens/lib/query/TestAbstractFileFormatter.java
----------------------------------------------------------------------
diff --git a/lens-query-lib/src/test/java/org/apache/lens/lib/query/TestAbstractFileFormatter.java b/lens-query-lib/src/test/java/org/apache/lens/lib/query/TestAbstractFileFormatter.java
index c0b5177..b1b79c3 100644
--- a/lens-query-lib/src/test/java/org/apache/lens/lib/query/TestAbstractFileFormatter.java
+++ b/lens-query-lib/src/test/java/org/apache/lens/lib/query/TestAbstractFileFormatter.java
@@ -25,8 +25,11 @@ import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hive.service.cli.ColumnDescriptor;
+import org.apache.lens.api.LensException;
import org.apache.lens.server.api.LensConfConstants;
+import org.apache.lens.server.api.driver.LensDriver;
import org.apache.lens.server.api.driver.LensResultSetMetadata;
+import org.apache.lens.server.api.driver.MockDriver;
import org.apache.lens.server.api.query.QueryContext;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -212,7 +215,17 @@ public abstract class TestAbstractFileFormatter {
*/
protected void testFormatter(Configuration conf, String charsetEncoding, String outputParentDir, String fileExtn,
LensResultSetMetadata columnNames) throws IOException {
- QueryContext ctx = new QueryContext("test writer query", "testuser", null, conf);
+
+ final LensDriver mockDriver = new MockDriver();
+ try {
+ mockDriver.configure(conf);
+ } catch (LensException e) {
+ Assert.fail(e.getMessage());
+ }
+ QueryContext ctx = new QueryContext("test writer query", "testuser", conf, new ArrayList<LensDriver>() {{
+ add(mockDriver); }} );
+
+ ctx.setSelectedDriver(mockDriver);
formatter = createFormatter();
formatter.init(ctx, columnNames);
@@ -337,7 +350,7 @@ public abstract class TestAbstractFileFormatter {
csvRows.add("\"4\",\"NULL\",\"NULL\",\"NULL\",\"1,2,NULL,4\",\"4:NULL\",\"1=one,2=two,NULL=three,4=NULL\"");
csvRows
.add("\"NULL\",\"NULL\",\"NULL\",\"NULL\",\"1,2,NULL,4,NULL\",\"NULL:NULL\","
- + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
+ + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
csvRows.add("Total rows:5");
return csvRows;
}
@@ -363,7 +376,7 @@ public abstract class TestAbstractFileFormatter {
csvRows.add("\"4\",\"NULL\",\"NULL\",\"NULL\",\"1,2,NULL,4\",\"4:NULL\",\"1=one,2=two,NULL=three,4=NULL\"");
csvRows
.add("\"NULL\",\"NULL\",\"NULL\",\"NULL\",\"1,2,NULL,4,NULL\",\"NULL:NULL\","
- + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
+ + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
csvRows.add("Total rows:5");
return csvRows;
}
@@ -421,7 +434,7 @@ public abstract class TestAbstractFileFormatter {
.add("\"firstcol\",\"format(secondcol,2)\",\"thirdcol\",\"fourthcol\",\"fifthcol\",\"sixthcol\",\"seventhcol\"");
csvRows
.add("\"NULL\",\"NULL\",\"NULL\",\"NULL\",\"1,2,NULL,4,NULL\",\"NULL:NULL\","
- + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
+ + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
csvRows.add("Total rows:5");
return csvRows;
}
@@ -451,7 +464,7 @@ public abstract class TestAbstractFileFormatter {
csvRows.add("\"firstcol\",\"secondcol\",\"thirdcol\",\"fourthcol\",\"fifthcol\",\"sixthcol\",\"seventhcol\"");
csvRows
.add("\"NULL\",\"NULL\",\"NULL\",\"NULL\",\"1,2,NULL,4,NULL\",\"NULL:NULL\","
- + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
+ + "\"1=one,2=two,NULL=three,4=NULL,5=NULL\"");
csvRows.add("Total rows:5");
return csvRows;
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MinQueryCostSelector.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MinQueryCostSelector.java b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MinQueryCostSelector.java
index 6c42795..5ad1083 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MinQueryCostSelector.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MinQueryCostSelector.java
@@ -38,10 +38,11 @@ public class MinQueryCostSelector implements DriverSelector {
@Override
public LensDriver select(final AbstractQueryContext ctx,
final Configuration conf) {
- return Collections.min(ctx.getDriverQueries().keySet(), new Comparator<LensDriver>() {
+ return Collections.min(ctx.getDriverContext().getDrivers(), new Comparator<LensDriver>() {
@Override
public int compare(LensDriver d1, LensDriver d2) {
- return comparePlans(ctx.getDriverQueryPlans().get(d1), ctx.getDriverQueryPlans().get(d2));
+ return comparePlans(ctx.getDriverContext().getDriverQueryPlan(d1), ctx
+ .getDriverContext().getDriverQueryPlan(d2));
}
});
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockDriver.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockDriver.java b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockDriver.java
index 6763e2e..833edc5 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockDriver.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockDriver.java
@@ -193,7 +193,7 @@ public class MockDriver implements LensDriver {
*/
@Override
public DriverQueryPlan explainAndPrepare(PreparedQueryContext pContext) throws LensException {
- DriverQueryPlan p = new MockQueryPlan(pContext.getDriverQuery());
+ DriverQueryPlan p = new MockQueryPlan(pContext.getSelectedDriverQuery());
p.setPrepareHandle(pContext.getPrepareHandle());
return p;
}
@@ -216,7 +216,7 @@ public class MockDriver implements LensDriver {
*/
@Override
public LensResultSet execute(QueryContext context) throws LensException {
- this.query = context.getDriverQuery();
+ this.query = context.getSelectedDriverQuery();
return new PersistentResultSet() {
@Override
@@ -253,7 +253,7 @@ public class MockDriver implements LensDriver {
*/
@Override
public void executeAsync(QueryContext context) throws LensException {
- this.query = context.getDriverQuery();
+ this.query = context.getSelectedDriverQuery();
}
/*
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockQueryContext.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockQueryContext.java b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockQueryContext.java
index c4d6b6d..7c31338 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockQueryContext.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/driver/MockQueryContext.java
@@ -18,13 +18,17 @@
*/
package org.apache.lens.server.api.driver;
+import java.util.Collection;
import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensException;
import org.apache.lens.server.api.query.AbstractQueryContext;
public class MockQueryContext extends AbstractQueryContext {
- public MockQueryContext(Map<LensDriver, String> driverQueries) throws LensException {
- setDriverQueriesAndPlans(driverQueries);
+
+ public MockQueryContext(final String query, final LensConf qconf, final Configuration conf, final Collection<LensDriver> drivers) {
+ super(query, qconf, conf, drivers);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/query/AbstractQueryContext.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/AbstractQueryContext.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/AbstractQueryContext.java
index 0b3e976..45770a4 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/AbstractQueryContext.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/AbstractQueryContext.java
@@ -19,13 +19,19 @@
package org.apache.lens.server.api.query;
import lombok.Getter;
+import lombok.Setter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.lens.api.LensConf;
+import org.apache.lens.api.LensException;
+import org.apache.lens.server.api.driver.DriverQueryPlan;
+import org.apache.lens.server.api.driver.LensDriver;
import java.io.Serializable;
+import java.util.Collection;
-public abstract class AbstractQueryContext extends DriverSelectorQueryContext implements Serializable {
+public abstract class AbstractQueryContext implements Serializable {
/** The Constant LOG */
public static final Log LOG = LogFactory.getLog(AbstractQueryContext.class);
@@ -33,7 +39,97 @@ public abstract class AbstractQueryContext extends DriverSelectorQueryContext im
@Getter
protected String userQuery;
- /** The qconf. */
+ /** The merged Query conf. */
+ @Getter @Setter
+ transient protected Configuration conf;
+
+ /** The query conf. */
+ @Getter
+ protected LensConf lensConf;
+
+ /** The driver ctx */
@Getter
- protected LensConf qconf;
+ @Setter
+ transient protected DriverSelectorQueryContext driverContext;
+
+ /** The selected Driver query. */
+ @Getter
+ protected String driverQuery;
+
+ protected AbstractQueryContext(final String query, final LensConf qconf, final Configuration conf, final
+ Collection<LensDriver> drivers) {
+ driverContext = new DriverSelectorQueryContext(query, conf, drivers);
+ userQuery = query;
+ this.lensConf = qconf;
+ this.conf = conf;
+ this.driverQuery = query;
+ }
+
+ /** Wrapper method for convenience on driver context
+ *
+ * @return the selected driver's query
+ */
+ public String getSelectedDriverQuery() {
+ if(driverQuery != null) {
+ return driverQuery;
+ } else if(driverContext != null) {
+ return driverContext.getSelectedDriverQuery();
+ }
+ return null;
+ }
+
+ /** Wrapper method for convenience on driver context
+ *
+ * @return the selected driver's conf
+ */
+ public Configuration getSelectedDriverConf() {
+ if(driverContext != null) {
+ return driverContext.getSelectedDriverConf();
+ }
+ return null;
+ }
+
+ /**
+ * Sets the selected driver query for persistence and also in the driver context
+ * @param driverQuery
+ */
+ public void setSelectedDriverQuery(String driverQuery) {
+ this.driverQuery = driverQuery;
+ if(driverContext != null) {
+ driverContext.setSelectedDriverQuery(driverQuery);
+ }
+ }
+
+ /** Wrapper method for convenience on driver context
+ * @param driver Lens driver
+ */
+
+ public void setSelectedDriver(LensDriver driver) {
+ if(driverContext != null) {
+ driverContext.setSelectedDriver(driver);
+ driverQuery = driverContext.getSelectedDriverQuery();
+ }
+ }
+
+ /** Wrapper method for convenience on driver context
+ *
+ * @return the selected driver
+ */
+ public LensDriver getSelectedDriver() {
+ if(driverContext != null) {
+ return driverContext.getSelectedDriver();
+ }
+ return null;
+ }
+
+ /** Wrapper method for convenience on driver context
+ *
+ * @return the selected driver
+ */
+ public DriverQueryPlan getSelectedDriverQueryPlan() throws LensException {
+ if(driverContext != null) {
+ return driverContext.getSelectedDriverQueryPlan();
+ }
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/query/DriverSelectorQueryContext.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/DriverSelectorQueryContext.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/DriverSelectorQueryContext.java
index 0098a1a..6446297 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/DriverSelectorQueryContext.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/DriverSelectorQueryContext.java
@@ -27,58 +27,104 @@ import org.apache.lens.api.LensException;
import org.apache.lens.server.api.driver.DriverQueryPlan;
import org.apache.lens.server.api.driver.LensDriver;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-public abstract class DriverSelectorQueryContext {
+public class DriverSelectorQueryContext {
/** The constant LOG */
public static final Log LOG = LogFactory.getLog(DriverSelectorQueryContext.class);
- /** The conf. */
- @Getter @Setter
- transient protected Configuration conf;
/** The selected driver. */
@Getter
@Setter
- transient protected LensDriver selectedDriver;
+ protected LensDriver selectedDriver;
- /** The driver query. */
+ /** Map of driver to driver specific query context */
@Getter
@Setter
- protected String driverQuery;
- /** Map of driver to driver query */
- @Getter
- protected Map<LensDriver, String> driverQueries;
+ protected Map<LensDriver, DriverQueryContext> driverQueryContextMap = new HashMap<LensDriver,
+ DriverQueryContext>();
- /** Map of driver to query plan */
- @Getter
- protected Map<LensDriver, DriverQueryPlan> driverQueryPlans;
+ public DriverSelectorQueryContext(final String userQuery, final Configuration queryConf, final Collection<LensDriver>
+ drivers) {
+ for(LensDriver driver : drivers) {
+ DriverQueryContext ctx = new DriverQueryContext(driver);
+ ctx.setDriverSpecificConf(mergeConf(driver, queryConf));
+ ctx.setQuery(userQuery);
+ driverQueryContextMap.put(driver, ctx);
+ }
+ }
+
+ public static class DriverQueryContext {
+
+ @Getter
+ protected LensDriver driver;
+
+ DriverQueryContext(LensDriver driver) {
+ this.driver = driver;
+ }
+
+ /** Map of driver to query plan */
+ @Getter
+ @Setter
+ protected DriverQueryPlan driverQueryPlan;
+
+ /** driver specific query conf */
+ @Getter
+ @Setter
+ protected Configuration driverSpecificConf;
+
+ @Getter
+ @Setter
+ /** exceptions occurred while trying to generate plans by explain call */
+ protected Exception driverQueryPlanGenerationError;
- /** Map of exceptions occurred while trying to generate plans by explain call */
- protected Map<LensDriver, Exception> driverQueryPlanGenerationErrors;
+ @Getter
+ @Setter
+ /** driver query */
+ protected String query;
+
+ }
+
+ /**
+ * Gets the driver query conf.
+ *
+ * @param driver
+ * the driver
+ * @param queryConf
+ * the query conf
+ * @return the final query conf
+ */
+ private Configuration mergeConf(LensDriver driver, Configuration queryConf) {
+ Configuration conf = new Configuration(driver.getConf());
+ for (Map.Entry<String, String> entry : queryConf) {
+ conf.set(entry.getKey(), entry.getValue());
+ }
+ conf.setClassLoader(queryConf.getClassLoader());
+ return conf;
+ }
/**
* Sets driver queries, generates plans for each driver by calling explain with respective queries,
* Sets driverQueryPlans
* @param driverQueries
* @throws LensException
- * @see #driverQueryPlans
*/
public void setDriverQueriesAndPlans(Map<LensDriver, String> driverQueries) throws LensException {
- driverQueryPlanGenerationErrors = new HashMap<LensDriver, Exception>();
- this.driverQueries = driverQueries;
- this.driverQueryPlans = new HashMap<LensDriver, DriverQueryPlan>();
- for (LensDriver driver : driverQueries.keySet()) {
- DriverQueryPlan plan = null;
+ for(LensDriver driver : driverQueries.keySet()) {
+ final DriverQueryContext driverQueryContext = driverQueryContextMap.get(driver);
+ driverQueryContext.setQuery(driverQueries.get(driver));
try {
- plan = driver.explain(driverQueries.get(driver), getConf());
+ driverQueryContext.setDriverQueryPlan(driver.explain(driverQueries.get(driver), driverQueryContext.getDriverSpecificConf()));
} catch (Exception e) {
LOG.error(e.getStackTrace());
- driverQueryPlanGenerationErrors.put(driver, e);
+ driverQueryContext.setDriverQueryPlanGenerationError(e);
}
- driverQueryPlans.put(driver, plan);
}
}
@@ -88,16 +134,70 @@ public abstract class DriverSelectorQueryContext {
* @throws LensException
*/
public DriverQueryPlan getSelectedDriverQueryPlan() throws LensException {
- if(getDriverQueryPlans() == null) {
- throw new LensException("No Driver query plans. Check if re-write happened or not");
+ final Map<LensDriver, DriverQueryContext> driverQueryCtxs = getDriverQueryContextMap();
+ if(driverQueryCtxs == null) {
+ throw new LensException("No Driver query ctx. Check if re-write happened or not");
}
if(getSelectedDriver() == null) {
throw new LensException("Selected Driver is NULL.");
}
- if(getDriverQueryPlans().get(getSelectedDriver()) == null) {
+
+ if(driverQueryCtxs.get(getSelectedDriver()) == null) {
+ throw new LensException("Could not find Driver Context for selected driver " + getSelectedDriver());
+ }
+
+ if(driverQueryCtxs.get(getSelectedDriver()).getDriverQueryPlanGenerationError() != null) {
throw new LensException("Driver Query Plan of the selected driver is null",
- driverQueryPlanGenerationErrors.get(getSelectedDriver()));
+ driverQueryCtxs.get(getSelectedDriver()).getDriverQueryPlanGenerationError());
}
- return getDriverQueryPlans().get(getSelectedDriver());
+ return driverQueryCtxs.get(getSelectedDriver()).getDriverQueryPlan();
+ }
+
+ public Configuration getSelectedDriverConf() {
+ return driverQueryContextMap.get(getSelectedDriver()).getDriverSpecificConf();
+ }
+
+ public String getSelectedDriverQuery() {
+ return driverQueryContextMap.get(getSelectedDriver()).getQuery();
+ }
+
+ public void setDriverConf(LensDriver driver, Configuration conf) {
+ driverQueryContextMap.get(driver).setDriverSpecificConf(conf);
+ }
+
+ public void setSelectedDriverQuery(String driverQuery) {
+ if(driverQueryContextMap != null && driverQueryContextMap.get(getSelectedDriver()) != null) {
+ driverQueryContextMap.get(getSelectedDriver()).setQuery(driverQuery);
+ }
+ }
+
+ public Collection<LensDriver> getDrivers() {
+ return driverQueryContextMap.keySet();
+ }
+
+ public Collection<String> getDriverQueries() {
+ List<String> queries = new ArrayList<String>();
+ final Collection<DriverQueryContext> values = driverQueryContextMap.values();
+ for(DriverQueryContext ctx : values) {
+ if(ctx.getQuery() != null) {
+ queries.add(ctx.getQuery());
+ }
+ }
+ return queries;
+ }
+
+ public DriverQueryPlan getDriverQueryPlan(LensDriver driver) {
+ return driverQueryContextMap.get(driver) != null ?
+ driverQueryContextMap.get(driver).getDriverQueryPlan() : null;
+ }
+
+ public Configuration getDriverConf(LensDriver driver) {
+ return driverQueryContextMap.get(driver) != null ?
+ driverQueryContextMap.get(driver).getDriverSpecificConf() : null;
+ }
+
+ public String getDriverQuery(LensDriver driver) {
+ return driverQueryContextMap.get(driver) != null ?
+ driverQueryContextMap.get(driver).getQuery() : null;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/query/ExplainQueryContext.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/ExplainQueryContext.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/ExplainQueryContext.java
index e248437..e4f3140 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/ExplainQueryContext.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/ExplainQueryContext.java
@@ -19,6 +19,10 @@
package org.apache.lens.server.api.query;
import org.apache.hadoop.conf.Configuration;
+import org.apache.lens.api.LensConf;
+import org.apache.lens.server.api.driver.LensDriver;
+
+import java.util.Collection;
/**
* The class ExplainQueryContext
@@ -29,8 +33,7 @@ public class ExplainQueryContext extends AbstractQueryContext{
* @param query
* @param qconf
*/
- public ExplainQueryContext(String query, Configuration qconf) {
- userQuery = query;
- conf = qconf;
+ public ExplainQueryContext(String query, LensConf conf, Configuration qconf, Collection<LensDriver> drivers) {
+ super(query, conf, qconf, drivers);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/lens-server-api/src/main/java/org/apache/lens/server/api/query/PreparedQueryContext.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/query/PreparedQueryContext.java b/lens-server-api/src/main/java/org/apache/lens/server/api/query/PreparedQueryContext.java
index ed592a5..5cc3b0f 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/query/PreparedQueryContext.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/query/PreparedQueryContext.java
@@ -18,6 +18,7 @@
*/
package org.apache.lens.server.api.query;
+import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
@@ -31,6 +32,7 @@ import org.apache.lens.api.query.QueryPrepareHandle;
import lombok.Getter;
import lombok.Setter;
+import org.apache.lens.server.api.driver.LensDriver;
/**
* The Class PreparedQueryContext.
@@ -67,8 +69,8 @@ public class PreparedQueryContext extends AbstractQueryContext implements Delaye
* @param conf
* the conf
*/
- public PreparedQueryContext(String query, String user, Configuration conf) {
- this(query, user, conf, new LensConf());
+ public PreparedQueryContext(String query, String user, Configuration conf, Collection<LensDriver> drivers) {
+ this(query, user, conf, new LensConf(), drivers);
}
/**
@@ -83,14 +85,14 @@ public class PreparedQueryContext extends AbstractQueryContext implements Delaye
* @param qconf
* the qconf
*/
- public PreparedQueryContext(String query, String user, Configuration conf, LensConf qconf) {
- this.userQuery = query;
+ public PreparedQueryContext(String query, String user, Configuration conf, LensConf qconf, Collection<LensDriver>
+ drivers) {
+ super(query, qconf, conf, drivers);
this.preparedTime = new Date();
this.preparedUser = user;
this.prepareHandle = new QueryPrepareHandle(UUID.randomUUID());
this.conf = conf;
- this.qconf = qconf;
- this.driverQuery = query;
+ this.lensConf = qconf;
}
/*
@@ -128,7 +130,7 @@ public class PreparedQueryContext extends AbstractQueryContext implements Delaye
* the conf to set
*/
public void updateConf(Map<String, String> confoverlay) {
- qconf.getProperties().putAll(confoverlay);
+ lensConf.getProperties().putAll(confoverlay);
for (Map.Entry<String, String> prop : confoverlay.entrySet()) {
this.conf.set(prop.getKey(), prop.getValue());
}
@@ -141,6 +143,8 @@ public class PreparedQueryContext extends AbstractQueryContext implements Delaye
*/
public LensPreparedQuery toPreparedQuery() {
return new LensPreparedQuery(prepareHandle, userQuery, preparedTime, preparedUser,
- selectedDriver != null ? selectedDriver.getClass().getCanonicalName() : null, driverQuery, qconf);
+ getDriverContext().getSelectedDriver() != null ? getDriverContext().getSelectedDriver().getClass()
+ .getCanonicalName() : null, getDriverContext().getSelectedDriverQuery(),
+ lensConf);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-lens/blob/fcbb44a1/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 04b8aa4..82cb165 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
@@ -18,6 +18,7 @@
*/
package org.apache.lens.server.api.query;
+import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
@@ -127,6 +128,20 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
@Setter
private String queryName;
+ /**
+ * Instantiates a new query context.
+ *
+ * @param query
+ * the query
+ * @param user
+ * the user
+ * @param conf
+ * the conf
+ */
+ public QueryContext(String query, String user, Configuration conf, Collection<LensDriver> drivers) {
+ this(query, user, new LensConf(), conf, drivers, new Date().getTime());
+ }
+
/**
* Instantiates a new query context.
*
@@ -137,8 +152,9 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
* @param conf
* the conf
*/
- public QueryContext(String query, String user, Configuration conf) {
- this(query, user, new LensConf(), conf, query, null, new Date().getTime());
+ public QueryContext(String query, String user, Configuration conf, Collection<LensDriver> drivers, long
+ submissionTime) {
+ this(query, user, new LensConf(), conf, drivers, submissionTime);
}
/**
@@ -152,23 +168,30 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
* the qconf
* @param conf
* the conf
+ * @param drivers
+ * Collection of drivers
*/
- public QueryContext(String query, String user, LensConf qconf, Configuration conf) {
- this(query, user, qconf, conf, query, null, new Date().getTime());
+ public QueryContext(String query, String user, LensConf qconf, Configuration conf, Collection<LensDriver> drivers) {
+ this(query, user, qconf, conf, drivers, null, new Date().getTime());
}
/**
* Instantiates a new query context.
*
- * @param prepared
- * the prepared
+ * @param query
+ * the query
* @param user
* the user
+ * @param qconf
+ * the qconf
* @param conf
* the conf
+ * @param drivers
+ * Collection of drivers
*/
- public QueryContext(PreparedQueryContext prepared, String user, Configuration conf) {
- this(prepared, user, new LensConf(), conf);
+ public QueryContext(String query, String user, LensConf qconf, Configuration conf, Collection<LensDriver> drivers,
+ Long submissionTime) {
+ this(query, user, qconf, conf, drivers, null, submissionTime);
}
/**
@@ -178,35 +201,35 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
* the prepared
* @param user
* the user
- * @param qconf
- * the qconf
* @param conf
* the conf
*/
- public QueryContext(PreparedQueryContext prepared, String user, LensConf qconf, Configuration conf) {
- this(prepared.getUserQuery(), user, qconf, mergeConf(prepared.getConf(), conf), prepared.getDriverQuery(), prepared
- .getSelectedDriver(), new Date().getTime());
- driverQueries = prepared.getDriverQueries();
- driverQueryPlans = prepared.getDriverQueryPlans();
+ public QueryContext(PreparedQueryContext prepared, String user, Configuration conf) {
+ this(prepared, user, new LensConf(), conf);
}
/**
* Instantiates a new query context.
*
- * @param query
- * the query
+ * @param prepared
+ * the prepared
* @param user
* the user
+ * @param qconf
+ * the qconf
* @param conf
* the conf
- * @param submissionTime
- * the submission time
*/
- public QueryContext(String query, String user, Configuration conf, long submissionTime) {
- this(query, user, new LensConf(), conf, query, null, submissionTime);
+ public QueryContext(PreparedQueryContext prepared, String user, LensConf qconf, Configuration conf) {
+ this(prepared.getUserQuery(), user, qconf, mergeConf(prepared.getConf(), conf), prepared.getDriverContext()
+ .getDriverQueryContextMap().keySet(),
+ prepared.getDriverContext()
+ .getSelectedDriver(), new Date().getTime());
+ setDriverContext(prepared.getDriverContext());
+ setSelectedDriverQuery(prepared.getDriverQuery());
}
- /**
+ /**
* Instantiates a new query context.
*
* @param userQuery
@@ -217,19 +240,19 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
* the qconf
* @param conf
* the conf
- * @param driverQuery
- * the driver query
* @param selectedDriver
* the selected driver
* @param submissionTime
* the submission time
*/
- public QueryContext(String userQuery, String user, LensConf qconf, Configuration conf, String driverQuery,
- LensDriver selectedDriver, long submissionTime) {
+ public QueryContext(String userQuery, String user, LensConf qconf, Configuration conf,
+ Collection<LensDriver> drivers, LensDriver selectedDriver, long submissionTime) {
+ super(userQuery, qconf, conf, drivers);
this.submissionTime = submissionTime;
this.queryHandle = new QueryHandle(UUID.randomUUID());
this.status = new QueryStatus(0.0f, Status.NEW, "Query just got created", false, null, null);
this.priority = Priority.NORMAL;
+ this.lensConf = qconf;
this.conf = conf;
this.isPersistent = conf.getBoolean(LensConfConstants.QUERY_PERSISTENT_RESULT_SET,
LensConfConstants.DEFAULT_PERSISTENT_RESULT_SET);
@@ -237,12 +260,15 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
LensConfConstants.DEFAULT_DRIVER_PERSISTENT_RESULT_SET);
this.userQuery = userQuery;
this.submittedUser = user;
- this.driverQuery = driverQuery;
- this.selectedDriver = selectedDriver;
- this.qconf = qconf;
+ if(selectedDriver != null) {
+ this.setSelectedDriver(selectedDriver);
+ }
+ this.lensConf = qconf;
this.driverStatus = new DriverQueryStatus();
}
+
+
/**
* Merge conf.
*
@@ -285,7 +311,7 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
* the conf to set
*/
public void updateConf(Map<String, String> confoverlay) {
- qconf.getProperties().putAll(confoverlay);
+ lensConf.getProperties().putAll(confoverlay);
for (Map.Entry<String, String> prop : confoverlay.entrySet()) {
this.conf.set(prop.getKey(), prop.getValue());
}
@@ -307,8 +333,11 @@ public class QueryContext extends AbstractQueryContext implements Comparable<Que
*/
public LensQuery toLensQuery() {
return new LensQuery(queryHandle, userQuery, submittedUser, priority, isPersistent,
- getSelectedDriver() != null ? getSelectedDriver().getClass().getCanonicalName() : null, getDriverQuery(), status,
- resultSetPath, driverOpHandle, qconf, submissionTime, launchTime, driverStatus.getDriverStartTime(),
+ getSelectedDriver() != null ? getSelectedDriver().getClass()
+ .getCanonicalName() : null,
+ getSelectedDriverQuery(),
+ status,
+ resultSetPath, driverOpHandle, lensConf, submissionTime, launchTime, driverStatus.getDriverStartTime(),
driverStatus.getDriverFinishTime(), endTime, closedTime, queryName);
}