Browse Source

HUE-2257 [hive] Supporting new column format and old row format

If Thrift version >= 7, we use the new column format.
We could optimize more but we need to stay barkward compatible without
modifying every pages.
Romain Rigaux 11 năm trước cách đây
mục cha
commit
d30132f

+ 2 - 1
apps/beeswax/src/beeswax/api.py

@@ -269,8 +269,9 @@ def execute(request, design_id=None):
   design = safe_get_design(request, query_type, design_id)
 
   try:
+    print 'aaa'
     query_form = get_query_form(request)
-
+    print 'bb'
     if query_form.is_valid():
       query_str = query_form.query.cleaned_data["query"]
       explain = request.GET.get('explain', 'false').lower() == 'true'

+ 2 - 2
apps/beeswax/src/beeswax/conf.py

@@ -60,7 +60,7 @@ SERVER_CONN_TIMEOUT = Config(
   type=int,
   help=_t('Timeout in seconds for Thrift calls.'))
 
-USE_GET_LOG_API = Config(
+USE_GET_LOG_API = Config( # To remove in Hue 4
   key='use_get_log_api',
   default=False,
   type=coerce_bool,
@@ -92,7 +92,7 @@ THRIFT_VERSION = Config(
   key="thrift_version",
   help=_t("Thrift version to use when communicating with HiveServer2"),
   type=int,
-  default=5
+  default=7
 )
 
 SSL = ConfigSection(

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

@@ -16,6 +16,7 @@
 # limitations under the License.
 
 import logging
+import itertools
 import re
 
 from itertools import imap
@@ -33,7 +34,7 @@ from TCLIService.ttypes import TOpenSessionReq, TGetTablesReq, TFetchResultsReq,
   TStatusCode, TGetResultSetMetadataReq, TGetColumnsReq, TTypeId,\
   TExecuteStatementReq, TGetOperationStatusReq, TFetchOrientation,\
   TCloseSessionReq, TGetSchemasReq, TGetLogReq, TCancelOperationReq,\
-  TCloseOperationReq, TFetchResultsResp, TRowSet
+  TCloseOperationReq, TFetchResultsResp, TRowSet, TProtocolVersion
 
 from beeswax import conf as beeswax_conf
 from beeswax import hive_site
@@ -53,9 +54,15 @@ class HiveServerTable(Table):
   Impala only supports a simple DESCRIBE.
   """
   def __init__(self, table_results, table_schema, desc_results, desc_schema):
-    if not table_results.rows:
-      raise NoSuchObjectException()
-    self.table = table_results.rows and table_results.rows[0] or ''
+    if beeswax_conf.THRIFT_VERSION.get() >= 7:
+      if not table_results.columns:
+        raise NoSuchObjectException()
+      self.table = table_results.columns or ''
+    else: # Deprecated. To remove in Hue 4.
+      if not table_results.rows:
+        raise NoSuchObjectException()
+      self.table = table_results.rows and table_results.rows[0] or ''
+
     self.table_schema = table_schema
     self.desc_results = desc_results
     self.desc_schema = desc_schema
@@ -155,7 +162,7 @@ class HiveServerTable(Table):
     return props
 
 
-class HiveServerTRowSet:
+class HiveServerTRowSet2:
   def __init__(self, row_set, schema):
     self.row_set = row_set
     self.rows = row_set.rows
@@ -167,24 +174,70 @@ class HiveServerTRowSet:
 
   def cols(self, col_names):
     cols_rows = []
-    for row in self.rows:
-      row = HiveServerTRow(row, self.schema)
-      cols = {}
-      for col_name in col_names:
-        cols[col_name] = row.col(col_name)
-      cols_rows.append(cols)
+
+    rs = HiveServerTRow2(self.row_set.columns, self.schema)
+    cols = [rs.full_col(name) for name in col_names]
+
+    for cols_row in itertools.izip(*cols):
+      cols_rows.append(dict(zip(col_names, cols_row)))
+
     return cols_rows
 
   def __iter__(self):
     return self
 
   def next(self):
-    if self.rows:
-      return HiveServerTRow(self.rows.pop(0), self.schema)
+    if self.row_set.columns:
+      return HiveServerTRow2(self.row_set.columns, self.schema)
     else:
       raise StopIteration
 
 
+class HiveServerTRow2:
+  def __init__(self, cols, schema):
+    self.cols = cols
+    self.schema = schema
+
+  def col(self, colName):
+    pos = self._get_col_position(colName)
+    return HiveServerTColumnValue2(self.cols[pos]).val[0] # Return only first element
+
+  def full_col(self, colName):
+    pos = self._get_col_position(colName)
+    return HiveServerTColumnValue2(self.cols[pos]).val # Return the full column and its values
+
+  def _get_col_position(self, column_name):
+    return filter(lambda (i, col): col.columnName == column_name, enumerate(self.schema.columns))[0][0]
+
+  def fields(self):
+    try:
+      return [HiveServerTColumnValue2(field).val.pop(0) for field in self.cols]
+    except IndexError:
+      raise StopIteration
+
+
+class HiveServerTColumnValue2:
+  def __init__(self, tcolumn_value):
+    self.column_value = tcolumn_value
+
+  @property
+  def val(self):
+    # TODO get index from schema
+    if self.column_value.boolVal is not None:
+      return self.column_value.boolVal.values
+    elif self.column_value.byteVal is not None:
+      return self.column_value.byteVal.values
+    elif self.column_value.i16Val is not None:
+      return self.column_value.i16Val.values
+    elif self.column_value.i32Val is not None:
+      return self.column_value.i32Val.values
+    elif self.column_value.i64Val is not None:
+      return self.column_value.i64Val.values
+    elif self.column_value.doubleVal is not None:
+      return self.column_value.doubleVal.values
+    elif self.column_value.stringVal is not None:
+      return self.column_value.stringVal.values
+
 
 class HiveServerDataTable(DataTable):
   def __init__(self, results, schema, operation_handle):
@@ -218,7 +271,7 @@ class HiveServerTTableSchema:
   def cols(self):
     try:
       return HiveServerTRowSet(self.columns, self.schema).cols(('col_name', 'data_type', 'comment'))
-    except:
+    except Exception:
       # Impala API is different
       cols = HiveServerTRowSet(self.columns, self.schema).cols(('name', 'type', 'comment'))
       for col in cols:
@@ -234,20 +287,54 @@ class HiveServerTTableSchema:
     return filter(lambda (i, col): col.columnName == column_name, enumerate(self.schema.columns))[0][0]
 
 
-class HiveServerTRow:
-  def __init__(self, row, schema):
-    self.row = row
-    self.schema = schema
-
-  def col(self, colName):
-    pos = self._get_col_position(colName)
-    return HiveServerTColumnValue(self.row.colVals[pos]).val
-
-  def _get_col_position(self, column_name):
-    return filter(lambda (i, col): col.columnName == column_name, enumerate(self.schema.columns))[0][0]
-
-  def fields(self):
-    return [HiveServerTColumnValue(field).val for field in self.row.colVals]
+if beeswax_conf.THRIFT_VERSION.get() >= 7:
+  HiveServerTRow = HiveServerTRow2
+  HiveServerTRowSet = HiveServerTRowSet2
+else:
+  # Deprecated. To remove in Hue 4.
+  class HiveServerTRow:
+    def __init__(self, row, schema):
+      self.row = row
+      self.schema = schema
+
+    def col(self, colName):
+      pos = self._get_col_position(colName)
+      return HiveServerTColumnValue(self.row.colVals[pos]).val
+
+    def _get_col_position(self, column_name):
+      return filter(lambda (i, col): col.columnName == column_name, enumerate(self.schema.columns))[0][0]
+
+    def fields(self):
+      return [HiveServerTColumnValue(field).val for field in self.row.colVals]
+
+  class HiveServerTRowSet:
+    def __init__(self, row_set, schema):
+      self.row_set = row_set
+      self.rows = row_set.rows
+      self.schema = schema
+      self.startRowOffset = row_set.startRowOffset
+
+    def is_empty(self):
+      return len(self.rows) == 0
+
+    def cols(self, col_names):
+      cols_rows = []
+      for row in self.rows:
+        row = HiveServerTRow(row, self.schema)
+        cols = {}
+        for col_name in col_names:
+          cols[col_name] = row.col(col_name)
+        cols_rows.append(cols)
+      return cols_rows
+
+    def __iter__(self):
+      return self
+
+    def next(self):
+      if self.rows:
+        return HiveServerTRow(self.rows.pop(0), self.schema)
+      else:
+        raise StopIteration
 
 
 class HiveServerTColumnValue:

+ 5 - 5
apps/beeswax/thrift/TCLIService.thrift

@@ -41,19 +41,19 @@ enum TProtocolVersion {
   HIVE_CLI_SERVICE_PROTOCOL_V1,
   
   // V2 adds support for asynchronous execution
-  HIVE_CLI_SERVICE_PROTOCOL_V2
+  HIVE_CLI_SERVICE_PROTOCOL_V2,
 
   // V3 add varchar type, primitive type qualifiers
-  HIVE_CLI_SERVICE_PROTOCOL_V3
+  HIVE_CLI_SERVICE_PROTOCOL_V3,
 
   // V4 add decimal precision/scale, char type
-  HIVE_CLI_SERVICE_PROTOCOL_V4
+  HIVE_CLI_SERVICE_PROTOCOL_V4,
 
   // V5 adds error details when GetOperationStatus returns in error state
-  HIVE_CLI_SERVICE_PROTOCOL_V5
+  HIVE_CLI_SERVICE_PROTOCOL_V5,
 
   // V6 uses binary type for binary payload (was string) and uses columnar result set
-  HIVE_CLI_SERVICE_PROTOCOL_V6
+  HIVE_CLI_SERVICE_PROTOCOL_V6,
 
   // V7 adds support for delegation token based connection
   HIVE_CLI_SERVICE_PROTOCOL_V7

+ 1 - 1
desktop/conf.dist/hue.ini

@@ -792,7 +792,7 @@
   # This will free all the query resources in HiveServer2, but also make its results inaccessible.
   ## close_queries=false
 
-  # Thrift version to use when communicating with HiveServer2
+  # Thrift version to use when communicating with HiveServer2. Version 7 is a faster prototype.
   ## thrift_version=5
 
   [[ssl]]

+ 1 - 1
desktop/conf/pseudo-distributed.ini.tmpl

@@ -798,7 +798,7 @@
   # This will free all the query resources in HiveServer2, but also make its results inaccessible.
   ## close_queries=false
 
-  # Thrift version to use when communicating with HiveServer2
+  # Thrift version to use when communicating with HiveServer2. Version 7 is a faster prototype.
   ## thrift_version=5
 
   [[ssl]]