|
@@ -112,7 +112,7 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
private class RunningQueryState {
|
|
private class RunningQueryState {
|
|
|
private QueryState state = QueryState.CREATED;
|
|
private QueryState state = QueryState.CREATED;
|
|
|
// Thread local used by Hive quite a bit.
|
|
// Thread local used by Hive quite a bit.
|
|
|
- private SessionState sessionState;
|
|
|
|
|
|
|
+ private CleanableSessionState sessionState;
|
|
|
private Throwable exception;
|
|
private Throwable exception;
|
|
|
private Driver driver;
|
|
private Driver driver;
|
|
|
private ByteArrayOutputStream errStream = new ByteArrayOutputStream();
|
|
private ByteArrayOutputStream errStream = new ByteArrayOutputStream();
|
|
@@ -179,9 +179,13 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
synchronized public void compile() throws BeeswaxException {
|
|
synchronized public void compile() throws BeeswaxException {
|
|
|
- assertState(QueryState.INITIALIZED);
|
|
|
|
|
- checkedCompile();
|
|
|
|
|
- state = QueryState.COMPILED;
|
|
|
|
|
|
|
+ try {
|
|
|
|
|
+ assertState(QueryState.INITIALIZED);
|
|
|
|
|
+ checkedCompile();
|
|
|
|
|
+ state = QueryState.COMPILED;
|
|
|
|
|
+ } finally {
|
|
|
|
|
+ cleanSessionState();
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
private void assertState(QueryState expected) {
|
|
private void assertState(QueryState expected) {
|
|
@@ -267,8 +271,8 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
}
|
|
}
|
|
|
hiveConf.setClassLoader(loader);
|
|
hiveConf.setClassLoader(loader);
|
|
|
Thread.currentThread().setContextClassLoader(loader);
|
|
Thread.currentThread().setContextClassLoader(loader);
|
|
|
- SessionState.start(hiveConf); // this is thread-local
|
|
|
|
|
- this.sessionState = SessionState.get();
|
|
|
|
|
|
|
+ this.sessionState = new CleanableSessionState(hiveConf);
|
|
|
|
|
+ SessionState.start(this.sessionState);
|
|
|
|
|
|
|
|
// If this work has a LogContext, associate the children output to the logContext
|
|
// If this work has a LogContext, associate the children output to the logContext
|
|
|
OutputStream lcOutStream = null;
|
|
OutputStream lcOutStream = null;
|
|
@@ -426,6 +430,7 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
} finally {
|
|
} finally {
|
|
|
// Don't let folks re-use the state object.
|
|
// Don't let folks re-use the state object.
|
|
|
state = QueryState.FINISHED;
|
|
state = QueryState.FINISHED;
|
|
|
|
|
+ cleanSessionState();
|
|
|
}
|
|
}
|
|
|
return new QueryExplanation(sb.toString());
|
|
return new QueryExplanation(sb.toString());
|
|
|
}
|
|
}
|
|
@@ -446,6 +451,8 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
materializeResults(r, fromBeginning);
|
|
materializeResults(r, fromBeginning);
|
|
|
} catch (IOException e) {
|
|
} catch (IOException e) {
|
|
|
throw new BeeswaxException(e.toString(), logContext.getName(), handle);
|
|
throw new BeeswaxException(e.toString(), logContext.getName(), handle);
|
|
|
|
|
+ } finally {
|
|
|
|
|
+ cleanSessionState();
|
|
|
}
|
|
}
|
|
|
break;
|
|
break;
|
|
|
case EXCEPTION:
|
|
case EXCEPTION:
|
|
@@ -498,6 +505,8 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
} catch (Throwable t) {
|
|
} catch (Throwable t) {
|
|
|
LOG.error("Exception while processing query", t);
|
|
LOG.error("Exception while processing query", t);
|
|
|
state.saveException(t);
|
|
state.saveException(t);
|
|
|
|
|
+ } finally {
|
|
|
|
|
+ cleanSessionState();
|
|
|
}
|
|
}
|
|
|
return null;
|
|
return null;
|
|
|
}
|
|
}
|
|
@@ -505,6 +514,10 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
|
|
|
}
|
|
}
|
|
|
});
|
|
});
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ private void cleanSessionState() {
|
|
|
|
|
+ ((CleanableSessionState) SessionState.get()).destroyHiveHistory();
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|