瀏覽代碼

[hive] Support parameters in multi queries

Romain Rigaux 10 年之前
父節點
當前提交
76e3000

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

@@ -150,6 +150,7 @@ def parameters(request, design_id=None):
 def execute_directly(request, query, design, query_server, tablename=None, **kwargs):
   if design is not None:
     design = authorized_get_design(request, design.id)
+  parameters = kwargs.pop('parameters', None)
 
   db = dbms.get(request.user, query_server)
   database = query.query.get('database', 'default')
@@ -158,6 +159,10 @@ def execute_directly(request, query, design, query_server, tablename=None, **kwa
   history_obj = db.execute_query(query, design)
   watch_url = reverse(get_app_name(request) + ':api_watch_query_refresh_json', kwargs={'id': history_obj.id})
 
+  if parameters is not None:
+    history_obj.update_extra('parameters', parameters)
+    history_obj.save()
+
   response = {
     'status': 0,
     'id': history_obj.id,
@@ -298,7 +303,8 @@ def execute(request, design_id=None):
           parameterization_form = parameterization_form_cls(request.REQUEST, prefix="parameterization")
 
           if parameterization_form.is_valid():
-            real_query = substitute_variables(query_str, parameterization_form.cleaned_data)
+            parameters = parameterization_form.cleaned_data
+            real_query = substitute_variables(query_str, parameters)
             query = HQLdesign(query_form, query_type=query_type)
             query._data_dict['query']['query'] = real_query
 
@@ -306,7 +312,7 @@ def execute(request, design_id=None):
               if explain:
                 return explain_directly(request, query, design, query_server)
               else:
-                return execute_directly(request, query, design, query_server)
+                return execute_directly(request, query, design, query_server, parameters=parameters)
 
             except Exception, ex:
               db = dbms.get(request.user, query_server)

+ 114 - 0
apps/beeswax/src/beeswax/migrations/0012_auto__add_field_queryhistory_extra.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+from south.utils import datetime_utils as 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 'QueryHistory.extra'
+        db.add_column(u'beeswax_queryhistory', 'extra',
+                      self.gf('django.db.models.fields.TextField')(default='{}'),
+                      keep_default=False)
+
+
+    def backwards(self, orm):
+        # Deleting field 'QueryHistory.extra'
+        db.delete_column(u'beeswax_queryhistory', 'extra')
+
+
+    models = {
+        u'auth.group': {
+            'Meta': {'object_name': 'Group'},
+            u'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': u"orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        u'auth.permission': {
+            'Meta': {'ordering': "(u'content_type__app_label', u'content_type__model', u'codename')", 'unique_together': "((u'content_type', u'codename'),)", 'object_name': 'Permission'},
+            'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': u"orm['contenttypes.ContentType']"}),
+            u'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
+        },
+        u'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', [], {'symmetrical': 'False', 'related_name': "u'user_set'", 'blank': 'True', 'to': u"orm['auth.Group']"}),
+            u'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
+            'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            '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', [], {'symmetrical': 'False', 'related_name': "u'user_set'", 'blank': 'True', 'to': u"orm['auth.Permission']"}),
+            'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
+        },
+        u'beeswax.metainstall': {
+            'Meta': {'object_name': 'MetaInstall'},
+            u'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'installed_example': ('django.db.models.fields.BooleanField', [], {})
+        },
+        u'beeswax.queryhistory': {
+            'Meta': {'ordering': "['-submission_date']", 'object_name': 'QueryHistory'},
+            'design': ('django.db.models.fields.related.ForeignKey', [], {'to': u"orm['beeswax.SavedQuery']", 'null': 'True'}),
+            'extra': ('django.db.models.fields.TextField', [], {'default': "'{}'"}),
+            'has_results': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            u'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_redacted': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            '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'}),
+            'operation_type': ('django.db.models.fields.SmallIntegerField', [], {'null': 'True'}),
+            'owner': ('django.db.models.fields.related.ForeignKey', [], {'to': u"orm['auth.User']"}),
+            'query': ('django.db.models.fields.TextField', [], {}),
+            'query_type': ('django.db.models.fields.SmallIntegerField', [], {'default': '0'}),
+            '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.PositiveIntegerField', [], {'default': '10000'}),
+            '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'})
+        },
+        u'beeswax.savedquery': {
+            'Meta': {'ordering': "['-mtime']", 'object_name': 'SavedQuery'},
+            'data': ('django.db.models.fields.TextField', [], {'max_length': '65536'}),
+            'desc': ('django.db.models.fields.TextField', [], {'max_length': '1024'}),
+            u'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_auto': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'db_index': 'True'}),
+            'is_redacted': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_trashed': ('django.db.models.fields.BooleanField', [], {'default': 'False', 'db_index': 'True'}),
+            'mtime': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '80'}),
+            'owner': ('django.db.models.fields.related.ForeignKey', [], {'to': u"orm['auth.User']"}),
+            'type': ('django.db.models.fields.IntegerField', [], {})
+        },
+        u'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'"}),
+            u'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': u"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', [], {})
+        },
+        u'contenttypes.contenttype': {
+            'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
+            'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            u'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']

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

@@ -83,7 +83,7 @@ class QueryHistory(models.Model):
   notify = models.BooleanField(default=False)                        # Notify on completion
 
   is_redacted = models.BooleanField(default=False)
-
+  extra = models.TextField(default='{}')                   # Json fields for extra properties
 
   class Meta:
     ordering = ['-submission_date']
@@ -137,7 +137,7 @@ class QueryHistory(models.Model):
     query.hql_query = hql_query
     self.design.data = query.dumps()
     self.query = hql_query
- 
+
   def is_finished(self):
     is_statement_finished = not self.is_running()
 
@@ -188,6 +188,14 @@ class QueryHistory(models.Model):
 
     super(QueryHistory, self).save(*args, **kwargs)
 
+  def update_extra(self, key, val):
+    extra = json.loads(self.extra)
+    extra[key] = val
+    self.extra = json.dumps(extra)
+
+  def get_extra(self, key):
+    return json.loads(self.extra).get(key)
+
 
 def make_query_context(type, info):
   """

+ 10 - 5
apps/beeswax/src/beeswax/server/dbms.py

@@ -23,17 +23,17 @@ from django.core.urlresolvers import reverse
 from django.utils.encoding import force_unicode
 from django.utils.translation import ugettext as _
 
+from desktop.lib.django_util import format_preserving_redirect
+from desktop.lib.i18n import smart_str
+from jobsub.parameterization import substitute_variables
+from filebrowser.views import location_to_url
+
 from beeswax import hive_site
 from beeswax.conf import HIVE_SERVER_HOST, HIVE_SERVER_PORT, BROWSE_PARTITIONED_TABLE_LIMIT, SERVER_CONN_TIMEOUT
 from beeswax.design import hql_query
 from beeswax.hive_site import hiveserver2_use_ssl
 from beeswax.models import QueryHistory, QUERY_TYPES
 
-from filebrowser.views import location_to_url
-from desktop.lib.django_util import format_preserving_redirect
-from desktop.lib.i18n import smart_str
-
-
 
 LOG = logging.getLogger(__name__)
 
@@ -529,6 +529,11 @@ class HiveServer2Dbms(object):
     query_history.save()
     query = query_history.design.get_design()
 
+    # In case of multiquery, we need to re-replace the parameters as we save the non substituted query
+    if query._data_dict['query']['is_parameterized']:
+      real_query = substitute_variables(query._data_dict['query']['query'], query_history.get_extra('parameters'))
+      query._data_dict['query']['query'] = real_query
+
     return self.execute_and_watch(query, query_history=query_history)
 
 

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

@@ -311,9 +311,9 @@ def verify_history(client, fragment, design=None, reverse=False, server_name='be
   """
   resp = client.get('/%(server_name)s/query_history' % {'server_name': server_name})
   my_assert = reverse and assert_false or assert_true
-  my_assert(fragment in resp.content)
+  my_assert(fragment in resp.content, resp.content)
   if design:
-    my_assert(design in resp.content)
+    my_assert(design in resp.content, resp.content)
 
   if resp.context:
     try:

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

@@ -49,6 +49,7 @@ from desktop.lib.test_utils import grant_access, add_to_group
 from desktop.lib.security_util import get_localhost_name
 from hadoop.fs.hadoopfs import Hdfs
 from hadoop.pseudo_hdfs4 import is_live_cluster
+from jobsub.parameterization import substitute_variables
 
 import desktop.conf as desktop_conf
 
@@ -89,6 +90,8 @@ def _make_query(client, query, submission_type="Execute",
 
   # Should be in the history if it's submitted.
   if submission_type == 'Execute':
+    if is_parameterized and params:
+      query = substitute_variables(query, dict(params))
     fragment = collapse_whitespace(smart_str(escape(query[:20])))
     verify_history(client, fragment=fragment, server_name=server_name)
 
@@ -623,6 +626,30 @@ for x in sys.stdin:
     assert_true([0, u'0x0'] in content.get('results'), content)
 
 
+  def test_multiple_statements_with_params(self):
+    hql = """
+      select ${x} from test;
+      select ${y} from test;
+    """
+
+    resp = _make_query(self.client, hql, params=[('x', '1'), ('y', '2')], database=self.db_name)
+
+    # First statement
+    content = json.loads(resp.content)
+    watch_url = content['watch_url']
+    assert_equal('select ${x} from test', content.get('statement'), content)
+
+    resp = wait_for_query_to_finish(self.client, resp, max=30.0)
+    content = fetch_query_result_data(self.client, resp)
+
+    # Next statement
+    resp = self.client.post(watch_url, {'next': True, 'query-query': hql})
+    content = json.loads(resp.content)
+    assert_equal('select ${y} from test', content.get('statement'), content)
+
+    resp = wait_for_query_to_finish(self.client, resp, max=30.0)
+
+
   def test_multiple_statements_with_error(self):
     hql = """
       show tables;