dbms.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import logging
  18. import thrift
  19. import time
  20. from django.core.urlresolvers import reverse
  21. from django.shortcuts import redirect
  22. from django.utils.encoding import force_unicode
  23. from django.utils.translation import ugettext as _
  24. from beeswaxd.ttypes import BeeswaxException
  25. from desktop.conf import KERBEROS
  26. from filebrowser.views import location_to_url
  27. from beeswax import hive_site
  28. from beeswax.conf import BEESWAX_SERVER_HOST, BEESWAX_SERVER_PORT,\
  29. BROWSE_PARTITIONED_TABLE_LIMIT, SERVER_INTERFACE
  30. from beeswax.design import hql_query
  31. from beeswax.models import QueryHistory, HIVE_SERVER2, BEESWAX
  32. from desktop.lib.django_util import format_preserving_redirect
  33. from desktop.lib.exceptions_renderable import PopupException
  34. LOG = logging.getLogger(__name__)
  35. def get(user, query_server=None):
  36. # Avoid circular dependency
  37. from beeswax.server.hive_server2_lib import HiveServerClientCompatible, HiveServerClient
  38. from beeswax.server.beeswax_lib import BeeswaxClient
  39. if query_server is None:
  40. query_server = get_query_server_config()
  41. if query_server['server_interface'] == HIVE_SERVER2:
  42. return Dbms(HiveServerClientCompatible(HiveServerClient(query_server, user)), QueryHistory.SERVER_TYPE[1][0])
  43. else:
  44. return Dbms(BeeswaxClient(query_server, user), QueryHistory.SERVER_TYPE[0][0])
  45. def get_query_server_config(name='beeswax'):
  46. if name == 'impala':
  47. from impala.conf import SERVER_HOST, SERVER_PORT, IMPALA_PRINCIPAL, SERVER_INTERFACE as IMPALA_SERVER_INTERFACE
  48. # Backward compatibility until Hue 3.0
  49. # If no interface specified and port is beeswax, switch port to HS2 default as we want to use HS2 from now on
  50. if IMPALA_SERVER_INTERFACE.get() == 'hiveserver2' and SERVER_PORT.get() == 21000:
  51. port = 21050
  52. else:
  53. port = SERVER_PORT.get()
  54. query_server = {
  55. 'server_name': 'impala',
  56. 'server_host': SERVER_HOST.get(),
  57. 'server_port': port,
  58. 'server_interface': IMPALA_SERVER_INTERFACE.get(),
  59. 'principal': IMPALA_PRINCIPAL.get(),
  60. }
  61. else:
  62. if SERVER_INTERFACE.get() == 'hiveserver2':
  63. kerberos_principal = hive_site.get_hiveserver2_kerberos_principal(BEESWAX_SERVER_HOST.get())
  64. else:
  65. # Beeswaxd runs as 'hue'
  66. kerberos_principal = KERBEROS.HUE_PRINCIPAL.get()
  67. query_server = {
  68. 'server_name': 'beeswax', # Aka HS2 too
  69. 'server_host': BEESWAX_SERVER_HOST.get(),
  70. 'server_port': BEESWAX_SERVER_PORT.get(),
  71. 'server_interface': SERVER_INTERFACE.get(),
  72. 'principal': kerberos_principal
  73. }
  74. LOG.debug("Query Server: %s" % query_server)
  75. return query_server
  76. class QueryServerException(Exception):
  77. # Ideally the query handle will be stored here too.
  78. def __init__(self, e):
  79. super(QueryServerException, self).__init__(e)
  80. class NoSuchObjectException: pass
  81. class Dbms:
  82. """SQL"""
  83. def __init__(self, client, server_type):
  84. self.client = client
  85. self.server_type = server_type
  86. def get_table(self, database, table_name):
  87. # DB name not supported in SHOW PARTITIONS required in Table
  88. self.use(database)
  89. return self.client.get_table(database, table_name)
  90. def get_tables(self, database='default', table_names='.*'):
  91. return self.client.get_tables(database, table_names)
  92. def get_databases(self):
  93. return self.client.get_databases()
  94. def execute_query(self, query, design):
  95. return self.execute_and_watch(query, design=design)
  96. def select_star_from(self, database, table):
  97. hql = "SELECT * FROM `%s.%s` %s" % (database, table.name, self._get_browse_limit_clause(table))
  98. return self.execute_statement(hql)
  99. def execute_statement(self, hql):
  100. query = hql_query(hql)
  101. return self.execute_and_watch(query)
  102. def fetch(self, query_handle, start_over=False, rows=None):
  103. no_start_over_support = [config_variable for config_variable in self.get_default_configuration(False)
  104. if config_variable.key == 'support_start_over'
  105. and config_variable.value == 'false']
  106. if no_start_over_support:
  107. start_over = False
  108. return self.client.fetch(query_handle, start_over, rows)
  109. def close_operation(self, query_handle):
  110. if self.server_type == BEESWAX:
  111. raise PopupException(_('%s interface does not support close_operation. %s interface does.') % (BEESWAX, HIVE_SERVER2))
  112. return self.client.close_operation(query_handle)
  113. def cancel_operation(self, query_handle):
  114. resp = self.client.cancel_operation(query_handle)
  115. if self.client.query_server['server_name'] == 'impala':
  116. resp = self.client.close_operation(query_handle)
  117. return resp
  118. def get_sample(self, database, table):
  119. """No samples if it's a view (HUE-526)"""
  120. if not table.is_view:
  121. limit = min(100, BROWSE_PARTITIONED_TABLE_LIMIT.get())
  122. hql = "SELECT * FROM `%s.%s` LIMIT %s" % (database, table.name, limit)
  123. query = hql_query(hql)
  124. handle = self.execute_and_wait(query, timeout_sec=5.0)
  125. if handle:
  126. return self.fetch(handle)
  127. def drop_table(self, database, table):
  128. if table.is_view:
  129. hql = "DROP VIEW `%s.%s`" % (database, table.name,)
  130. else:
  131. hql = "DROP TABLE `%s.%s`" % (database, table.name,)
  132. return self.execute_statement(hql)
  133. def load_data(self, database, table, form, design):
  134. hql = "LOAD DATA INPATH"
  135. hql += " '%s'" % form.cleaned_data['path']
  136. if form.cleaned_data['overwrite']:
  137. hql += " OVERWRITE"
  138. hql += " INTO TABLE "
  139. hql += "`%s.%s`" % (database, table.name,)
  140. if form.partition_columns:
  141. hql += " PARTITION ("
  142. vals = []
  143. for key, column_name in form.partition_columns.iteritems():
  144. vals.append("%s='%s'" % (column_name, form.cleaned_data[key]))
  145. hql += ", ".join(vals)
  146. hql += ")"
  147. query = hql_query(hql, database)
  148. design.data = query.dumps()
  149. design.save()
  150. return self.execute_query(query, design)
  151. def drop_tables(self, database, tables, design):
  152. hql = []
  153. for table in tables:
  154. if table.is_view:
  155. hql.append("DROP VIEW `%s.%s`" % (database, table.name,))
  156. else:
  157. hql.append("DROP TABLE `%s.%s`" % (database, table.name,))
  158. query = hql_query(';'.join(hql), database)
  159. design.data = query.dumps()
  160. design.save()
  161. return self.execute_query(query, design)
  162. def drop_database(self, database):
  163. return self.execute_statement("DROP DATABASE `%s`" % database)
  164. def drop_databases(self, databases, design):
  165. hql = []
  166. for database in databases:
  167. hql.append("DROP DATABASE `%s`" % database)
  168. query = hql_query(';'.join(hql), database)
  169. design.data = query.dumps()
  170. design.save()
  171. return self.execute_query(query, design)
  172. def insert_query_into_directory(self, query_history, target_dir):
  173. design = query_history.design.get_design()
  174. hql = "INSERT OVERWRITE DIRECTORY '%s' %s" % (target_dir, design.query['query'])
  175. return self.execute_statement(hql)
  176. def create_table_as_a_select(self, request, query_history, target_table, result_meta):
  177. design = query_history.design.get_design()
  178. database = design.query['database']
  179. # Case 1: Hive Server 2 backend or results straight from an existing table
  180. if result_meta.in_tablename:
  181. hql = 'CREATE TABLE `%s.%s` AS %s' % (database, target_table, design.query['query'])
  182. #query = hql_query(hql, database=database)
  183. query_history = self.execute_statement(hql)
  184. url = redirect(reverse('beeswax:watch_query', args=[query_history.id]) + '?on_success_url=' + reverse('metastore:describe_table', args=[database, target_table]))
  185. else:
  186. # Case 2: The results are in some temporary location
  187. # Beeswax backward compatibility and optimization
  188. # 1. Create table
  189. cols = ''
  190. schema = result_meta.schema
  191. for i, field in enumerate(schema.fieldSchemas):
  192. if i != 0:
  193. cols += ',\n'
  194. cols += '`%s` %s' % (field.name, field.type)
  195. # The representation of the delimiter is messy.
  196. # It came from Java as a string, which might has been converted from an integer.
  197. # So it could be "1" (^A), or "10" (\n), or "," (a comma literally).
  198. delim = result_meta.delim
  199. if not delim.isdigit():
  200. delim = str(ord(delim))
  201. hql = '''
  202. CREATE TABLE `%s` (
  203. %s
  204. )
  205. ROW FORMAT DELIMITED
  206. FIELDS TERMINATED BY '\%s'
  207. STORED AS TextFile
  208. ''' % (target_table, cols, delim.zfill(3))
  209. query = hql_query(hql)
  210. self.execute_and_wait(query)
  211. try:
  212. # 2. Move the results into the table's storage
  213. table_obj = self.get_table('default', target_table)
  214. table_loc = request.fs.urlsplit(table_obj.path_location)[2]
  215. result_dir = request.fs.urlsplit(result_meta.table_dir)[2]
  216. request.fs.rename_star(result_dir, table_loc)
  217. LOG.debug("Moved results from %s to %s" % (result_meta.table_dir, table_loc))
  218. request.info(request, _('Saved query results as new table %(table)s.') % {'table': target_table})
  219. query_history.save_state(QueryHistory.STATE.expired)
  220. except Exception, ex:
  221. query = hql_query('DROP TABLE `%s`' % target_table)
  222. try:
  223. self.execute_and_wait(query)
  224. except Exception, double_trouble:
  225. LOG.exception('Failed to drop table "%s" as well: %s' % (target_table, double_trouble))
  226. raise ex
  227. url = format_preserving_redirect(request, reverse('metastore:index'))
  228. return url
  229. def use(self, database):
  230. """Beeswax interface does not support use directly."""
  231. if self.server_type == HIVE_SERVER2:
  232. query = hql_query('USE %s' % database)
  233. self.client.query(query)
  234. # TODO sync + close query
  235. def get_log(self, query_handle):
  236. return self.client.get_log(query_handle)
  237. def get_state(self, handle):
  238. return self.client.get_state(handle)
  239. def execute_and_wait(self, query, timeout_sec=30.0):
  240. """
  241. Run query and check status until it finishes or timeouts.
  242. """
  243. SLEEP_INTERVAL = 0.5
  244. handle = self.client.query(query)
  245. curr = time.time()
  246. end = curr + timeout_sec
  247. while curr <= end:
  248. state = self.client.get_state(handle)
  249. if state not in (QueryHistory.STATE.running, QueryHistory.STATE.submitted):
  250. return handle
  251. time.sleep(SLEEP_INTERVAL)
  252. curr = time.time()
  253. return None
  254. def execute_next_statement(self, query_history):
  255. query_history.statement_number += 1
  256. query_history.last_state = QueryHistory.STATE.submitted.index
  257. query_history.save()
  258. query = query_history.design.get_design()
  259. return self.execute_and_watch(query, query_history=query_history)
  260. def execute_and_watch(self, query, design=None, query_history=None):
  261. """
  262. Run query and return a QueryHistory object in order to see its progress on a Web page.
  263. """
  264. hql_query = query.hql_query
  265. if query_history is None:
  266. query_history = QueryHistory.build(
  267. owner=self.client.user,
  268. query=hql_query,
  269. server_host='%(server_host)s' % self.client.query_server,
  270. server_port='%(server_port)d' % self.client.query_server,
  271. server_name='%(server_name)s' % self.client.query_server,
  272. server_type=self.server_type,
  273. last_state=QueryHistory.STATE.submitted.index,
  274. design=design,
  275. notify=query.query.get('email_notify', False),
  276. query_type=query.query['type'],
  277. statement_number=0
  278. )
  279. query_history.save()
  280. LOG.debug("Made new QueryHistory id %s user %s query: %s..." % (query_history.id, self.client.user, query_history.query[:25]))
  281. try:
  282. handle = self.client.query(query, query_history.statement_number)
  283. if not handle.is_valid():
  284. msg = _("Server returning invalid handle for query id %(id)d [%(query)s]...") % \
  285. {'id': query_history.id, 'query': query[:40]}
  286. raise BeeswaxException(msg)
  287. except BeeswaxException, ex: # TODO HS2
  288. LOG.exception(ex)
  289. # Kind of expected (hql compile/syntax error, etc.)
  290. if hasattr(ex, 'handle') and ex.handle:
  291. query_history.server_id = ex.handle.id
  292. query_history.log_context = ex.handle.log_context
  293. query_history.save_state(QueryHistory.STATE.failed)
  294. raise ex
  295. # All good
  296. query_history.server_id, query_history.server_guid = handle.get()
  297. query_history.operation_type = handle.operation_type
  298. query_history.has_results = handle.has_result_set
  299. query_history.modified_row_count = handle.modified_row_count
  300. query_history.log_context = handle.log_context
  301. query_history.query_type = query.query['type']
  302. query_history.set_to_running()
  303. query_history.save()
  304. LOG.debug("Updated QueryHistory id %s user %s statement_number: %s" % (query_history.id, self.client.user, query_history.statement_number))
  305. return query_history
  306. def get_results_metadata(self, handle):
  307. return self.client.get_results_metadata(handle)
  308. def close(self, handle):
  309. return self.client.close(handle)
  310. def get_partitions(self, db_name, table, max_parts=None):
  311. if max_parts is None or max_parts > BROWSE_PARTITIONED_TABLE_LIMIT.get():
  312. max_parts = BROWSE_PARTITIONED_TABLE_LIMIT.get()
  313. # DB name not supported in SHOW PARTITIONS
  314. self.use(db_name)
  315. return self.client.get_partitions(db_name, table.name, max_parts)
  316. def get_partition(self, db_name, table_name, partition_id):
  317. table = self.get_table(db_name, table_name)
  318. partitions = self.get_partitions(db_name, table, max_parts=None)
  319. partition_query = ""
  320. for idx, key in enumerate(partitions[partition_id].values):
  321. partition_query += (idx > 0 and " AND " or "") + table.partition_keys[idx].name + "='%s'" % key
  322. hql = "SELECT * FROM `%s.%s` WHERE %s" % (db_name, table_name, partition_query)
  323. return self.execute_statement(hql)
  324. def explain(self, statement):
  325. return self.client.explain(statement)
  326. def echo(self, text):
  327. return self.client.echo(text)
  328. def getStatus(self):
  329. return self.client.getStatus()
  330. def get_default_configuration(self, include_hadoop):
  331. return self.client.get_default_configuration(include_hadoop)
  332. def _get_browse_limit_clause(self, table):
  333. """Get the limit clause when browsing a partitioned table"""
  334. if table.partition_keys:
  335. limit = BROWSE_PARTITIONED_TABLE_LIMIT.get()
  336. if limit > 0:
  337. return "LIMIT %d" % (limit,)
  338. return ""
  339. class Table:
  340. """
  341. Represents the metadata of a Hive Table.
  342. """
  343. @property
  344. def hdfs_link(self):
  345. return location_to_url(self.path_location)
  346. class DataTable:
  347. """
  348. Represents the data of a Hive Table.
  349. If the dataset has more rows, a new fetch should be done in order to return a new data table with the next rows.
  350. """
  351. pass
  352. # TODO decorator?
  353. def expand_exception(exc, db, handle=None):
  354. try:
  355. if handle is not None:
  356. log = db.get_log(handle)
  357. elif hasattr(exc, 'get_rpc_handle') or hasattr(exc, 'log_context'):
  358. log = db.get_log(exc)
  359. else:
  360. log = _("No server logs for this query.")
  361. except Exception, e:
  362. # Always show something, even if server has died on the job.
  363. log = _("Could not retrieve logs: %s." % e)
  364. if not exc.message:
  365. error_message = _("Unknown exception.")
  366. else:
  367. error_message = force_unicode(exc.message, strings_only=True, errors='replace')
  368. return error_message, log