BeeswaxServiceImpl.java 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860
  1. // Licensed to Cloudera, Inc. under one
  2. // or more contributor license agreements. See the NOTICE file
  3. // distributed with this work for additional information
  4. // regarding copyright ownership. Cloudera, Inc. licenses this file
  5. // to you under the Apache License, Version 2.0 (the
  6. // "License"); you may not use this file except in compliance
  7. // with the License. You may obtain a copy of the License at
  8. //
  9. // http://www.apache.org/licenses/LICENSE-2.0
  10. //
  11. // Unless required by applicable law or agreed to in writing, software
  12. // distributed under the License is distributed on an "AS IS" BASIS,
  13. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. // See the License for the specific language governing permissions and
  15. // limitations under the License.
  16. package com.cloudera.beeswax;
  17. import java.io.ByteArrayOutputStream;
  18. import java.io.File;
  19. import java.io.IOException;
  20. import java.io.OutputStream;
  21. import java.io.PrintStream;
  22. import java.io.UnsupportedEncodingException;
  23. import java.lang.reflect.UndeclaredThrowableException;
  24. import java.net.HttpURLConnection;
  25. import java.net.URL;
  26. import java.security.KeyManagementException;
  27. import java.security.NoSuchAlgorithmException;
  28. import java.security.PrivilegedActionException;
  29. import java.security.PrivilegedExceptionAction;
  30. import java.security.SecureRandom;
  31. import java.text.SimpleDateFormat;
  32. import java.util.ArrayList;
  33. import java.util.Collections;
  34. import java.util.concurrent.ConcurrentHashMap;
  35. import java.util.concurrent.Executors;
  36. import java.util.concurrent.ExecutorService;
  37. import java.util.Date;
  38. import java.util.List;
  39. import java.util.Map;
  40. import java.util.Map.Entry;
  41. import java.util.Properties;
  42. import java.util.UUID;
  43. import java.util.Vector;
  44. import javax.net.ssl.HttpsURLConnection;
  45. import javax.net.ssl.SSLContext;
  46. import org.apache.commons.lang.StringUtils;
  47. import org.apache.hadoop.fs.Path;
  48. import org.apache.hadoop.hive.conf.HiveConf;
  49. import org.apache.hadoop.hive.metastore.api.FieldSchema;
  50. import org.apache.hadoop.hive.metastore.api.Schema;
  51. import org.apache.hadoop.hive.ql.Driver;
  52. import org.apache.hadoop.hive.ql.exec.FetchTask;
  53. import org.apache.hadoop.hive.ql.exec.Utilities;
  54. import org.apache.hadoop.hive.ql.metadata.Hive;
  55. import org.apache.hadoop.hive.ql.metadata.HiveException;
  56. import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
  57. import org.apache.hadoop.hive.ql.parse.ExplainSemanticAnalyzer;
  58. import org.apache.hadoop.hive.ql.plan.fetchWork;
  59. import org.apache.hadoop.hive.ql.plan.tableDesc;
  60. import org.apache.hadoop.hive.ql.processors.CommandProcessor;
  61. import org.apache.hadoop.hive.ql.processors.CommandProcessorFactory;
  62. import org.apache.hadoop.hive.ql.QueryPlan;
  63. import org.apache.hadoop.hive.ql.session.SessionState;
  64. import org.apache.hadoop.hive.serde.Constants;
  65. import org.apache.hadoop.security.UserGroupInformation;
  66. import org.apache.log4j.Logger;
  67. import org.apache.thrift.TException;
  68. import com.cloudera.beeswax.api.BeeswaxException;
  69. import com.cloudera.beeswax.api.BeeswaxService;
  70. import com.cloudera.beeswax.api.ConfigVariable;
  71. import com.cloudera.beeswax.api.Query;
  72. import com.cloudera.beeswax.api.QueryExplanation;
  73. import com.cloudera.beeswax.api.QueryHandle;
  74. import com.cloudera.beeswax.api.QueryNotFoundException;
  75. import com.cloudera.beeswax.api.QueryState;
  76. import com.cloudera.beeswax.api.Results;
  77. import com.cloudera.beeswax.api.ResultsMetadata;
  78. /**
  79. * Implementation of BeeswaxService interface.
  80. */
  81. public class BeeswaxServiceImpl implements BeeswaxService.Iface {
  82. // We wrap each query in its own little Runnable, then use the executor to run them.
  83. private ExecutorService executor;
  84. // runningQueries keeps track of the concurrently running queries, with the query id as key.
  85. private ConcurrentHashMap<String, RunningQueryState> runningQueries;
  86. private String notifyUrl;
  87. /** Mapping between configuration variable names and descriptions. */
  88. private ConfigDescriptions configDescriptions = ConfigDescriptions.get();
  89. private static final long RUNNING_QUERY_LIFETIME = 7*24*60*60*1000; // 1 week
  90. private static final long EVICTION_INTERVAL = 3*60*60*1000; // 3 hours
  91. private static final String NOTIFY_URL_BASE = "/beeswax/query_cb/done/";
  92. private static Logger LOG = Logger.getLogger(BeeswaxServiceImpl.class.getName());
  93. /**
  94. * To be read and modified while holding a lock on the state object.
  95. *
  96. * Essentially, this annotates a Driver with a SessionState object.
  97. * These go together, but SessionState is accessed via a thread-local,
  98. * so manual management has to happen.
  99. *
  100. * The state changes are...
  101. * CREATED -> INITIALIZED -> COMPILED -> RUNNING -> FINISHED
  102. * EXCEPTION is also valid.
  103. */
  104. private class RunningQueryState {
  105. private QueryState state = QueryState.CREATED;
  106. // Thread local used by Hive quite a bit.
  107. private SessionState sessionState;
  108. private Throwable exception;
  109. private Driver driver;
  110. private ByteArrayOutputStream errStream = new ByteArrayOutputStream();
  111. private ByteArrayOutputStream outStream = new ByteArrayOutputStream();
  112. private long startRow = 0;
  113. private HiveConf hiveConf = null;
  114. private final Query query;
  115. private long atime = 0;
  116. private LogContext logContext;
  117. /** The client handle, if applicable */
  118. private QueryHandle handle = null;
  119. /**
  120. * Create an instance with the given query and LogContext.
  121. * @param query The Beeswax Thrift Query object.
  122. * @param logContext The context to associate with.
  123. */
  124. public RunningQueryState(Query query, LogContext logContext) {
  125. this.query = query;
  126. this.atime = System.currentTimeMillis();
  127. this.logContext = logContext;
  128. }
  129. public String toString() {
  130. return "RunningQueryState" + (handle == null ? "" : " id: " + handle.id) +
  131. " started " + new SimpleDateFormat().format(new Date(atime)) +
  132. "; (state " + state + "); query: " + query.query;
  133. }
  134. public long getAtime() {
  135. return this.atime;
  136. }
  137. public HiveConf getHiveConf() {
  138. return hiveConf;
  139. }
  140. public void setQueryHandle(QueryHandle handle) {
  141. this.handle = handle;
  142. }
  143. public QueryHandle getQueryHandle() {
  144. return handle;
  145. }
  146. public String getInfoStreamAsString() {
  147. sessionState.out.flush();
  148. sessionState.childOut.flush();
  149. try {
  150. return outStream.toString("UTF-8");
  151. } catch (UnsupportedEncodingException e) {
  152. throw new RuntimeException(e);
  153. }
  154. }
  155. public String getErrorStreamAsString() {
  156. sessionState.err.flush();
  157. sessionState.childErr.flush();
  158. try {
  159. return errStream.toString("UTF-8");
  160. } catch (UnsupportedEncodingException e) {
  161. throw new RuntimeException(e);
  162. }
  163. }
  164. synchronized public void compile() throws BeeswaxException {
  165. assertState(QueryState.INITIALIZED);
  166. checkedCompile();
  167. state = QueryState.COMPILED;
  168. }
  169. private void assertState(QueryState expected) {
  170. if (state != expected) {
  171. throw new IllegalStateException(String.format("Expected %s, but state is: %s",
  172. expected, state));
  173. }
  174. }
  175. /**
  176. * Set the state to EXCEPTION and remember the Throwable that is causing it.
  177. * Then rethrow the exception.
  178. */
  179. private <T extends Throwable> void throwException(T ex) throws T {
  180. saveException(ex);
  181. throw ex;
  182. }
  183. private void checkedCompile() throws BeeswaxException {
  184. // Run through configuration commands
  185. for (String cmd : query.configuration) {
  186. // This is pretty whacky; SET and ADD get treated differently
  187. // than CREATE TEMPORARY FUNCTION... The trimming logic
  188. // here is borrowed from CliDriver; oy.
  189. String cmd_trimmed = cmd.trim();
  190. String[] tokens = cmd_trimmed.split("\\s+");
  191. String cmd1 = cmd_trimmed.substring(tokens[0].length()).trim();
  192. CommandProcessor p = CommandProcessorFactory.get(tokens[0]);
  193. int res;
  194. if (p instanceof Driver) {
  195. res = p.run(cmd);
  196. } else {
  197. res = p.run(cmd1);
  198. }
  199. if (res != 0) {
  200. throwException(new RuntimeException(getErrorStreamAsString()));
  201. }
  202. }
  203. // Note that driver.compile() talks to HDFS, so it's
  204. // not as quick as one might think.
  205. int compileRes = driver.compile(query.query);
  206. if (compileRes != 0) {
  207. throwException(new BeeswaxException(getErrorStreamAsString(),
  208. this.logContext.getName(),
  209. this.handle));
  210. }
  211. }
  212. /**
  213. * Separate from constructor, because initialize() may
  214. * need to be called in a separate thread.
  215. */
  216. synchronized void initialize() {
  217. assertState(QueryState.CREATED);
  218. this.hiveConf = new HiveConf(Driver.class);
  219. // Update configuration with user/group info.
  220. if (query.hadoop_user == null) {
  221. throw new RuntimeException("User must be specified.");
  222. }
  223. // Update scratch dir (to have one per user)
  224. File scratchDir = new File("/tmp/hive-beeswax-" + query.hadoop_user);
  225. hiveConf.set(HiveConf.ConfVars.SCRATCHDIR.varname, scratchDir.getPath());
  226. // Create the temporary directory if necessary.
  227. // If mapred.job.tracker is set to local, this is used by MapRedTask.
  228. if (!scratchDir.isDirectory()) {
  229. if (scratchDir.exists() || !scratchDir.mkdirs()) {
  230. LOG.warn("Could not create tmp dir:" + scratchDir);
  231. }
  232. }
  233. driver = new Driver(hiveConf);
  234. ClassLoader loader = hiveConf.getClassLoader();
  235. String auxJars = HiveConf.getVar(hiveConf, HiveConf.ConfVars.HIVEAUXJARS);
  236. if (StringUtils.isNotBlank(auxJars)) {
  237. try {
  238. loader = Utilities.addToClassPath(loader, StringUtils.split(auxJars, ","));
  239. } catch (Exception e) {
  240. LOG.error("Failed to add jars to class loader: " + auxJars, e);
  241. }
  242. }
  243. hiveConf.setClassLoader(loader);
  244. Thread.currentThread().setContextClassLoader(loader);
  245. SessionState.start(hiveConf); // this is thread-local
  246. this.sessionState = SessionState.get();
  247. // If this work has a LogContext, associate the children output to the logContext
  248. OutputStream lcOutStream = null;
  249. if (this.logContext != null)
  250. lcOutStream = this.logContext.getOutputStream();
  251. // A copy of everything goes to the LogContext.
  252. // In addition, stderr goes to errStream for error reporting.
  253. // Note that child output is explicitly tee to System.{out,err},
  254. // otherwise it'll be swallowed by outStream.
  255. this.sessionState.out = new PrintStream(new TeeOutputStream(lcOutStream, this.outStream));
  256. this.sessionState.err = new PrintStream(new TeeOutputStream(lcOutStream, this.errStream));
  257. this.sessionState.childOut =
  258. new PrintStream(new TeeOutputStream(System.out, sessionState.out));
  259. this.sessionState.childErr =
  260. new PrintStream(new TeeOutputStream(System.err, sessionState.err));
  261. this.state = QueryState.INITIALIZED;
  262. }
  263. /**
  264. * Executes query. Updates state. (QueryState variable can be polled.)
  265. * @throws BeeswaxException
  266. */
  267. public void execute() throws BeeswaxException {
  268. synchronized (this) {
  269. assertState(QueryState.COMPILED);
  270. state = QueryState.RUNNING;
  271. }
  272. int ret = driver.execute();
  273. try {
  274. synchronized (this) {
  275. assertState(QueryState.RUNNING);
  276. if (ret == 0) {
  277. state = QueryState.FINISHED;
  278. } else {
  279. throwException(new BeeswaxException("Driver returned: " + ret
  280. + ". Errors: " + getErrorStreamAsString(), logContext.getName(),
  281. this.handle));
  282. }
  283. }
  284. } finally {
  285. notifyDone(this);
  286. }
  287. }
  288. public void bringUp() {
  289. SessionState.start(this.sessionState);
  290. }
  291. private void materializeResults(Results r, boolean startOver) throws IOException {
  292. if (driver.getPlan().getPlan().getFetchTask() == null) {
  293. // This query is never going to return anything.
  294. r.has_more = false;
  295. r.setData(Collections.<String>emptyList());
  296. r.setColumns(Collections.<String>emptyList());
  297. return;
  298. }
  299. if (startOver) {
  300. // This is totally inappropriately reaching into internals.
  301. driver.getPlan().getPlan().getFetchTask().initialize(hiveConf,
  302. driver.getPlan());
  303. startRow = 0;
  304. }
  305. Vector<String> v = new Vector<String>();
  306. r.setData(v);
  307. r.has_more = driver.getResults(v);
  308. r.start_row = startRow;
  309. startRow += v.size();
  310. r.setColumns(new ArrayList<String>());
  311. try {
  312. for (FieldSchema f : driver.getSchema().getFieldSchemas()) {
  313. r.addToColumns(f.getName());
  314. }
  315. } catch (Exception e) {
  316. // An empty partitioned table may not have table description
  317. LOG.error("Error getting column names of results.", e);
  318. }
  319. }
  320. /**
  321. * Get the result schema and misc metadata, in the context of SELECT.
  322. */
  323. synchronized public ResultsMetadata getResultMetadata() {
  324. Schema schema = null;
  325. try {
  326. schema = driver.getSchema();
  327. } catch (Exception ex) {
  328. LOG.error("Error getting schema for query: " + query.query, ex);
  329. }
  330. fetchWork work = getFetchWork();
  331. tableDesc desc = work.getTblDesc();
  332. String tabledir = null;
  333. String tablename = null;
  334. String sep = null;
  335. if (work != null) {
  336. tabledir = work.getTblDir();
  337. }
  338. if (desc != null) {
  339. sep = desc.getProperties().getProperty(
  340. Constants.SERIALIZATION_FORMAT,
  341. "" + Utilities.ctrlaCode);
  342. tablename = desc.getTableName();
  343. }
  344. return new ResultsMetadata(schema, tabledir, tablename, sep);
  345. }
  346. /**
  347. * Get the fetchWork. Only SELECTs have them.
  348. */
  349. synchronized private fetchWork getFetchWork() {
  350. QueryPlan plan = driver.getPlan();
  351. FetchTask fetchTask = null;
  352. if (plan != null) {
  353. BaseSemanticAnalyzer sem = plan.getPlan();
  354. if (sem.getFetchTask() != null) {
  355. if (!sem.getFetchTaskInit()) {
  356. sem.setFetchTaskInit(true);
  357. sem.getFetchTask().initialize(hiveConf, plan);
  358. }
  359. fetchTask = (FetchTask) sem.getFetchTask();
  360. }
  361. }
  362. if (fetchTask == null) {
  363. return null;
  364. }
  365. fetchWork work = (fetchWork) fetchTask.getWork();
  366. return work;
  367. }
  368. synchronized public QueryExplanation explain() throws BeeswaxException {
  369. assertState(QueryState.INITIALIZED);
  370. // By manipulating the query, this will make errors harder to find.
  371. query.query = "EXPLAIN " + query.query;
  372. checkedCompile();
  373. if (!(driver.getPlan().getPlan() instanceof ExplainSemanticAnalyzer)) {
  374. throwException(new RuntimeException("Expected explain plan."));
  375. }
  376. int ret;
  377. if (0 != (ret = driver.execute())) {
  378. throwException(new RuntimeException("Failed to execute: EXPLAIN " + ret));
  379. }
  380. StringBuilder sb = new StringBuilder();
  381. Vector<String> v = new Vector<String>();
  382. try {
  383. while (driver.getResults(v)) {
  384. for (String s : v) {
  385. sb.append(s);
  386. sb.append("\n");
  387. }
  388. v.clear();
  389. }
  390. } catch (IOException e) {
  391. throwException(new RuntimeException(e));
  392. } finally {
  393. // Don't let folks re-use the state object.
  394. state = QueryState.FINISHED;
  395. }
  396. return new QueryExplanation(sb.toString());
  397. }
  398. public Results fetch(boolean fromBeginning) throws BeeswaxException {
  399. this.atime = System.currentTimeMillis();
  400. Results r = new Results();
  401. // Only one person can access a running query at a time.
  402. synchronized(this) {
  403. switch(state) {
  404. case RUNNING:
  405. r.ready = false;
  406. break;
  407. case FINISHED:
  408. bringUp();
  409. r.ready = true;
  410. try {
  411. materializeResults(r, fromBeginning);
  412. } catch (IOException e) {
  413. throw new BeeswaxException(e.toString(), logContext.getName(), handle);
  414. }
  415. break;
  416. case EXCEPTION:
  417. if (exception instanceof BeeswaxException) {
  418. throw (BeeswaxException) exception;
  419. } else {
  420. throw new BeeswaxException(exception.toString(), logContext.getName(), handle);
  421. }
  422. }
  423. }
  424. return r;
  425. }
  426. /** Store the first exception we see. */
  427. private void saveException(Throwable t) {
  428. synchronized (this) {
  429. if (state != QueryState.EXCEPTION) {
  430. state = QueryState.EXCEPTION;
  431. exception = t;
  432. }
  433. }
  434. }
  435. /**
  436. * Submits this query to the given executor to be run.
  437. *
  438. * @param executor
  439. * @param lc
  440. */
  441. void submitTo(ExecutorService executor, final LogContext lc) {
  442. final RunningQueryState state = this;
  443. executor.submit(new Runnable() {
  444. @Override
  445. public void run() {
  446. try {
  447. lc.registerCurrentThread();
  448. try {
  449. Hive.closeCurrent();
  450. Hive.get(state.hiveConf);
  451. } catch (HiveException ex) {
  452. throw new RuntimeException(ex);
  453. }
  454. state.bringUp();
  455. state.execute();
  456. } catch (Throwable t) {
  457. LOG.error("Exception while processing query", t);
  458. state.saveException(t);
  459. }
  460. }
  461. });
  462. }
  463. }
  464. /**
  465. * Notify Desktop that this query has finished, by sending a GET request
  466. * to a specific URL. We expect Desktop that view to always return HTTP_OK,
  467. * so that we know the notification has been delivered to the right view.
  468. * (We don't care whether the notification handling fails or succeeds.)
  469. */
  470. void notifyDone(RunningQueryState state) {
  471. QueryHandle handle = state.getQueryHandle();
  472. if (handle == null) {
  473. LOG.error("Finished execution of a query without a handle: " + state.toString());
  474. return;
  475. }
  476. String urlString = notifyUrl + handle.id;
  477. try {
  478. URL url = new URL(urlString);
  479. HttpURLConnection conn = (HttpURLConnection) url.openConnection();
  480. conn.setRequestMethod("GET");
  481. conn.connect();
  482. if (conn.getResponseCode() != HttpURLConnection.HTTP_OK) {
  483. throw new IOException("Desktop returns error: " + conn.getResponseMessage());
  484. }
  485. LOG.debug("Notified query done at " + url);
  486. } catch (IOException ioe) {
  487. LOG.error("Error when notifying Desktop at " + urlString, ioe);
  488. }
  489. }
  490. /**
  491. * Create a new BeeswaxServiceImpl.
  492. * @param dtHost The Hue host (ip or hostname).
  493. * @param dtPort The port Desktop runs on.
  494. * @param dtHttps Whether Desktop is running https.
  495. */
  496. public BeeswaxServiceImpl(String dtHost, int dtPort, boolean dtHttps) {
  497. LogContext.initLogCapture();
  498. this.executor = Executors.newCachedThreadPool(new NamingThreadFactory("Beeswax-%d"));
  499. this.runningQueries = new ConcurrentHashMap<String, RunningQueryState>();
  500. String protocol;
  501. if (dtHttps) {
  502. protocol = "https";
  503. try {
  504. // Disable SSL verification. HUE cert may be signed by untrusted CA.
  505. SSLContext sslcontext = SSLContext.getInstance("SSL");
  506. sslcontext.init(null,
  507. new DummyX509TrustManager[] { new DummyX509TrustManager() },
  508. new SecureRandom());
  509. HttpsURLConnection.setDefaultSSLSocketFactory(sslcontext.getSocketFactory());
  510. } catch (NoSuchAlgorithmException ex) {
  511. LOG.warn("Failed to disable SSL certificate check " + ex);
  512. } catch (KeyManagementException ex) {
  513. LOG.warn("Failed to disable SSL certificate check " + ex);
  514. }
  515. DummyHostnameVerifier dummy = new DummyHostnameVerifier();
  516. HttpsURLConnection.setDefaultHostnameVerifier(dummy);
  517. } else {
  518. protocol = "http";
  519. }
  520. this.notifyUrl = protocol + "://" + dtHost + ":" + dtPort + NOTIFY_URL_BASE;
  521. // A daemon thread that periodically evict stale RunningQueryState objects
  522. Thread evicter = new Thread(new Runnable() {
  523. @Override
  524. public void run() {
  525. while (true) {
  526. long now = System.currentTimeMillis();
  527. for (Map.Entry<String, RunningQueryState> entry : runningQueries.entrySet()) {
  528. RunningQueryState rqState = entry.getValue();
  529. if (rqState.getAtime() + RUNNING_QUERY_LIFETIME < now) {
  530. String id = entry.getKey();
  531. runningQueries.remove(id);
  532. LOG.debug("Removed " + rqState.toString());
  533. Thread.yield(); // be nice
  534. }
  535. }
  536. LogContext.garbageCollect(RUNNING_QUERY_LIFETIME);
  537. long wakeup = now + EVICTION_INTERVAL;
  538. while (System.currentTimeMillis() < wakeup) {
  539. try {
  540. Thread.sleep(EVICTION_INTERVAL);
  541. } catch (InterruptedException e) { }
  542. }
  543. }
  544. }
  545. }, "Evicter");
  546. evicter.setDaemon(true);
  547. evicter.start();
  548. }
  549. private <T> T doWithState(RunningQueryState state, PrivilegedExceptionAction<T> action)
  550. throws BeeswaxException
  551. {
  552. try{
  553. UserGroupInformation ugi;
  554. if (UserGroupInformation.isSecurityEnabled())
  555. ugi = UserGroupInformation.createProxyUser(state.query.hadoop_user, UserGroupInformation.getLoginUser());
  556. else {
  557. ugi = UserGroupInformation.createRemoteUser(state.query.hadoop_user);
  558. }
  559. return ugi.doAs(action);
  560. } catch (UndeclaredThrowableException e) {
  561. if (e.getUndeclaredThrowable() instanceof PrivilegedActionException) {
  562. Throwable bwe = e.getUndeclaredThrowable().getCause();
  563. if (bwe instanceof BeeswaxException) {
  564. LOG.error("Caught BeeswaxException", (BeeswaxException) bwe);
  565. throw (BeeswaxException) bwe;
  566. }
  567. }
  568. LOG.error("Caught unexpected exception.", e);
  569. throw new BeeswaxException(e.getMessage(), state.handle.log_context, state.handle);
  570. } catch (IOException e) {
  571. LOG.error("Caught IOException", e);
  572. throw new BeeswaxException(e.getMessage(), state.handle.log_context, state.handle);
  573. } catch (InterruptedException e) {
  574. LOG.error("Caught InterruptedException", e);
  575. throw new BeeswaxException(e.getMessage(), state.handle.log_context, state.handle);
  576. }
  577. }
  578. /**
  579. * Submit a query and return a handle (QueryHandle). The query runs asynchronously.
  580. * Queries can be long-lasting, so we push the execution into a new state.
  581. * Compiling happens in the current context so we report errors early.
  582. */
  583. @Override
  584. public QueryHandle query(final Query query) throws BeeswaxException {
  585. // First, create an id and reset the LogContext
  586. String uuid = UUID.randomUUID().toString();
  587. final QueryHandle handle = new QueryHandle(uuid, uuid);
  588. final LogContext lc = LogContext.registerCurrentThread(handle.log_context);
  589. lc.resetLog();
  590. // Make an administrative record
  591. final RunningQueryState state = new RunningQueryState(query, lc);
  592. try {
  593. return doWithState(state,
  594. new PrivilegedExceptionAction<QueryHandle>() {
  595. public QueryHandle run() throws Exception {
  596. state.setQueryHandle(handle);
  597. runningQueries.put(handle.id, state);
  598. state.initialize();
  599. // All kinds of things can go wrong when we compile it. So catch all.
  600. try {
  601. state.compile();
  602. } catch (BeeswaxException perr) {
  603. state.saveException(perr);
  604. throw perr;
  605. } catch (Throwable t) {
  606. state.saveException(t);
  607. throw new BeeswaxException(t.toString(), handle.log_context, handle);
  608. }
  609. // Now spin off the query.
  610. state.submitTo(executor, lc);
  611. return handle;
  612. }
  613. });
  614. } catch (BeeswaxException e) {
  615. throw e;
  616. }
  617. }
  618. /**
  619. * Verify that the handle data is not null.
  620. */
  621. private void validateHandle(QueryHandle handle) throws QueryNotFoundException {
  622. if (handle == null) {
  623. LOG.error("Encountered null QueryHandle");
  624. throw new QueryNotFoundException();
  625. }
  626. if (handle.id == null || handle.log_context == null) {
  627. LOG.error("Invalid QueryHandle: id " + handle.id + "; log context " + handle.log_context);
  628. throw new QueryNotFoundException();
  629. }
  630. }
  631. @Override
  632. public String echo(String s) throws TException {
  633. return s;
  634. }
  635. /**
  636. * Get the query plan for a query.
  637. */
  638. @Override
  639. public QueryExplanation explain(final Query query) throws BeeswaxException, TException {
  640. final String contextName = UUID.randomUUID().toString();
  641. LogContext lc = LogContext.registerCurrentThread(contextName);
  642. final RunningQueryState state = new RunningQueryState(query, lc);
  643. try {
  644. return doWithState(state,
  645. new PrivilegedExceptionAction<QueryExplanation>() {
  646. public QueryExplanation run() throws Exception {
  647. state.initialize();
  648. QueryExplanation exp;
  649. // All kinds of things can go wrong when we compile it. So catch all.
  650. try {
  651. exp = state.explain();
  652. } catch (BeeswaxException perr) {
  653. throw perr;
  654. } catch (Throwable t) {
  655. throw new BeeswaxException(t.toString(), contextName, null);
  656. }
  657. // On success, we remove the LogContext
  658. LogContext.destroyContext(contextName);
  659. return exp;
  660. }
  661. });
  662. } catch (BeeswaxException e) {
  663. throw e;
  664. }
  665. }
  666. /**
  667. * Get the results of a query. This is non-blocking. Caller should check
  668. * Results.ready to determine if the results are in yet.
  669. *
  670. * @param handle The handle from query()
  671. * @param fromBeginning If true, rewind to the first row. Otherwise fetch from last position.
  672. */
  673. @Override
  674. public Results fetch(final QueryHandle handle, final boolean fromBeginning)
  675. throws QueryNotFoundException, BeeswaxException {
  676. LogContext.unregisterCurrentThread();
  677. validateHandle(handle);
  678. LogContext.registerCurrentThread(handle.log_context);
  679. final RunningQueryState state = runningQueries.get(handle.id);
  680. try {
  681. return doWithState(state,
  682. new PrivilegedExceptionAction<Results>() {
  683. public Results run() throws Exception {
  684. if (state == null) {
  685. throw new QueryNotFoundException();
  686. }
  687. return state.fetch(fromBeginning);
  688. }
  689. });
  690. } catch (BeeswaxException e) {
  691. throw e;
  692. }
  693. }
  694. @Override
  695. public String dump_config() throws TException {
  696. HiveConf c = new HiveConf();
  697. ByteArrayOutputStream b = new ByteArrayOutputStream();
  698. try {
  699. c.writeXml(b);
  700. return new String(b.toByteArray(), "UTF-8");
  701. } catch (IOException e) {
  702. throw new TException(e);
  703. }
  704. }
  705. /**
  706. * Get the state of the query
  707. *
  708. * @param handle The handle from query()
  709. */
  710. @Override
  711. public QueryState get_state(final QueryHandle handle) throws QueryNotFoundException {
  712. LogContext.unregisterCurrentThread();
  713. validateHandle(handle);
  714. LogContext.registerCurrentThread(handle.log_context);
  715. RunningQueryState state = runningQueries.get(handle.id);
  716. if (state == null) {
  717. throw new QueryNotFoundException();
  718. }
  719. return state.state;
  720. }
  721. /**
  722. * Get the results metadata
  723. *
  724. * @param handle
  725. */
  726. @Override
  727. public ResultsMetadata get_results_metadata(final QueryHandle handle) throws QueryNotFoundException {
  728. LogContext.unregisterCurrentThread();
  729. validateHandle(handle);
  730. LogContext.registerCurrentThread(handle.log_context);
  731. final RunningQueryState state = runningQueries.get(handle.id);
  732. try {
  733. return doWithState(state,
  734. new PrivilegedExceptionAction<ResultsMetadata>() {
  735. public ResultsMetadata run() throws Exception {
  736. if (state == null) {
  737. throw new QueryNotFoundException();
  738. }
  739. return state.getResultMetadata();
  740. }
  741. });
  742. } catch (BeeswaxException e) {
  743. LOG.error("Caught BeeswaxException.", e);
  744. throw new QueryNotFoundException();
  745. }
  746. }
  747. /**
  748. * Get the log messages related to the context.
  749. *
  750. * @param contextName The log context name
  751. * @return The log message as a string.
  752. */
  753. @Override
  754. public String get_log(String contextName) throws QueryNotFoundException, TException {
  755. final boolean DONT_CREATE = false;
  756. LogContext.unregisterCurrentThread();
  757. if (contextName == null) {
  758. throw new QueryNotFoundException();
  759. }
  760. LogContext lc = LogContext.getByName(contextName, DONT_CREATE);
  761. if (lc == null)
  762. throw new QueryNotFoundException();
  763. return lc.readLog();
  764. }
  765. /*
  766. * This is similar in spirit to Hive's own SetProcessor
  767. */
  768. @Override
  769. public List<ConfigVariable> get_default_configuration(boolean includeHadoop)
  770. throws TException {
  771. HiveConf conf = new HiveConf(BeeswaxServiceImpl.class);
  772. Properties p;
  773. if (includeHadoop) {
  774. p = conf.getAllProperties();
  775. } else {
  776. p = conf.getChangedProperties();
  777. }
  778. List<ConfigVariable> ret = new ArrayList<ConfigVariable>();
  779. for (Entry<Object, Object> e : p.entrySet()) {
  780. String key = (String)e.getKey();
  781. String value = (String)e.getValue();
  782. ConfigVariable cv = new ConfigVariable();
  783. cv.setKey(key);
  784. cv.setValue(value);
  785. cv.setDescription(configDescriptions.lookup(key));
  786. ret.add(cv);
  787. }
  788. return ret;
  789. }
  790. }