Преглед на файлове

HUE-1561 [beeswax] HiveServer2 0.11 async query support

Romain Rigaux преди 12 години
родител
ревизия
52736494ae

+ 4 - 0
apps/beeswax/gen-py/TCLIService/constants.py

@@ -21,6 +21,8 @@ PRIMITIVE_TYPES = set([
     8,
     9,
     15,
+    16,
+    17,
 ])
 COMPLEX_TYPES = set([
     10,
@@ -49,4 +51,6 @@ TYPE_NAMES = {
     12 : "STRUCT",
     13 : "UNIONTYPE",
     15 : "DECIMAL",
+    16 : "NULL",
+    17 : "DATE",
 }

+ 27 - 3
apps/beeswax/gen-py/TCLIService/ttypes.py

@@ -18,13 +18,16 @@ except:
 
 class TProtocolVersion(object):
   HIVE_CLI_SERVICE_PROTOCOL_V1 = 0
+  HIVE_CLI_SERVICE_PROTOCOL_V2 = 1
 
   _VALUES_TO_NAMES = {
     0: "HIVE_CLI_SERVICE_PROTOCOL_V1",
+    1: "HIVE_CLI_SERVICE_PROTOCOL_V2",
   }
 
   _NAMES_TO_VALUES = {
     "HIVE_CLI_SERVICE_PROTOCOL_V1": 0,
+    "HIVE_CLI_SERVICE_PROTOCOL_V2": 1,
   }
 
 class TTypeId(object):
@@ -44,6 +47,8 @@ class TTypeId(object):
   UNION_TYPE = 13
   USER_DEFINED_TYPE = 14
   DECIMAL_TYPE = 15
+  NULL_TYPE = 16
+  DATE_TYPE = 17
 
   _VALUES_TO_NAMES = {
     0: "BOOLEAN_TYPE",
@@ -62,6 +67,8 @@ class TTypeId(object):
     13: "UNION_TYPE",
     14: "USER_DEFINED_TYPE",
     15: "DECIMAL_TYPE",
+    16: "NULL_TYPE",
+    17: "DATE_TYPE",
   }
 
   _NAMES_TO_VALUES = {
@@ -81,6 +88,8 @@ class TTypeId(object):
     "UNION_TYPE": 13,
     "USER_DEFINED_TYPE": 14,
     "DECIMAL_TYPE": 15,
+    "NULL_TYPE": 16,
+    "DATE_TYPE": 17,
   }
 
 class TStatusCode(object):
@@ -114,6 +123,7 @@ class TOperationState(object):
   CLOSED_STATE = 4
   ERROR_STATE = 5
   UKNOWN_STATE = 6
+  PENDING_STATE = 7
 
   _VALUES_TO_NAMES = {
     0: "INITIALIZED_STATE",
@@ -123,6 +133,7 @@ class TOperationState(object):
     4: "CLOSED_STATE",
     5: "ERROR_STATE",
     6: "UKNOWN_STATE",
+    7: "PENDING_STATE",
   }
 
   _NAMES_TO_VALUES = {
@@ -133,6 +144,7 @@ class TOperationState(object):
     "CLOSED_STATE": 4,
     "ERROR_STATE": 5,
     "UKNOWN_STATE": 6,
+    "PENDING_STATE": 7,
   }
 
 class TOperationType(object):
@@ -2425,7 +2437,7 @@ class TOpenSessionReq(object):
 
   thrift_spec = (
     None, # 0
-    (1, TType.I32, 'client_protocol', None,     0, ), # 1
+    (1, TType.I32, 'client_protocol', None,     1, ), # 1
     (2, TType.STRING, 'username', None, None, ), # 2
     (3, TType.STRING, 'password', None, None, ), # 3
     (4, TType.MAP, 'configuration', (TType.STRING,None,TType.STRING,None), None, ), # 4
@@ -2534,7 +2546,7 @@ class TOpenSessionResp(object):
   thrift_spec = (
     None, # 0
     (1, TType.STRUCT, 'status', (TStatus, TStatus.thrift_spec), None, ), # 1
-    (2, TType.I32, 'serverProtocolVersion', None,     0, ), # 2
+    (2, TType.I32, 'serverProtocolVersion', None,     1, ), # 2
     (3, TType.STRUCT, 'sessionHandle', (TSessionHandle, TSessionHandle.thrift_spec), None, ), # 3
     (4, TType.MAP, 'configuration', (TType.STRING,None,TType.STRING,None), None, ), # 4
   )
@@ -3041,6 +3053,7 @@ class TExecuteStatementReq(object):
    - sessionHandle
    - statement
    - confOverlay
+   - runAsync
   """
 
   thrift_spec = (
@@ -3048,12 +3061,14 @@ class TExecuteStatementReq(object):
     (1, TType.STRUCT, 'sessionHandle', (TSessionHandle, TSessionHandle.thrift_spec), None, ), # 1
     (2, TType.STRING, 'statement', None, None, ), # 2
     (3, TType.MAP, 'confOverlay', (TType.STRING,None,TType.STRING,None), None, ), # 3
+    (4, TType.BOOL, 'runAsync', None, False, ), # 4
   )
 
-  def __init__(self, sessionHandle=None, statement=None, confOverlay=None,):
+  def __init__(self, sessionHandle=None, statement=None, confOverlay=None, runAsync=thrift_spec[4][4],):
     self.sessionHandle = sessionHandle
     self.statement = statement
     self.confOverlay = confOverlay
+    self.runAsync = runAsync
 
   def read(self, iprot):
     if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
@@ -3086,6 +3101,11 @@ class TExecuteStatementReq(object):
           iprot.readMapEnd()
         else:
           iprot.skip(ftype)
+      elif fid == 4:
+        if ftype == TType.BOOL:
+          self.runAsync = iprot.readBool();
+        else:
+          iprot.skip(ftype)
       else:
         iprot.skip(ftype)
       iprot.readFieldEnd()
@@ -3112,6 +3132,10 @@ class TExecuteStatementReq(object):
         oprot.writeString(viter135)
       oprot.writeMapEnd()
       oprot.writeFieldEnd()
+    if self.runAsync is not None:
+      oprot.writeFieldBegin('runAsync', TType.BOOL, 4)
+      oprot.writeBool(self.runAsync)
+      oprot.writeFieldEnd()
     oprot.writeFieldStop()
     oprot.writeStructEnd()
 

+ 0 - 15
apps/beeswax/gen-py/__init__.py

@@ -1,15 +0,0 @@
-# Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.

+ 1 - 0
apps/beeswax/src/beeswax/models.py

@@ -166,6 +166,7 @@ class HiveServerQueryHistory(QueryHistory):
     TOperationState.CLOSED_STATE         : QueryHistory.STATE.expired,
     TOperationState.ERROR_STATE        : QueryHistory.STATE.failed,
     TOperationState.UKNOWN_STATE        : QueryHistory.STATE.failed,
+    TOperationState.PENDING_STATE        : QueryHistory.STATE.submitted,
   }
 
   node_type = HIVE_SERVER2

+ 2 - 4
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -416,10 +416,8 @@ class HiveServerClient:
           self.execute_statement(resource.strip())
 
     configuration = self._get_query_configuration(query)
-    if self.query_server['server_name'] == 'beeswax':
-      configuration.update({'hive.server2.blocking.query': 'false'})
-
     query_statement =  query.get_query_statement(statement)
+
     return self.execute_async_statement(statement=query_statement, confOverlay=configuration)
 
 
@@ -431,7 +429,7 @@ class HiveServerClient:
 
 
   def execute_async_statement(self, statement, confOverlay):
-    req = TExecuteStatementReq(statement=statement.encode('utf-8'), confOverlay=confOverlay)
+    req = TExecuteStatementReq(statement=statement.encode('utf-8'), confOverlay=confOverlay, runAsync=True)
     res = self.call(self._client.ExecuteStatement, req)
 
     return HiveServerQueryHandle(secret=res.operationHandle.operationId.secret,

+ 33 - 20
apps/beeswax/thrift/TCLIService.thrift

@@ -38,7 +38,10 @@ namespace cpp apache.hive.service.cli.thrift
 // List of protocol versions. A new token should be
 // added to the end of this list every time a change is made.
 enum TProtocolVersion {
-  HIVE_CLI_SERVICE_PROTOCOL_V1
+  HIVE_CLI_SERVICE_PROTOCOL_V1,
+
+  // V2 adds support for asynchronous execution
+  HIVE_CLI_SERVICE_PROTOCOL_V2
 }
 
 enum TTypeId {
@@ -57,21 +60,25 @@ enum TTypeId {
   STRUCT_TYPE,
   UNION_TYPE,
   USER_DEFINED_TYPE,
-  DECIMAL_TYPE
+  DECIMAL_TYPE,
+  NULL_TYPE,
+  DATE_TYPE
 }
 
 const set<TTypeId> PRIMITIVE_TYPES = [
-  TTypeId.BOOLEAN_TYPE
-  TTypeId.TINYINT_TYPE
-  TTypeId.SMALLINT_TYPE
-  TTypeId.INT_TYPE
-  TTypeId.BIGINT_TYPE
-  TTypeId.FLOAT_TYPE
-  TTypeId.DOUBLE_TYPE
-  TTypeId.STRING_TYPE
-  TTypeId.TIMESTAMP_TYPE
+  TTypeId.BOOLEAN_TYPE,
+  TTypeId.TINYINT_TYPE,
+  TTypeId.SMALLINT_TYPE,
+  TTypeId.INT_TYPE,
+  TTypeId.BIGINT_TYPE,
+  TTypeId.FLOAT_TYPE,
+  TTypeId.DOUBLE_TYPE,
+  TTypeId.STRING_TYPE,
+  TTypeId.TIMESTAMP_TYPE,
   TTypeId.BINARY_TYPE,
-  TTypeId.DECIMAL_TYPE
+  TTypeId.DECIMAL_TYPE,
+  TTypeId.NULL_TYPE
+  TTypeId.DATE_TYPE
 ]
 
 const set<TTypeId> COMPLEX_TYPES = [
@@ -101,8 +108,10 @@ const map<TTypeId,string> TYPE_NAMES = {
   TTypeId.ARRAY_TYPE: "ARRAY",
   TTypeId.MAP_TYPE: "MAP",
   TTypeId.STRUCT_TYPE: "STRUCT",
-  TTypeId.UNION_TYPE: "UNIONTYPE"
-  TTypeId.DECIMAL_TYPE: "DECIMAL"
+  TTypeId.UNION_TYPE: "UNIONTYPE",
+  TTypeId.DECIMAL_TYPE: "DECIMAL",
+  TTypeId.NULL_TYPE: "NULL"
+  TTypeId.DATE_TYPE: "DATE"
 }
 
 // Thrift does not support recursively defined types or forward declarations,
@@ -285,7 +294,7 @@ union TColumnValue {
   4: TI32Value    i32Val       // INT
   5: TI64Value    i64Val       // BIGINT, TIMESTAMP
   6: TDoubleValue doubleVal    // FLOAT, DOUBLE
-  7: TStringValue stringVal    // STRING, LIST, MAP, STRUCT, UNIONTYPE, BINARY, DECIMAL
+  7: TStringValue stringVal    // STRING, LIST, MAP, STRUCT, UNIONTYPE, BINARY, DECIMAL, NULL
 }
 
 // Represents a row in a rowset.
@@ -350,6 +359,9 @@ enum TOperationState {
 
   // The operation is in an unrecognized state
   UKNOWN_STATE,
+
+  // The operation is in an pending state
+  PENDING_STATE,
 }
 
 
@@ -446,7 +458,7 @@ struct TOperationHandle {
 // which operations may be executed.
 struct TOpenSessionReq {
   // The version of the HiveServer2 protocol that the client is using.
-  1: required TProtocolVersion client_protocol = TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V1
+  1: required TProtocolVersion client_protocol = TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V2
 
   // Username and password for authentication.
   // Depending on the authentication scheme being used,
@@ -465,7 +477,7 @@ struct TOpenSessionResp {
   1: required TStatus status
 
   // The protocol version that the server is using.
-  2: required TProtocolVersion serverProtocolVersion = TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V1
+  2: required TProtocolVersion serverProtocolVersion = TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V2
 
   // Session Handle
   3: optional TSessionHandle sessionHandle
@@ -576,7 +588,7 @@ struct TGetInfoResp {
 // status of the statement, and to fetch results once the
 // statement has finished executing.
 struct TExecuteStatementReq {
-  // The session to exexcute the statement against
+  // The session to execute the statement against
   1: required TSessionHandle sessionHandle
 
   // The statement to be executed (DML, DDL, SET, etc)
@@ -587,6 +599,9 @@ struct TExecuteStatementReq {
   // is executed. These properties apply to this statement
   // only and will not affect the subsequent state of the Session.
   3: optional map<string, string> confOverlay
+
+  // Execute asynchronously when runAsync is true
+  4: optional bool runAsync = false
 }
 
 struct TExecuteStatementResp {
@@ -594,7 +609,6 @@ struct TExecuteStatementResp {
   2: optional TOperationHandle operationHandle
 }
 
-
 // GetTypeInfo()
 //
 // Get information about types supported by the HiveServer instance.
@@ -1013,4 +1027,3 @@ service TCLIService {
 
   TGetLogResp GetLog(1:TGetLogReq req);
 }
-