Jelajahi Sumber

HUE-1110 [impala] Support Hive Server 2 API

In order to use HS2 interface with Impala, update [impala]
server_port=21050.
Beeswax sessions are now differentiated among Beeswax and Impala
(that way both sessions do not conflict).
Bunch of little fixes for supporting both HS2 Hive/Impala apis.
Romain Rigaux 12 tahun lalu
induk
melakukan
bdaedab

+ 108 - 0
apps/beeswax/src/beeswax/migrations/0006_auto__add_field_session_application.py

@@ -0,0 +1,108 @@
+# encoding: utf-8
+import datetime
+from south.db import db
+from south.v2 import SchemaMigration
+from django.db import models
+
+class Migration(SchemaMigration):
+
+    def forwards(self, orm):
+
+        # Adding field 'Session.application'
+        db.add_column('beeswax_session', 'application', self.gf('django.db.models.fields.CharField')(default='beeswax', max_length=128), keep_default=False)
+
+
+    def backwards(self, orm):
+
+        # Deleting field 'Session.application'
+        db.delete_column('beeswax_session', 'application')
+
+
+    models = {
+        'auth.group': {
+            'Meta': {'object_name': 'Group'},
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
+            'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        'auth.permission': {
+            'Meta': {'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
+            'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
+        },
+        'auth.user': {
+            'Meta': {'object_name': 'User'},
+            'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
+            'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True', 'blank': 'True'}),
+            'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'blank': 'True'}),
+            'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'blank': 'True'}),
+            'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
+            'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
+            'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
+        },
+        'beeswax.metainstall': {
+            'Meta': {'object_name': 'MetaInstall'},
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'installed_example': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'blank': 'True'})
+        },
+        'beeswax.queryhistory': {
+            'Meta': {'object_name': 'QueryHistory'},
+            'design': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['beeswax.SavedQuery']", 'null': 'True'}),
+            'has_results': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'last_state': ('django.db.models.fields.IntegerField', [], {'db_index': 'True'}),
+            'log_context': ('django.db.models.fields.CharField', [], {'max_length': '1024', 'null': 'True'}),
+            'modified_row_count': ('django.db.models.fields.FloatField', [], {'null': 'True'}),
+            'notify': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'blank': 'True'}),
+            'operation_type': ('django.db.models.fields.SmallIntegerField', [], {'null': 'True'}),
+            'owner': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
+            'query': ('django.db.models.fields.TextField', [], {}),
+            'server_guid': ('django.db.models.fields.CharField', [], {'default': 'None', 'max_length': '1024', 'null': 'True'}),
+            'server_host': ('django.db.models.fields.CharField', [], {'default': "''", 'max_length': '128'}),
+            'server_id': ('django.db.models.fields.CharField', [], {'max_length': '1024', 'null': 'True'}),
+            'server_name': ('django.db.models.fields.CharField', [], {'default': "''", 'max_length': '128'}),
+            'server_port': ('django.db.models.fields.SmallIntegerField', [], {'default': '0'}),
+            'server_type': ('django.db.models.fields.CharField', [], {'default': "'beeswax'", 'max_length': '128'}),
+            'statement_number': ('django.db.models.fields.SmallIntegerField', [], {'default': '0'}),
+            'submission_date': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'})
+        },
+        'beeswax.savedquery': {
+            'Meta': {'object_name': 'SavedQuery'},
+            'data': ('django.db.models.fields.TextField', [], {'max_length': '65536'}),
+            'desc': ('django.db.models.fields.TextField', [], {'max_length': '1024'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_auto': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'db_index': 'True', 'blank': 'True'}),
+            'mtime': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '64'}),
+            'owner': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
+            'type': ('django.db.models.fields.IntegerField', [], {})
+        },
+        'beeswax.session': {
+            'Meta': {'object_name': 'Session'},
+            'application': ('django.db.models.fields.CharField', [], {'default': "'beeswax'", 'max_length': '128'}),
+            'guid': ('django.db.models.fields.TextField', [], {'max_length': "'100'"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'last_used': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'db_index': 'True', 'blank': 'True'}),
+            'owner': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"}),
+            'secret': ('django.db.models.fields.TextField', [], {'max_length': "'100'"}),
+            'server_protocol_version': ('django.db.models.fields.SmallIntegerField', [], {'default': '0'}),
+            'status_code': ('django.db.models.fields.PositiveSmallIntegerField', [], {})
+        },
+        'contenttypes.contenttype': {
+            'Meta': {'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
+            'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
+        }
+    }
+
+    complete_apps = ['beeswax']

+ 6 - 2
apps/beeswax/src/beeswax/models.py

@@ -333,20 +333,24 @@ class SavedQuery(models.Model):
 
 
 class SessionManager(models.Manager):
-  def get_session(self, user):
+  def get_session(self, user, application='beeswax'):
     try:
-      return self.filter(owner=user).latest("last_used")
+      return self.filter(owner=user, application=application).latest("last_used")
     except Session.DoesNotExist:
       pass
 
 
 class Session(models.Model):
+  """
+  A sessions is bound to a user and an application (e.g. Bob with the Impala application).
+  """
   owner = models.ForeignKey(User, db_index=True)
   status_code = models.PositiveSmallIntegerField()
   secret = models.TextField(max_length='100')
   guid = models.TextField(max_length='100')
   server_protocol_version = models.SmallIntegerField(default=0)
   last_used = models.DateTimeField(auto_now=True, db_index=True, verbose_name=_t('Last used'))
+  application = models.CharField(max_length=128, help_text=_t('Application we communicate with.'), default='beeswax')
 
   objects = SessionManager()
 

+ 2 - 2
apps/beeswax/src/beeswax/server/dbms.py

@@ -186,8 +186,8 @@ class Dbms:
 
 
   def use(self, database):
-    """Beeswax does not support use directly."""
-    if SERVER_INTERFACE.get() == HIVE_SERVER2:
+    """Beeswax interface does not support use directly. Impala app does not support it."""
+    if SERVER_INTERFACE.get() == HIVE_SERVER2 and self.client.query_server['server_name'] != 'impala':
       query = hql_query('USE %s' % database)
       return self.execute_and_wait(query)
 

+ 29 - 11
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -150,7 +150,7 @@ class HiveServerTTableSchema:
     return HiveServerTColumnDesc(self.columns[pos]).val
 
   def _get_col_position(self, column_name):
-    return filter(lambda col: col.columnName == column_name, self.schema.columns)[0].position - 1
+    return filter(lambda (i, col): col.columnName == column_name, enumerate(self.schema.columns))[0][0]
 
 
 class HiveServerTRow:
@@ -163,7 +163,7 @@ class HiveServerTRow:
     return HiveServerTColumnValue(self.row.colVals[pos]).val
 
   def _get_col_position(self, column_name):
-    return filter(lambda col: col.columnName == column_name, self.schema.columns)[0].position - 1
+    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]
@@ -249,6 +249,7 @@ class HiveServerClient:
     encoded_status, encoded_guid = HiveServerQueryHandle(secret=sessionId.secret, guid=sessionId.guid).get()
 
     return Session.objects.create(owner=user,
+                                  application=self.query_server['server_name'],
                                   status_code=res.status.statusCode,
                                   secret=encoded_status,
                                   guid=encoded_guid,
@@ -256,7 +257,7 @@ class HiveServerClient:
 
 
   def call(self, fn, req, status=TStatusCode.SUCCESS_STATUS):
-    session = Session.objects.get_session(self.user)
+    session = Session.objects.get_session(self.user, self.query_server['server_name'])
 
     if session is None:
       session = self.open_session(self.user)
@@ -285,7 +286,7 @@ class HiveServerClient:
 
 
   def close_session(self):
-    session = Session.objects.get_session(self.user).get_handle()
+    session = Session.objects.get_session(self.user, self.query_server['server_name']).get_handle()
 
     req = TCloseSessionReq(sessionHandle=session)
     return self._client.CloseSession(req)
@@ -298,7 +299,12 @@ class HiveServerClient:
 
     results, schema = self.fetch_result(res.operationHandle)
 
-    return HiveServerTRowSet(results.results, schema.schema).cols(('TABLE_SCHEMA',))
+    if self.query_server['server_name'] == 'impala':
+      col = 'TABLE_SCHEM'
+    else:
+      col = 'TABLE_SCHEMA'
+
+    return HiveServerTRowSet(results.results, schema.schema).cols((col,))
 
 
   def get_tables(self, database, table_names):
@@ -323,14 +329,16 @@ class HiveServerClient:
 
   def execute_query(self, query, max_rows=100):
     # TODO: Need to set jars, UDF etc
-    self.execute_statement(statement='SET hive.server2.blocking.query=true')
+    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)
     return HiveServerDataTable(results, schema)
 
 
   def execute_async_query(self, query, statement=0):
-    self.execute_statement(statement='SET hive.server2.blocking.query=false')
+    if self.query_server['server_name'] == 'beeswax':
+      self.execute_statement(statement='SET hive.server2.blocking.query=false')
 
     query_statement = query.get_query_statement(statement)
     return self.execute_async_statement(statement=query_statement)
@@ -367,7 +375,7 @@ class HiveServerClient:
 
     return self.fetch_result(res.operationHandle)
 
-
+# TFetchOrientation.FETCH_NEXT
   def fetch_result(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=100):
     fetch_req = TFetchResultsReq(operationHandle=operation_handle, orientation=orientation, maxRows=max_rows)
     res = self.call(self._client.FetchResults, fetch_req)
@@ -456,6 +464,9 @@ class HiveServerClientCompatible:
     if max_rows is None:
       max_rows = 10000
 
+    # Both Hive Server 2 and Impala do not support FETCH_FIRST yet
+    start_over = False
+
     if start_over:
       orientation = TFetchOrientation.FETCH_FIRST
     else:
@@ -475,12 +486,19 @@ class HiveServerClientCompatible:
 
 
   def get_log(self, handle):
-    operationHandle = handle.get_rpc_handle()
-    return self._client.get_log(operationHandle)
+    if self.query_server['server_name'] == 'impala':
+      return 'Impala does not support GetLog()'
+    else:
+      operationHandle = handle.get_rpc_handle()
+      return self._client.get_log(operationHandle)
 
 
   def get_databases(self):
-    return [table['TABLE_SCHEMA'] for table in self._client.get_databases()]
+    if self.query_server['server_name'] == 'impala':
+      col = 'TABLE_SCHEM'
+    else:
+      col = 'TABLE_SCHEMA'
+    return [table[col] for table in self._client.get_databases()]
 
 
   def get_tables(self, database, table_names):

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

@@ -399,8 +399,10 @@
    # Host of the Impala Server
    ## server_host=localhost
 
-   # Port of the Impala Server
+   # Port of the Impala Server when using Beeswax Thrift interface
    ## server_port=21000
+   # Port of the Impala Server when using  Hive Server 2 Thrift interface
+   ## server_port=21050
 
    # Kerberos principal
    ## impala_principal=impala/hostname.foo.com

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

@@ -402,8 +402,10 @@
    # Host of the Impala Server
    ## server_host=localhost
 
-   # Port of the Impala Server
+   # Port of the Impala Server when using Beeswax Thrift interface
    ## server_port=21000
+   # Port of the Impala Server when using  Hive Server 2 Thrift interface
+   ## server_port=21050
 
    # Kerberos principal
    ## impala_principal=impala/hostname.foo.com