Explorar o código

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 %!s(int64=12) %!d(string=hai) anos
pai
achega
88d1d2c

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

@@ -92,6 +92,9 @@ class Dbms:
 
 
   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)
 
 
@@ -305,11 +308,15 @@ class Dbms:
     if max_parts is None or 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)
 
   def get_partition(self, db_name, table_name, partition_id):
     table = self.get_table(db_name, table_name)
     partitions = self.get_partitions(db_name, table, max_parts=None)
+
     partition_query = ""
     for idx, key in enumerate(partitions[partition_id].values):
       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):
-
+  """
+  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):
     if not table_results.rows:
       raise NoSuchObjectException()
@@ -57,18 +59,34 @@ class HiveServerTable(Table):
 
   @property
   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
   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
   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
   def cols(self):
@@ -76,13 +94,16 @@ class HiveServerTable(Table):
     if sum([bool(col['col_name']) for col in cols]) == len(cols):
       return cols
     else:
-      return cols[:-2] # Drop last 2 lines of Extended describe
+      return cols[:-2] # Drop last 2 lines of extended describe
 
   @property
   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:
@@ -328,16 +349,23 @@ class HiveServerClient:
     table_results, table_schema = self.fetch_result(res.operationHandle)
 
     # 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)
     return HiveServerTable(table_results.results, table_schema.schema, desc_results.results, desc_schema.schema)
 
 
   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
     if self.query_server['server_name'] == 'beeswax':
       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)
 
 
@@ -417,6 +445,15 @@ class HiveServerClient:
     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):
   """Same API as Beeswax"""
 
@@ -446,11 +483,28 @@ class ResultCompatible:
   def columns(self):
     return self.cols()
 
-
   def cols(self):
     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:
   """Same API as Beeswax"""
 
@@ -546,7 +600,8 @@ class HiveServerClientCompatible:
   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()

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

@@ -268,10 +268,10 @@ $(document).ready(function () {
       });
     }
   });
-  
+
   $(".dataTables_wrapper").css("min-height", "0");
   $(".dataTables_filter").hide();
-  
+
   $("input[name='save_target']").change(function () {
     $("#fieldRequired").addClass("hide");
     $("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:
-  
+
   def __init__(self, client, server_type):
     pass
 
@@ -1498,15 +1498,15 @@ class MockServer(object):
 
   def tearDown(self):
     dbms.Dbms = dbms.OriginalBeeswaxApi
-  
-  def test_save_design_properties(self):  
+
+  def test_save_design_properties(self):
     resp = self.client.get('/beeswax/save_design_properties')
     content = json.loads(resp.content)
     assert_equal(-1, content['status'])
-  
+
     response = _make_query(self.client, 'SELECT', submission_type='Save', name='My Name', desc='My Description')
     design = response.context['design']
-  
+
     try:
       resp = self.client.post('/beeswax/save_design_properties', {'name': 'name', 'value': 'New Name', 'pk': design.id})
       design = SavedQuery.objects.get(id=design.id)
@@ -1516,10 +1516,10 @@ class MockServer(object):
       assert_equal('My Description', design.desc)
     finally:
       design.delete()
-  
+
     response = _make_query(self.client, 'SELECT', submission_type='Save', name='My Name', desc='My Description')
     design = response.context['design']
-  
+
     try:
       resp = self.client.post('/beeswax/save_design_properties', {'name': 'description', 'value': 'New Description', 'pk': 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 }
 
 <div class="container-fluid">
-<h1>${_('Partitions')}</h1>
-${ components.breadcrumbs(breadcrumbs) }
+  <h1>${_('Partitions')}</h1>
+  ${ components.breadcrumbs(breadcrumbs) }
 
   <div class="row-fluid">
     <div class="span2">
@@ -69,8 +69,6 @@ ${ components.breadcrumbs(breadcrumbs) }
       </table>
     </div>
   </div>
-
-
 </div>
 
 <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)
 
   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)
     assert_true("baz_one" 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)
 
   return render("describe_table.mako", request, {
-    'breadcrumbs': [
-      {
+    'breadcrumbs': [{
         'name': database,
         'url': reverse('catalog:show_tables', kwargs={'database': database})
-      },
-      {
+      }, {
         'name': str(table.name),
         'url': reverse('catalog:describe_table', kwargs={'database': database, 'table': table.name})
       },