Quellcode durchsuchen

HUE-7738 [hive] Support GetFunction execution and fetching

Note: seems to work only for Hive
Romain vor 5 Jahren
Ursprung
Commit
214b0efdc6

+ 11 - 5
apps/beeswax/src/beeswax/models.py

@@ -521,11 +521,17 @@ class HiveServerQueryHandle(QueryHandle):
   def get_rpc_handle(self):
     secret, guid = self.get_decoded(self.secret, self.guid)
 
-    operation = getattr(TOperationType, TOperationType._NAMES_TO_VALUES.get(self.operation_type, 'EXECUTE_STATEMENT'))
-    return TOperationHandle(operationId=THandleIdentifier(guid=guid, secret=secret),
-                            operationType=operation,
-                            hasResultSet=self.has_result_set,
-                            modifiedRowCount=self.modified_row_count)
+    operation = getattr(
+      TOperationType,
+      TOperationType._VALUES_TO_NAMES.get(self.operation_type, 'EXECUTE_STATEMENT')
+    )
+
+    return TOperationHandle(
+        operationId=THandleIdentifier(guid=guid, secret=secret),
+        operationType=operation,
+        hasResultSet=self.has_result_set,
+        modifiedRowCount=self.modified_row_count
+    )
 
   @classmethod
   def get_decoded(cls, secret, guid):

+ 13 - 9
apps/beeswax/src/beeswax/server/dbms.py

@@ -1099,20 +1099,24 @@ class HiveServer2Dbms(object):
 
 
   def get_functions(self, prefix=None, database=None):
+    '''
+    Not using self.client.get_functions() as pretty limited. More comments there.
+    '''
+    result = None
+
     if self.client.query_server['dialect'] == 'impala':
       if database is None:
         database = '_impala_builtins'
-      filter = '"%s.*"' % prefix if prefix else '".*"'
-      hql = 'SHOW FUNCTIONS %s' % filter
 
-      query = hql_query(hql)
-      handle = self.execute_and_wait(query, timeout_sec=15.0)
+    filter = '"%s.*"' % prefix if prefix else '".*"'
+    hql = 'SHOW FUNCTIONS %s' % filter
 
-      if handle:
-        result = self.fetch(handle, rows=5000).rows()
-        self.close(handle)
-    else:
-      result = self.client.get_functions('aa', 'bb')
+    query = hql_query(hql)
+    handle = self.execute_and_wait(query, timeout_sec=5.0)
+
+    if handle:
+      result = self.fetch(handle, rows=1000).rows()
+      self.close(handle)
 
     return result
 

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

@@ -970,7 +970,7 @@ class HiveServerClient(object):
     configuration.update(self._get_query_configuration(query))
     query_statement = query.get_query_statement(statement)
 
-    return self.execute_async_statement(statement=query_statement, confOverlay=configuration, session=session)
+    return self.execute_async_statement(statement=query_statement, conf_overlay=configuration, session=session)
 
 
   def execute_statement(self, statement, max_rows=1000, configuration=None, orientation=TFetchOrientation.FETCH_NEXT, session=None):
@@ -989,15 +989,21 @@ class HiveServerClient(object):
     return results, schema, res.operationHandle, session
 
 
-  def execute_async_statement(self, statement, confOverlay, session=None):
+  def execute_async_statement(self, statement=None, thrift_function=None, thrift_request=None, conf_overlay=None, session=None):
+    if conf_overlay is None:
+      conf_overlay = {}
+    if thrift_function is None:
+      thrift_function = self._client.ExecuteStatement
+    if thrift_request is None:
+      thrift_request = TExecuteStatementReq(statement=statement, confOverlay=conf_overlay, runAsync=True)
+
     if self.query_server.get('dialect') == 'impala' and self.query_server['QUERY_TIMEOUT_S'] > 0:
-      confOverlay['QUERY_TIMEOUT_S'] = str(self.query_server['QUERY_TIMEOUT_S'])
+      conf_overlay['QUERY_TIMEOUT_S'] = str(self.query_server['QUERY_TIMEOUT_S'])
 
     if sys.version_info[0] == 2:
       statement = statement.encode('utf-8')
 
-    req = TExecuteStatementReq(statement=statement, confOverlay=confOverlay, runAsync=True)
-    (res, session) = self.call_return_result_and_session(self._client.ExecuteStatement, req, session=session)
+    (res, session) = self.call_return_result_and_session(thrift_function, thrift_request, session=session)
 
     return HiveServerQueryHandle(
         secret=res.operationHandle.operationId.secret,
@@ -1181,14 +1187,20 @@ class HiveServerClient(object):
     return dict([(setting['key'], setting['value']) for setting in query.settings])
 
 
-  def get_functions(self, database, table):
+  def get_functions(self):
+    '''
+    Could support parameters.
+
+    Result data is pretty limited, e.g.
+    [None, None, 'histogram_numeric', '', 1, 'org.apache.hadoop.hive.ql.exec.WindowFunctionInfo']
+
+    GetAllFunctionsResponse get_all_functions() would also be nice as more detailed it seems, but this would be a call to HMS.
+    '''
     req = TGetFunctionsReq(functionName='.*')
+    thrift_function = self._client.GetFunctions
     (res, session) = self.call(self._client.GetFunctions, req)
 
-    results, schema = self.fetch_result(res.operationHandle, orientation=TFetchOrientation.FETCH_NEXT)
-    self._close(res.operationHandle, session)
-
-    return results, schema
+    return self.execute_async_statement(thrift_function=thrift_function, thrift_request=req)
 
 
 class HiveServerTableCompatible(HiveServerTable):
@@ -1337,8 +1349,8 @@ class HiveServerClientCompatible(object):
     if max_rows is None:
       max_rows = 1000
 
-    if start_over and not (self.query_server.get('dialect') == 'impala' and self.query_server['querycache_rows'] == 0): # Backward compatibility for impala
-      orientation = TFetchOrientation.FETCH_FIRST
+    if start_over and not (self.query_server.get('dialect') == 'impala' and self.query_server['querycache_rows'] == 0):
+      orientation = TFetchOrientation.FETCH_FIRST  # Backward compatibility for impala
     else:
       orientation = TFetchOrientation.FETCH_NEXT
 
@@ -1450,8 +1462,10 @@ class HiveServerClientCompatible(object):
 
   def alter_partition(self, db_name, tbl_name, new_part): raise NotImplementedError()
 
+
   def get_configuration(self):
     return self._client.get_configuration()
 
-  def get_functions(self, database, table):
-    return self._client.get_functions(database, table)
+
+  def get_functions(self):
+    return self._client.get_functions()

+ 1 - 2
desktop/core/src/desktop/lib/thrift_util.py

@@ -796,8 +796,7 @@ def enum_as_sequence(enum):
   Arguments:
   - `enum`: The class of a Thrift-generated enum
   """
-  return [x for x in dir(enum) if not x.startswith("__")
-                and  x not in ["_VALUES_TO_NAMES", "_NAMES_TO_VALUES", "next"]]
+  return [x for x in dir(enum) if not x.startswith("__") and x not in ["_VALUES_TO_NAMES", "_NAMES_TO_VALUES", "next"]]
 
 def fixup_enums(obj, name_class_map, suffix="AsString"):
   """