Browse Source

HUE-1455 [impala] Close use and metastore queries

All operations are closed automatically except query() which is closed
by javascript when the user leaves the page.

We fetch 1000 rows for the other operations. We could have built
smart dataset that fetch/generate the data and then offer a close()
method but for now it should be fine.

Some operations that are watched (like drop tables, select from a partition,
create table as select, insert overwrite...) are closed like regular
query statements
Romain Rigaux 12 years ago
parent
commit
5689651cb3

+ 7 - 8
apps/beeswax/src/beeswax/server/dbms.py

@@ -162,7 +162,6 @@ class HS2Dbms(Dbms):
     if no_start_over_support:
       start_over = False
 
-
     return self.client.fetch(query_handle, start_over, rows)
 
 
@@ -191,7 +190,9 @@ class HS2Dbms(Dbms):
       handle = self.execute_and_wait(query, timeout_sec=5.0)
 
       if handle:
-        return self.fetch(handle)
+        result = self.fetch(handle)
+        self.close(handle)
+        return result
 
   def analyze_table_table(self, database, table):
     hql = 'analyze table `%(database)s.%(table_name)` compute statistics' % {'database': database, 'table_name': table.name}
@@ -334,8 +335,8 @@ class HS2Dbms(Dbms):
 
   def use(self, database):
     query = hql_query('USE %s' % database)
-    self.client.query(query)
-    # TODO sync + close query
+    return self.client.use(query)
+
 
   def get_log(self, query_handle):
     return self.client.get_log(query_handle)
@@ -345,14 +346,12 @@ class HS2Dbms(Dbms):
     return self.client.get_state(handle)
 
 
-  def execute_and_wait(self, query, timeout_sec=30.0):
+  def execute_and_wait(self, query, timeout_sec=30.0, sleep_interval=0.5):
     """
     Run query and check status until it finishes or timeouts.
 
     Check status until it finishes or timeouts.
     """
-    SLEEP_INTERVAL = 0.5
-
     handle = self.client.query(query)
     curr = time.time()
     end = curr + timeout_sec
@@ -360,7 +359,7 @@ class HS2Dbms(Dbms):
       state = self.client.get_state(handle)
       if state not in (QueryHistory.STATE.running, QueryHistory.STATE.submitted):
         return handle
-      time.sleep(SLEEP_INTERVAL)
+      time.sleep(sleep_interval)
       curr = time.time()
     return None
 

+ 38 - 14
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -168,9 +168,10 @@ class HiveServerTRowSet:
 
 
 class HiveServerDataTable(DataTable):
-  def __init__(self, results, schema):
+  def __init__(self, results, schema, operation_handle):
     self.schema = schema and schema.schema
     self.row_set = HiveServerTRowSet(results.results, schema)
+    self.operation_handle = operation_handle
     self.has_more = not self.row_set.is_empty()    # Should be results.hasMoreRows but always True in HS2
     self.startRowOffset = self.row_set.startRowOffset    # Always 0 in HS2
 
@@ -354,6 +355,13 @@ class HiveServerClient:
     req = TOpenSessionReq(**kwargs)
     res = self._client.OpenSession(req)
 
+    if res.status is not None and res.status.statusCode not in (TStatusCode.SUCCESS_STATUS,):
+      if hasattr(res.status, 'errorMessage') and res.status.errorMessage:
+        message = res.status.errorMessage
+      else:
+        message = ''
+      raise QueryServerException(Exception('Bad status for request %s:\n%s' % (req, res)), message=message)
+
     sessionId = res.sessionHandle.sessionId
     LOG.info('Opening session %s' % sessionId)
 
@@ -413,6 +421,7 @@ class HiveServerClient:
     res = self.call(self._client.GetSchemas, req)
 
     results, schema = self.fetch_result(res.operationHandle)
+    self.close_operation(res.operationHandle)
 
     col = 'TABLE_SCHEM'
     return HiveServerTRowSet(results.results, schema.schema).cols((col,))
@@ -423,6 +432,7 @@ class HiveServerClient:
     res = self.call(self._client.GetTables, req)
 
     results, schema = self.fetch_result(res.operationHandle)
+    self.close_operation(res.operationHandle)
 
     return HiveServerTRowSet(results.results, schema.schema).cols(('TABLE_NAME',))
 
@@ -432,24 +442,27 @@ class HiveServerClient:
     res = self.call(self._client.GetTables, req)
 
     table_results, table_schema = self.fetch_result(res.operationHandle)
+    self.close_operation(res.operationHandle)
+
     if self.query_server['server_name'] == 'impala':
       # Impala does not supported extended
       query = 'DESCRIBE %s' % table_name
     else:
       query = 'DESCRIBE EXTENDED %s' % table_name
-    desc_results, desc_schema = self.execute_statement(query)
+    (desc_results, desc_schema), operation_handle = self.execute_statement(query)
+    self.close_operation(operation_handle)
 
     return HiveServerTable(table_results.results, table_schema.schema, desc_results.results, desc_schema.schema)
 
 
-  def execute_query(self, query, max_rows=100):
+  def execute_query(self, query, max_rows=1000):
     configuration = self._get_query_configuration(query)
     return self.execute_query_statement(statement=query.query['query'], max_rows=max_rows, configuration=configuration)
 
 
-  def execute_query_statement(self, statement, max_rows=100, configuration={}):
-    results, schema = self.execute_statement(statement=statement, max_rows=max_rows, configuration=configuration)
-    return HiveServerDataTable(results, schema)
+  def execute_query_statement(self, statement, max_rows=1000, configuration={}):
+    (results, schema), operation_handle = self.execute_statement(statement=statement, max_rows=max_rows, configuration=configuration)
+    return HiveServerDataTable(results, schema, operation_handle)
 
 
   def execute_async_query(self, query, statement=0):
@@ -465,11 +478,11 @@ class HiveServerClient:
     return self.execute_async_statement(statement=query_statement, confOverlay=configuration)
 
 
-  def execute_statement(self, statement, max_rows=100, configuration={}):
+  def execute_statement(self, statement, max_rows=1000, configuration={}):
     req = TExecuteStatementReq(statement=statement.encode('utf-8'), confOverlay=configuration)
     res = self.call(self._client.ExecuteStatement, req)
 
-    return self.fetch_result(res.operationHandle, max_rows=max_rows)
+    return self.fetch_result(res.operationHandle, max_rows=max_rows), res.operationHandle
 
 
   def execute_async_statement(self, statement, confOverlay):
@@ -483,10 +496,10 @@ class HiveServerClient:
                                  modified_row_count=res.operationHandle.modifiedRowCount)
 
 
-  def fetch_data(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=100):
+  def fetch_data(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=1000):
     # The client should check for hasMoreRows and fetch until the result is empty dues to a HS2 bug
     results, schema = self.fetch_result(operation_handle, orientation, max_rows)
-    return HiveServerDataTable(results, schema)
+    return HiveServerDataTable(results, schema, operation_handle)
 
 
   def cancel_operation(self, operation_handle):
@@ -503,10 +516,13 @@ class HiveServerClient:
     req = TGetColumnsReq(schemaName=database, tableName=table)
     res = self.call(self._client.GetColumns, req)
 
-    return self.fetch_result(res.operationHandle)
+    res, schema = self.fetch_result(res.operationHandle)
+    self.close_operation(res.operationHandle)
+
+    return res, schema
 
 
-  def fetch_result(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=100):
+  def fetch_result(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=1000):
     fetch_req = TFetchResultsReq(operationHandle=operation_handle, orientation=orientation, maxRows=max_rows)
     res = self.call(self._client.FetchResults, fetch_req)
 
@@ -632,15 +648,23 @@ class HiveServerClientCompatible(object):
     return HiveServerQueryHistory.STATE_MAP[res.operationState]
 
 
+  def use(self, query):
+    data = self._client.execute_query(query)
+    self._client.close_operation(data.operation_handle)
+    return data
+
+
   def explain(self, query):
     data_table = self._client.explain(query)
-    return ExplainCompatible(data_table)
+    data = ExplainCompatible(data_table)
+    self._client.close_operation(data_table.operation_handle)
+    return data
 
 
   def fetch(self, handle, start_over=False, max_rows=None):
     operationHandle = handle.get_rpc_handle()
     if max_rows is None:
-      max_rows = 10000
+      max_rows = 1000
 
     # Impala does not support FETCH_FIRST
     if self.query_server['server_name'] == 'impala':

+ 5 - 2
apps/beeswax/src/beeswax/views.py

@@ -383,10 +383,13 @@ def execute_query(request, design_id=None):
   query_type = SavedQuery.TYPES_MAPPING[app_name]
   design = safe_get_design(request, query_type, design_id)
   on_success_url = request.REQUEST.get('on_success_url')
-
+  databases = []
   query_server = get_query_server_config(app_name)
   db = dbms.get(request.user, query_server)
-  databases = get_db_choices(request)
+  try:
+    databases = get_db_choices(request)
+  except Exception, ex:
+    error_message, log = expand_exception(ex, db)
 
   if request.method == 'POST':
     form.bind(request.POST)