浏览代码

HUE-75 [beeswax] HS2 support for partitions

Add table:
- comments
- parameters
- path location

Found HS2 bug with truncated extended describe.
We are parsing DESCRIBE EXTENDED text sometimes and might need to implement the metastore API instead at some point.
Romain Rigaux 12 年之前
父节点
当前提交
88d1d2c821

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

@@ -92,6 +92,9 @@ class Dbms:
 
 
 
 
   def get_table(self, database, table_name):
   def get_table(self, database, table_name):
+    # DB name not supported in SHOW PARTITIONS required in Table
+    self.use(database)
+
     return self.client.get_table(database, table_name)
     return self.client.get_table(database, table_name)
 
 
 
 
@@ -305,11 +308,15 @@ class Dbms:
     if max_parts is None or max_parts > BROWSE_PARTITIONED_TABLE_LIMIT.get():
     if max_parts is None or max_parts > BROWSE_PARTITIONED_TABLE_LIMIT.get():
       max_parts = BROWSE_PARTITIONED_TABLE_LIMIT.get()
       max_parts = BROWSE_PARTITIONED_TABLE_LIMIT.get()
 
 
+    # DB name not supported in SHOW PARTITIONS
+    self.use(db_name)
+
     return self.client.get_partitions(db_name, table.name, max_parts)
     return self.client.get_partitions(db_name, table.name, max_parts)
 
 
   def get_partition(self, db_name, table_name, partition_id):
   def get_partition(self, db_name, table_name, partition_id):
     table = self.get_table(db_name, table_name)
     table = self.get_table(db_name, table_name)
     partitions = self.get_partitions(db_name, table, max_parts=None)
     partitions = self.get_partitions(db_name, table, max_parts=None)
+
     partition_query = ""
     partition_query = ""
     for idx, key in enumerate(partitions[partition_id].values):
     for idx, key in enumerate(partitions[partition_id].values):
       partition_query += (idx > 0 and " AND " or "") + table.partition_keys[idx].name + "=" + key
       partition_query += (idx > 0 and " AND " or "") + table.partition_keys[idx].name + "=" + key

+ 68 - 13
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -38,7 +38,9 @@ LOG = logging.getLogger(__name__)
 
 
 
 
 class HiveServerTable(Table):
 class HiveServerTable(Table):
-
+  """
+  We are parsing DESCRIBE EXTENDED text sometimes and might need to implement the metastore API instead at some point.
+  """
   def __init__(self, table_results, table_schema, desc_results, desc_schema):
   def __init__(self, table_results, table_schema, desc_results, desc_schema):
     if not table_results.rows:
     if not table_results.rows:
       raise NoSuchObjectException()
       raise NoSuchObjectException()
@@ -57,18 +59,34 @@ class HiveServerTable(Table):
 
 
   @property
   @property
   def partition_keys(self):
   def partition_keys(self):
-    # TODO
-    return [][:conf.BROWSE_PARTITIONED_TABLE_LIMIT.get()]
+    describe = self.extended_describe
+    #  partitionKeys:[FieldSchema(name:datehour, type:int, comment:null)],
+    match = re.search('partitionKeys:\[([^\]]+)\]', describe)
+    if match is not None:
+      match = match.group(1)
+      return [PartitionKeyCompatible(partition)
+              for partition in re.findall('FieldSchema\((.+?)\)', match)]
+    else:
+      return []
 
 
   @property
   @property
   def path_location(self):
   def path_location(self):
-    # TODO
-    return ''
+    describe = self.extended_describe
+    match = re.search('location:([^,]+)', describe)
+    if match is not None:
+      match = match.group(1)
+    return match
 
 
   @property
   @property
   def parameters(self):
   def parameters(self):
-    # TODO
-    return {}
+    # Parses a list of: parameters:{serialization.format=1}),... parameters:{numPartitions=2, EXTERNAL=TRUE}
+    describe = self.extended_describe
+    params = re.findall('parameters:\{([^\}]+?)\}', describe)
+    if params:
+      params_list = ', '.join(params).split(', ')
+      return dict([param.split('=')for param in params_list])
+    else:
+      return {}
 
 
   @property
   @property
   def cols(self):
   def cols(self):
@@ -76,13 +94,16 @@ class HiveServerTable(Table):
     if sum([bool(col['col_name']) for col in cols]) == len(cols):
     if sum([bool(col['col_name']) for col in cols]) == len(cols):
       return cols
       return cols
     else:
     else:
-      return cols[:-2] # Drop last 2 lines of Extended describe
+      return cols[:-2] # Drop last 2 lines of extended describe
 
 
   @property
   @property
   def comment(self):
   def comment(self):
-    # TODO
-    return ''
+    return HiveServerTRow(self.table, self.table_schema).col('REMARKS')
 
 
+  @property
+  def extended_describe(self):
+    # Just keep the content and skip the last new line
+    return HiveServerTTableSchema(self.results, self.schema).cols()[-1]['data_type']
 
 
 
 
 class HiveServerTRowSet:
 class HiveServerTRowSet:
@@ -328,16 +349,23 @@ class HiveServerClient:
     table_results, table_schema = self.fetch_result(res.operationHandle)
     table_results, table_schema = self.fetch_result(res.operationHandle)
 
 
     # Using 'SELECT * from table' does not show column comments in the metadata
     # Using 'SELECT * from table' does not show column comments in the metadata
+    if self.query_server['server_name'] == 'beeswax':
+      self.execute_statement(statement='SET hive.server2.blocking.query=true')
+
     desc_results, desc_schema = self.execute_statement('DESCRIBE EXTENDED %s' % table_name)
     desc_results, desc_schema = self.execute_statement('DESCRIBE EXTENDED %s' % table_name)
     return HiveServerTable(table_results.results, table_schema.schema, desc_results.results, desc_schema.schema)
     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=100):
+    return self.execute_query_statement(statement=query.query['query'], max_rows=max_rows)
+
+
+  def execute_query_statement(self, statement, max_rows=100):
     # Only execute_async_query() supports configuration
     # Only execute_async_query() supports configuration
     if self.query_server['server_name'] == 'beeswax':
     if self.query_server['server_name'] == 'beeswax':
       self.execute_statement(statement='SET hive.server2.blocking.query=true')
       self.execute_statement(statement='SET hive.server2.blocking.query=true')
 
 
-    results, schema = self.execute_statement(statement=query.query['query'], max_rows=max_rows)
+    results, schema = self.execute_statement(statement=statement, max_rows=max_rows)
     return HiveServerDataTable(results, schema)
     return HiveServerDataTable(results, schema)
 
 
 
 
@@ -417,6 +445,15 @@ class HiveServerClient:
     return res.log
     return res.log
 
 
 
 
+  def get_partitions(self, database, table_name, max_parts):
+    table = self.get_table(database, table_name)
+    if self.query_server['server_name'] == 'beeswax':
+      self.execute_statement(statement='SET hive.server2.blocking.query=true')
+
+    partitionTable = self.execute_query_statement('SHOW PARTITIONS %s' % table_name) # DB prefix not supported
+    return [PartitionValueCompatible(partition, table) for partition in partitionTable.rows()][-max_parts:]
+
+
 class HiveServerTableCompatible(HiveServerTable):
 class HiveServerTableCompatible(HiveServerTable):
   """Same API as Beeswax"""
   """Same API as Beeswax"""
 
 
@@ -446,11 +483,28 @@ class ResultCompatible:
   def columns(self):
   def columns(self):
     return self.cols()
     return self.cols()
 
 
-
   def cols(self):
   def cols(self):
     return [col.name for col in self.data_table.cols()]
     return [col.name for col in self.data_table.cols()]
 
 
 
 
+class PartitionKeyCompatible:
+
+  def __init__(self, partition):
+    # Parses: ['name:datehour, type:int, comment:null']
+    name, type, comment = partition.split(', ')
+    self.name = name.split(':')[1]
+    self.type = type.split(':')[1]
+    self.comment = comment.split(':')[1]
+
+
+class PartitionValueCompatible:
+
+  def __init__(self, partition, table):
+    # Parses: ['datehour=2013022516']
+    self.values = [part.split('=')[1] for part in partition]
+    self.sd = type('Sd', (object,), {'location': '%s/%s' % (table.path_location, ','.join(partition)),})
+
+
 class HiveServerClientCompatible:
 class HiveServerClientCompatible:
   """Same API as Beeswax"""
   """Same API as Beeswax"""
 
 
@@ -546,7 +600,8 @@ class HiveServerClientCompatible:
   def get_partition(self, *args, **kwargs): raise NotImplementedError()
   def get_partition(self, *args, **kwargs): raise NotImplementedError()
 
 
 
 
-  def get_partitions(self, *args, **kwargs): raise NotImplementedError()
+  def get_partitions(self, database, table_name, max_parts):
+    return self._client.get_partitions(database, table_name, max_parts)
 
 
 
 
   def alter_partition(self, db_name, tbl_name, new_part): raise NotImplementedError()
   def alter_partition(self, db_name, tbl_name, new_part): raise NotImplementedError()

+ 2 - 2
apps/beeswax/src/beeswax/templates/watch_results.mako

@@ -268,10 +268,10 @@ $(document).ready(function () {
       });
       });
     }
     }
   });
   });
-  
+
   $(".dataTables_wrapper").css("min-height", "0");
   $(".dataTables_wrapper").css("min-height", "0");
   $(".dataTables_filter").hide();
   $(".dataTables_filter").hide();
-  
+
   $("input[name='save_target']").change(function () {
   $("input[name='save_target']").change(function () {
     $("#fieldRequired").addClass("hide");
     $("#fieldRequired").addClass("hide");
     $("input[name='target_dir']").removeClass("fieldError");
     $("input[name='target_dir']").removeClass("fieldError");

+ 7 - 7
apps/beeswax/src/beeswax/tests.py

@@ -1477,7 +1477,7 @@ def test_beeswax_get_kerberos_security():
 
 
 
 
 class MockDbms:
 class MockDbms:
-  
+
   def __init__(self, client, server_type):
   def __init__(self, client, server_type):
     pass
     pass
 
 
@@ -1498,15 +1498,15 @@ class MockServer(object):
 
 
   def tearDown(self):
   def tearDown(self):
     dbms.Dbms = dbms.OriginalBeeswaxApi
     dbms.Dbms = dbms.OriginalBeeswaxApi
-  
-  def test_save_design_properties(self):  
+
+  def test_save_design_properties(self):
     resp = self.client.get('/beeswax/save_design_properties')
     resp = self.client.get('/beeswax/save_design_properties')
     content = json.loads(resp.content)
     content = json.loads(resp.content)
     assert_equal(-1, content['status'])
     assert_equal(-1, content['status'])
-  
+
     response = _make_query(self.client, 'SELECT', submission_type='Save', name='My Name', desc='My Description')
     response = _make_query(self.client, 'SELECT', submission_type='Save', name='My Name', desc='My Description')
     design = response.context['design']
     design = response.context['design']
-  
+
     try:
     try:
       resp = self.client.post('/beeswax/save_design_properties', {'name': 'name', 'value': 'New Name', 'pk': design.id})
       resp = self.client.post('/beeswax/save_design_properties', {'name': 'name', 'value': 'New Name', 'pk': design.id})
       design = SavedQuery.objects.get(id=design.id)
       design = SavedQuery.objects.get(id=design.id)
@@ -1516,10 +1516,10 @@ class MockServer(object):
       assert_equal('My Description', design.desc)
       assert_equal('My Description', design.desc)
     finally:
     finally:
       design.delete()
       design.delete()
-  
+
     response = _make_query(self.client, 'SELECT', submission_type='Save', name='My Name', desc='My Description')
     response = _make_query(self.client, 'SELECT', submission_type='Save', name='My Name', desc='My Description')
     design = response.context['design']
     design = response.context['design']
-  
+
     try:
     try:
       resp = self.client.post('/beeswax/save_design_properties', {'name': 'description', 'value': 'New Description', 'pk': design.id})
       resp = self.client.post('/beeswax/save_design_properties', {'name': 'description', 'value': 'New Description', 'pk': design.id})
       design = SavedQuery.objects.get(id=design.id)
       design = SavedQuery.objects.get(id=design.id)

+ 2 - 4
apps/catalog/src/catalog/templates/describe_partitions.mako

@@ -24,8 +24,8 @@
 ${ commonheader(_('Table Partitions: %(tableName)s') % dict(tableName=table.name), app_name, user) | n,unicode }
 ${ commonheader(_('Table Partitions: %(tableName)s') % dict(tableName=table.name), app_name, user) | n,unicode }
 
 
 <div class="container-fluid">
 <div class="container-fluid">
-<h1>${_('Partitions')}</h1>
-${ components.breadcrumbs(breadcrumbs) }
+  <h1>${_('Partitions')}</h1>
+  ${ components.breadcrumbs(breadcrumbs) }
 
 
   <div class="row-fluid">
   <div class="row-fluid">
     <div class="span2">
     <div class="span2">
@@ -69,8 +69,6 @@ ${ components.breadcrumbs(breadcrumbs) }
       </table>
       </table>
     </div>
     </div>
   </div>
   </div>
-
-
 </div>
 </div>
 
 
 <link rel="stylesheet" href="/catalog/static/css/catalog.css" type="text/css">
 <link rel="stylesheet" href="/catalog/static/css/catalog.css" type="text/css">

+ 3 - 0
apps/catalog/src/catalog/tests.py

@@ -111,6 +111,9 @@ class TestCatalogWithHadoop(BeeswaxSampleProvider):
     assert_true("myview" in resp.content)
     assert_true("myview" in resp.content)
 
 
   def test_describe_partitions(self):
   def test_describe_partitions(self):
+    response = self.client.get("/catalog/table/default/test_partitions")
+    assert_true("Show Partitions (1)" in response.content)
+
     response = self.client.get("/catalog/table/default/test_partitions/partitions", follow=True)
     response = self.client.get("/catalog/table/default/test_partitions/partitions", follow=True)
     assert_true("baz_one" in response.content)
     assert_true("baz_one" in response.content)
     assert_true("boom_two" in response.content)
     assert_true("boom_two" in response.content)

+ 2 - 4
apps/catalog/src/catalog/views.py

@@ -131,12 +131,10 @@ def describe_table(request, database, table):
     error_message, logs = dbms.expand_exception(ex, db)
     error_message, logs = dbms.expand_exception(ex, db)
 
 
   return render("describe_table.mako", request, {
   return render("describe_table.mako", request, {
-    'breadcrumbs': [
-      {
+    'breadcrumbs': [{
         'name': database,
         'name': database,
         'url': reverse('catalog:show_tables', kwargs={'database': database})
         'url': reverse('catalog:show_tables', kwargs={'database': database})
-      },
-      {
+      }, {
         'name': str(table.name),
         'name': str(table.name),
         'url': reverse('catalog:describe_table', kwargs={'database': database, 'table': table.name})
         'url': reverse('catalog:describe_table', kwargs={'database': database, 'table': table.name})
       },
       },