if (sessionId != null) {
sessionName = sessionId;
}
Leases.Lease lease = null;
ExecutionEngineSession session = null;
boolean lastScan = false;
if (StringUtils.isNotEmpty(sessionId)) {
session = sessions.get(sessionName);
if (!closeSession) {
if (session == null) {
throw new UnknownSessionException("Name: " + sessionName
+ ", already closed? sessions size is " + sessions.size());
}
}
} else {
session = addSession();
sessionName = session.getSessionId();
if (plan instanceof LocalQueryPlan) {
LocalQueryPlan localQueryPlan = (LocalQueryPlan) plan;
if (localQueryPlan.getScanAction() != null) {
session.setExecutor(new LocalScanExecutor(localQueryPlan));
} else if (localQueryPlan.getGetAction() != null) {
localQueryPlan.getGetAction().setSessionId(sessionName);
session.setExecutor(new LocalGetExecutor(localQueryPlan));
}
} else if (plan instanceof GlobalQueryPlan) {
session.setExecutor(new GlobalQueryExecutor((GlobalQueryPlan) plan,
server));
} else if (plan instanceof AggregateQueryPlan) {
session.setExecutor(new AggregateQueryExecutor(
(AggregateQueryPlan) plan, server));
}
}
if (closeSession) {
session = sessions.remove(sessionName);
if (session != null) {
session.close();
try {
leases.cancelLease(sessionName);
} catch (LeaseException e) {
LOG.warn("Lease " + sessionName
+ "does't exsit,may has been closed.");
}
session = null;
}
return new Pair<Boolean, Pair<String, Pair<List<ClientProtos.QueryResultProto>, List<ClientProtos.StringDataTypePair>>>>(
lastScan,
new Pair<String, Pair<List<ClientProtos.QueryResultProto>, List<ClientProtos.StringDataTypePair>>>(
sessionName,
new Pair<List<ClientProtos.QueryResultProto>, List<ClientProtos.StringDataTypePair>>(
new ArrayList<ClientProtos.QueryResultProto>(),
new ArrayList<ClientProtos.StringDataTypePair>())));
} else {
ExecutionEngineSession.QueryExecutor executor = null;
try {
lease = leases.removeLease(sessionName);
executor = (ExecutionEngineSession.QueryExecutor) session
.getExecutor();
results = executor.execute();
lastScan = executor.isLastScan();
return new Pair<Boolean, Pair<String, Pair<List<ClientProtos.QueryResultProto>, List<ClientProtos.StringDataTypePair>>>>(
lastScan,
new Pair<String, Pair<List<ClientProtos.QueryResultProto>, List<ClientProtos.StringDataTypePair>>>(
sessionName, results));
} finally {
// We're done. On way out re-add the above removed lease.
// Adding resets expiration time on lease.
if (!lastScan) {
if (sessions.containsKey(sessionName)) {
if (lease != null) {
leases.addLease(lease);
}
}
} else {
session = sessions.remove(sessionName);
if (session != null) {
session.close();
session = null;
}
}
}
}