浏览代码

HUE-2768 [hive] Support nested types

Adds an additional beeswax API autocomplete endpoint for nested data types. For nested data types, response will include the extended_type, type, and the corresponding nested structures (elem, key, value, fields).

New API endpoints:
beeswax/api/autocomplete/<database>/<table>/<column>
beeswax/api/autocomplete/<database>/<table>/<column>/<nested_type>/<nested_type>/.../

where <nested_type> token can be:
$elem$ : returns the element type info, if the parent type is an array
$key$ : returns the key type info, if the parent type is a map
$value$ : returns the value type info, if the parent type is a map
fieldname : returns the field type info, if the parent type is a struct and fieldname is a field within the struct

All column and nested type responses include:
"extended_type" : full type-string of the current type (e.g. - "struct<address1:string,address2:string,city:string,state:string,zipcode:string>")
"type" : simple type-string of the current type (e.g. - "struct")

For arrays, maps, and structs, the additional info is included in the response:
Array:
  "elem": {"extended_type": <extended_type>, "type": <simple_type>}
Map:
  "key": {"extended_type": <extended_type", "type": <simple_type>}
  "value": {"extended_type": <extended_type", "type": <simple_type>}
Struct:
  "fields": [<list of fieldnames>]
  "extended_fields": [<list of field objects with "comment", "type", and "name">]

Primitive types will only include "type" and "extended_type" in the response, where both values are equivalent.
Jenny Kim 10 年之前
父节点
当前提交
4c02a42

+ 96 - 3
apps/beeswax/src/beeswax/api.py

@@ -39,7 +39,7 @@ from beeswax.conf import USE_GET_LOG_API
 from beeswax.server import dbms
 from beeswax.server.dbms import expand_exception, get_query_server_config, QueryServerException
 from beeswax.views import authorized_get_design, authorized_get_query_history, make_parameterization_form,\
-                          safe_get_design, save_design, massage_columns_for_json, _get_query_handle_and_state,\
+                          safe_get_design, save_design, massage_columns_for_json, _get_query_handle_and_state, \
                           _parse_out_hadoop_jobs
 
 
@@ -81,7 +81,7 @@ def error_handler(view_fn):
 
 
 @error_handler
-def autocomplete(request, database=None, table=None):
+def autocomplete(request, database=None, table=None, column=None, nested=None):
   app_name = get_app_name(request)
   query_server = get_query_server_config(app_name)
   do_as = request.user
@@ -95,11 +95,25 @@ def autocomplete(request, database=None, table=None):
       response['databases'] = db.get_databases()
     elif table is None:
       response['tables'] = db.get_tables(database=database)
-    else:
+    elif column is None:
       t = db.get_table(database, table)
       response['hdfs_link'] = t.hdfs_link
       response['columns'] = [column.name for column in t.cols]
       response['extended_columns'] = massage_columns_for_json(t.cols)
+    else:
+      if app_name == 'beeswax':
+        if nested is None:  # autocomplete column
+          t = db.get_table(database, table)
+          current = db.get_column(database, table, column)
+          extended_type, simple_type = _get_column_type_by_name(t.cols, column)
+        else:  # autocomplete nested data type
+          current, extended_type, simple_type = _get_nested_describe_and_types(db, database, table, column, nested)
+
+        response['extended_type'] = extended_type
+        response['type'] = simple_type
+
+        inner_type = _get_complex_inner_type(current, extended_type, simple_type)
+        response.update(inner_type)
   except TTransportException, tx:
     response['code'] = 503
     response['error'] = tx.message
@@ -683,3 +697,82 @@ def get_top_terms(request, database, table, column, prefix=None):
   response['status'] = 0
 
   return JsonResponse(response)
+
+
+"""
+Utils
+"""
+def _get_simple_data_type(type_string=None):
+  if type_string:
+    pattern = re.compile('^([a-z]+)(<.+>)?$', re.IGNORECASE)
+    match = re.search(pattern, type_string)
+    return match.group(1)
+  return None
+
+
+def _get_column_type_by_name(columns, column_name):
+  full_type = simple_type = None
+
+  for column in columns:
+    if column.name == column_name:
+      full_type = column.type
+      simple_type = _get_simple_data_type(column.type)
+
+  return full_type, simple_type
+
+
+def _extract_nested_type(type_string, token):
+  full_type = simple_type = None
+
+  if token == '$elem$':
+    full_type = re.search(r"array<(.+)>$", type_string).group(1)
+  elif token == '$key$':
+    full_type = re.search(r"map<(\w+),.+>$", type_string).group(1)
+  elif token == '$value$':
+    full_type= re.search(r"map<\w+,(.+)>$", type_string).group(1)
+
+  if full_type:
+    simple_type = _get_simple_data_type(full_type)
+
+  return full_type, simple_type
+
+
+def _get_parent_describe_and_types(db, database, table, column, nested_tokens):
+  parent_token = nested_tokens[-2] if len(nested_tokens) > 1 else column
+  parent = db.get_column(database, table, column, nested_tokens[:-1])
+  extended_type, simple_type = _get_column_type_by_name(parent.cols, parent_token)
+
+  return parent, extended_type, simple_type
+
+
+def _get_nested_describe_and_types(db, database, table, column, nested):
+  nested_tokens = nested.strip('/').split('/')
+  last_token = nested_tokens[-1]
+  parent, parent_type, parent_simple_type = _get_parent_describe_and_types(db, database, table, column, nested_tokens)
+  current = db.get_column(database, table, column, nested_tokens)
+
+  if last_token in ('$elem$', '$key$', '$value$'):  # ARRAY and MAP types must be parsed from parent_type
+    extended_type, simple_type = _extract_nested_type(parent_type, last_token)
+  else:  # STRUCT or primitive type must be looked up from parent DESCRIBE table
+    extended_type, simple_type = _get_column_type_by_name(parent.cols, last_token)
+
+  return current, extended_type, simple_type
+
+
+def _get_complex_inner_type(current, extended_type, simple_type):
+  inner_type = {}
+
+  # Add nested fields to response
+  if simple_type == 'array':
+    full, short = _extract_nested_type(extended_type, '$elem$')
+    inner_type['elem'] = {'extended_type': full, 'type': short}
+  elif simple_type == 'map':
+    full, short = _extract_nested_type(extended_type, '$key$')
+    inner_type['key'] = {'extended_type': full, 'type': short}
+    full, short = _extract_nested_type(extended_type, '$value$')
+    inner_type['value'] = {'extended_type': full, 'type': short}
+  elif simple_type == 'struct':
+    inner_type['extended_fields'] = massage_columns_for_json(current.cols)
+    inner_type['fields'] = [column.name for column in current.cols]
+
+  return inner_type

+ 19 - 14
apps/beeswax/src/beeswax/server/dbms.py

@@ -117,25 +117,30 @@ class HiveServer2Dbms(object):
     self.server_type = server_type
     self.server_name = self.client.query_server['server_name']
 
-  def get_table(self, database, table_name):
-    return self.client.get_table(database, table_name)
+
+  def get_databases(self):
+    return self.client.get_databases()
 
 
   def get_tables(self, database='default', table_names='*'):
-      hql = "SHOW TABLES IN %s '%s'" % (database, table_names) # self.client.get_tables(database, table_names) is too slow
-      query = hql_query(hql)
-      handle = self.execute_and_wait(query, timeout_sec=15.0)
+    hql = "SHOW TABLES IN %s '%s'" % (database, table_names) # self.client.get_tables(database, table_names) is too slow
+    query = hql_query(hql)
+    handle = self.execute_and_wait(query, timeout_sec=15.0)
 
-      if handle:
-        result = self.fetch(handle, rows=5000)
-        self.close(handle)
-        return [name for table in result.rows() for name in table]
-      else:
-        return []
+    if handle:
+      result = self.fetch(handle, rows=5000)
+      self.close(handle)
+      return [name for table in result.rows() for name in table]
+    else:
+      return []
 
 
-  def get_databases(self):
-    return self.client.get_databases()
+  def get_table(self, database, table_name):
+    return self.client.get_table(database, table_name)
+
+
+  def get_column(self, database, table_name, column_name, nested_tokens=None):
+    return self.client.get_table(database, table_name, column_name=column_name, nested_tokens=nested_tokens)
 
 
   def execute_query(self, query, design):
@@ -597,7 +602,7 @@ class HiveServer2Dbms(object):
     parts = ["%s='%s'" % (table.partition_keys[idx].name, key) for idx, key in enumerate(partitions[partition_id].values)]
     partition_spec = ','.join(parts)
 
-    describe_table = self.client.get_table(db_name, table_name, partition_spec)
+    describe_table = self.client.get_table(db_name, table_name, partition_spec=partition_spec)
 
     return describe_table
 

+ 15 - 8
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -101,10 +101,12 @@ class HiveServerTable(Table):
   @property
   def cols(self):
     rows = self.describe
+    col_row_index = 2
     try:
-      col_row_index = 2
       end_cols_index = map(itemgetter('col_name'), rows[col_row_index:]).index('')
       return rows[col_row_index:][:end_cols_index] + self._get_partition_column()
+    except ValueError:  # DESCRIBE on columns and nested columns does not contain add'l rows beyond cols
+      return rows[col_row_index:]
     except:
       LOG.exception('failed to extract columns')
       return rows
@@ -603,17 +605,22 @@ class HiveServerClient:
     return HiveServerTRowSet(results.results, schema.schema).cols(('TABLE_NAME',))
 
 
-  def get_table(self, database, table_name, partition_spec=None):
+  def get_table(self, database, table_name, column_name=None, nested_tokens=None, partition_spec=None):
     req = TGetTablesReq(schemaName=database, tableName=table_name)
     res = self.call(self._client.GetTables, req)
 
     table_results, table_schema = self.fetch_result(res.operationHandle, orientation=TFetchOrientation.FETCH_NEXT)
     self.close_operation(res.operationHandle)
 
-    if partition_spec:
-      query = 'DESCRIBE FORMATTED `%s`.`%s` PARTITION(%s)' % (database, table_name, partition_spec)
+    if column_name and nested_tokens:  # DESCRIBE on nested types cannot accept database name
+      nested_spec = '.'.join('`%s`' % token for token in nested_tokens)
+      query = 'DESCRIBE FORMATTED `%s`.`%s`.%s' % (table_name, column_name, nested_spec)
     else:
       query = 'DESCRIBE FORMATTED `%s`.`%s`' % (database, table_name)
+      if column_name:
+        query += ' `%s`' % column_name
+      elif partition_spec:
+        query += ' PARTITION(%s)' % partition_spec
 
     (desc_results, desc_schema), operation_handle = self.execute_statement(query, max_rows=5000, orientation=TFetchOrientation.FETCH_NEXT)
     self.close_operation(operation_handle)
@@ -787,8 +794,8 @@ class HiveServerTableCompatible(HiveServerTable):
   def cols(self):
     return [
         type('Col', (object,), {
-          'name': col.get('col_name', '').strip(),
-          'type': col.get('data_type', '').strip(),
+          'name': col.get('col_name', '').strip() if col.get('col_name') else '',
+          'type': col.get('data_type', '').strip() if col.get('data_type') else '',
           'comment': col.get('comment', '').strip() if col.get('comment') else ''
         }) for col in HiveServerTable.cols.fget(self)
   ]
@@ -950,8 +957,8 @@ class HiveServerClientCompatible(object):
     return tables
 
 
-  def get_table(self, database, table_name, partition_spec=None):
-    table = self._client.get_table(database, table_name, partition_spec)
+  def get_table(self, database, table_name, column_name=None, nested_tokens=None, partition_spec=None):
+    table = self._client.get_table(database, table_name, column_name, nested_tokens, partition_spec)
     return HiveServerTableCompatible(table)
 
 

+ 59 - 0
apps/beeswax/src/beeswax/tests.py

@@ -57,6 +57,7 @@ import beeswax.models
 import beeswax.views
 
 from beeswax import conf, hive_site
+from beeswax.api import _get_simple_data_type, _extract_nested_type
 from beeswax.conf import HIVE_SERVER_HOST
 from beeswax.views import collapse_whitespace, _save_design
 from beeswax.test_base import make_query, wait_for_query_to_finish, verify_history, get_query_server_config,\
@@ -1708,6 +1709,64 @@ for x in sys.stdin:
     assert_equal([[109, 1], [108, 1]], terms)
 
 
+  def test_beeswax_api_autocomplete(self):
+    CREATE_TABLE = "CREATE TABLE `%(db)s`.`nested_table` (foo ARRAY<STRUCT<bar:INT, baz:STRING>>);" % {'db': self.db_name}
+    _make_query(self.client, CREATE_TABLE, wait=True)
+
+    resp = self.client.get(reverse("beeswax:api_autocomplete_databases", kwargs={}))
+    databases = json.loads(resp.content)['databases']
+    assert_true(self.db_name in databases)
+
+    # Autocomplete tables for a given database
+    resp = self.client.get(reverse("beeswax:api_autocomplete_tables", kwargs={'database': self.db_name}))
+    tables = json.loads(resp.content)['tables']
+    assert_true("nested_table" in tables)
+
+    # Autocomplete columns for a given table
+    resp = self.client.get(reverse("beeswax:api_autocomplete_columns", kwargs={'database': self.db_name, 'table': 'nested_table'}))
+    columns = json.loads(resp.content)['columns']
+    assert_true("foo" in columns)
+    extended_columns = json.loads(resp.content)['extended_columns']
+    assert_equal({'comment': '', 'type': 'array<struct<bar:int,baz:string>>', 'name': 'foo'}, extended_columns[0])
+
+    # Autocomplete nested fields for a given column
+    resp = self.client.get(reverse("beeswax:api_autocomplete_column", kwargs={'database': self.db_name, 'table': 'nested_table', 'column': 'foo'}))
+    json_resp = json.loads(resp.content)
+    assert_equal("array", json_resp['type'])
+    assert_equal("array<struct<bar:int,baz:string>>", json_resp['extended_type'])
+    assert_true("elem" in json_resp)
+    assert_equal("struct", json_resp["elem"]["type"])
+    assert_equal("struct<bar:int,baz:string>", json_resp["elem"]["extended_type"])
+
+    # Autcomplete nested fields for a given nested type
+    resp = self.client.get(reverse("beeswax:api_autocomplete_nested", kwargs={'database': self.db_name, 'table': 'nested_table', 'column': 'foo', 'nested': '$elem$'}))
+    json_resp = json.loads(resp.content)
+    assert_equal("struct", json_resp['type'])
+    assert_equal("struct<bar:int,baz:string>", json_resp['extended_type'])
+    assert_true("fields" in json_resp)
+    assert_true("extended_fields" in json_resp)
+
+
+def test_beeswax_api_get_simple_data_type():
+  # Return just the outer type name for complex nested types
+  assert_equal('array', _get_simple_data_type('array<struct<first_name:string,last_name:string>>'))
+  # Return same value for primitive types
+  assert_equal('string', _get_simple_data_type('string'))
+  # Return None if None
+  assert_equal(None, _get_simple_data_type())
+
+
+def test_beeswax_api_extract_nested_type():
+  # Return full and simple type of array element if token = $elem$
+  assert_equal(('struct<first_name:string,last_name:string>', 'struct'),
+               _extract_nested_type('array<struct<first_name:string,last_name:string>>', '$elem$'))
+  # Return full and simple type of map key if token = $key$
+  assert_equal(('string', 'string'), _extract_nested_type('map<string,array<string>>', '$key$'))
+  # Return full and simple type of map value if token = $value$
+  assert_equal(('array<string>', 'array'), _extract_nested_type('map<string,array<string>>', '$value$'))
+  # Return None, None if token is not valid
+  assert_equal((None, None), _extract_nested_type('struct<first_name:string,last_name:string>', 'first_name'))
+
 
 def test_import_gzip_reader():
   """Test the gzip reader in create table"""

+ 3 - 0
apps/beeswax/src/beeswax/urls.py

@@ -62,6 +62,9 @@ urlpatterns += patterns(
   url(r'^api/autocomplete/(?P<database>\w+)/$', 'autocomplete', name='api_autocomplete_tables'),
   url(r'^api/autocomplete/(?P<database>\w+)/(?P<table>\w+)$', 'autocomplete', name='api_autocomplete_columns'),
   url(r'^api/autocomplete/(?P<database>\w+)/(?P<table>\w+)/$', 'autocomplete', name='api_autocomplete_columns'),
+  url(r'^api/autocomplete/(?P<database>\w+)/(?P<table>\w+)/(?P<column>\w+)$', 'autocomplete', name='api_autocomplete_column'),
+  url(r'^api/autocomplete/(?P<database>\w+)/(?P<table>\w+)/(?P<column>\w+)/$', 'autocomplete', name='api_autocomplete_column'),
+  url(r'^api/autocomplete/(?P<database>\w+)/(?P<table>\w+)/(?P<column>\w+)/(?P<nested>.+)$', 'autocomplete', name='api_autocomplete_nested'),
   url(r'^api/design/(?P<design_id>\d+)?$', 'save_query_design', name='api_save_design'),
   url(r'^api/design/(?P<design_id>\d+)/get$', 'fetch_saved_design', name='api_fetch_saved_design'),
   url(r'^api/query/(?P<query_history_id>\d+)/get$', 'fetch_query_history', name='api_fetch_query_history'),

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

@@ -646,7 +646,6 @@ def query_done_cb(request, server_id):
 """
 Utils
 """
-
 def massage_columns_for_json(cols):
   massaged_cols = []
   for column in cols: