Преглед изворни кода

HUE-6132 [metadata] Update the test and add a series of risk queries

Romain Rigaux пре 8 година
родитељ
комит
745e52b9a4
1 измењених фајлова са 194 додато и 97 уклоњено
  1. 194 97
      desktop/libs/metadata/src/metadata/optimizer_client_tests.py

+ 194 - 97
desktop/libs/metadata/src/metadata/optimizer_client_tests.py

@@ -16,6 +16,7 @@
 # limitations under the License.
 
 import logging
+import time
 
 from nose.plugins.skip import SkipTest
 from nose.tools import assert_equal, assert_true
@@ -25,20 +26,20 @@ from django.contrib.auth.models import User
 from desktop.auth.backend import rewrite_user
 from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.test_utils import add_to_group, grant_access
-from hadoop.pseudo_hdfs4 import is_live_cluster
 
 from metadata.optimizer_client import OptimizerApi
-from metadata.conf import has_optimizer
+from metadata.conf import OPTIMIZER, has_optimizer
+from desktop.models import uuid_default
 
 
 LOG = logging.getLogger(__name__)
 
 
-class TestOptimizerApi(object):
+class BaseTestOptimizerApi(object):
 
   @classmethod
   def setup_class(cls):
-    if not is_live_cluster() or not has_optimizer():
+    if not has_optimizer():
       raise SkipTest
 
     cls.client = make_logged_in_client(username='test', is_superuser=False)
@@ -57,86 +58,51 @@ class TestOptimizerApi(object):
     cls.user.save()
 
 
-  def test_tenant(self):
-    resp = self.api.get_tenant(email='romain@cloudera.com')
-
-    assert_equal('success', resp['status'], resp)
-
+class TestOptimizerApi(BaseTestOptimizerApi):
 
-  def test_create_tenant(self):
-    resp = self.api.create_tenant(group='hue')
+  def test_tenant(self):
+    resp = self.api.get_tenant(email=OPTIMIZER.EMAIL.get())
 
-    assert_equal('success', resp['status'], resp)
     assert_true('tenant' in resp, resp)
 
-
-  def test_authenticate(self):
-    resp = self.api.authenticate()
-
-    assert_true(resp['token'], resp)
-    assert_equal('success', resp['status'], resp)
-
-
-  def test_get_status(self):
-    resp = self.api.authenticate()
-    token = resp['token']
-
-    resp = self.api.get_status(token=token)
-
-    assert_equal('success', resp['status'], resp)
-    assert_true('filesFinished' in resp['details'], resp)
-    assert_true('filesProcessing' in resp['details'], resp)
-    assert_true('finished' in resp['details'], resp)
-
-
-  def test_delete_workload(self):
-    resp = self.api.authenticate()
-    token = resp['token']
-
-    resp = self.api.delete_workload(token=token)
-
-    assert_equal('success', resp['status'], resp)
-
-
+  # Should run first
   def test_upload(self):
     queries = [
-        "select emps.id from emps where emps.name = 'Joe' group by emps.mgr, emps.id;",
-        "select emps.name from emps where emps.num = 007 group by emps.state, emps.name;",
-        "select Part.partkey, Part.name, Part.type from db1.Part where Part.yyprice > 2095",
-        "select Part.partkey, Part.name, Part.mfgr FROM Part WHERE Part.name LIKE '%red';",
-        "select count(*) as loans from account a where a.account_state_id in (5,9);",
-        "select orders.key, orders.id from orders where orders.price < 9999",
-        "select mgr.name from mgr where mgr.reports > 10 group by mgr.state;"
+        (uuid_default(), 0, "select emps.id from emps where emps.name = 'Joe' group by emps.mgr, emps.id;", 'default'),
+        (uuid_default(), 0, "select emps.name from emps where emps.num = 007 group by emps.state, emps.name;", 'default'),
+        (uuid_default(), 0, "select Part.partkey, max(Part.salary), Part.name, Part.type from db1.Part where Part.yyprice > 2095", 'db1'),
+        (uuid_default(), 0, "elect Part.partkey, Part.name, Part.mfgr FROM Part WHERE Part.name LIKE '%red';", 'default'),
+        (uuid_default(), 0, "select count(*) as loans from account a where a.account_state_id in (5,9);", 'default'),
+        (uuid_default(), 0, "elect orders.key, orders.id from orders where orders.price < 9999", 'default'),
+        (uuid_default(), 0, "select mgr.name from mgr where mgr.reports > 10 group by mgr.state;", 'default'),
     ]
 
-    resp = self.api.upload(data=queries, data_type='queries')
+    resp = self.api.upload(data=queries, data_type='queries', source_platform='hive')
 
-    assert_equal('status' in resp, resp)
-    assert_equal('state' in resp['status'], resp)
-    assert_equal('workloadId' in resp['status'], resp)
+    assert_true('status' in resp, resp)
+    assert_true('count' in resp, resp)
 
+    assert_true('state' in resp['status'], resp)
+    assert_true('workloadId' in resp['status'], resp)
+    assert_true('failedQueries' in resp['status'], resp)
+    assert_true('successQueries' in resp['status'], resp)
     assert_true(resp['status']['state'] in ('WAITING', 'FINISHED', 'FAILED'), resp['status']['state'])
 
-    resp = self.api.upload_status(workfload_id=resp['status']['workloadId'])
-    assert_equal('status' in resp, resp)
-    assert_equal('state' in resp['status'], resp)
-    assert_equal('workloadId' in resp['status'], resp)
-
-
-  def test_upload_table_stats(self):
-    stats = [
-        "TABLE_NAME,NUM_ROWS",
-        "TEST_TABLE,10",
-        "TEST1,110",
-    ]
+    resp = self.api.upload_status(workload_id=resp['status']['workloadId'])
+    assert_true('status' in resp, resp)
+    assert_true('state' in resp['status'], resp)
+    assert_true('workloadId' in resp['status'], resp)
 
-    resp = self.api.upload(data=stats, data_type='table_stats')
 
-    assert_equal('status' in resp, resp)
-    assert_equal('state' in resp['status'], resp)
-    assert_equal('workloadId' in resp['status'], resp)
+    i = 0
+    while i < 60 and resp['status']['state'] not in ('FINISHED', 'FAILED'):
+      resp = self.api.upload_status(workload_id=resp['status']['workloadId'])
+      i += 1
+      time.sleep(1)
+      LOG.info('Upload state: %(state)s' % resp['status'])
 
-    assert_true(resp['status']['state'] in ('WAITING', 'FINISHED', 'FAILED'), resp['status']['state'])
+    assert_true(i < 60)
+    LOG.info('Final Upload state: %(state)s' % resp['status'])
 
 
   def test_top_tables(self):
@@ -148,11 +114,15 @@ class TestOptimizerApi(object):
     assert_true('eid' in resp['results'][0], resp)
     assert_true('name' in resp['results'][0], resp)
 
+    database_name = 'hue'
+    resp = self.api.top_tables(database_name=database_name)
+
+    assert_true(isinstance(resp['results'], list), resp)
+
 
   def test_table_details(self):  # Requires test_upload to run before
     resp = self.api.table_details(database_name='default', table_name='emps')
 
-    assert_equal('success', resp['status'], resp)
     assert_true('columnCount' in resp, resp)
     assert_true('createCount' in resp, resp)
     assert_true('table_ddl' in resp, resp)
@@ -172,27 +142,19 @@ class TestOptimizerApi(object):
 
     resp = self.api.table_details(database_name='db1', table_name='Part')
 
-    assert_equal('success', resp['status'], resp)
     assert_true('tid' in resp, resp)
     assert_true('columnCount' in resp, resp)
 
 
   def test_query_risk(self):
-    query = 'Select * from (Select item.id from item)'
+    query = 'Select * from items'
 
     resp = self.api.query_risk(query=query, source_platform='hive')
 
-    assert_equal('success', resp['status'], resp)
-
-    assert_true('impalaRisk' in resp, resp)
-    assert_true('riskAnalysis' in resp['impalaRisk'], resp)
-    assert_true('risk' in resp['impalaRisk'], resp)
-    assert_true('riskRecommendation' in resp['impalaRisk'], resp)
-
-    assert_true('hiveRisk' in resp, resp)
-    assert_true('riskAnalysis' in resp['hiveRisk'], resp)
-    assert_true('risk' in resp['hiveRisk'], resp)
-    assert_true('riskRecommendation' in resp['hiveRisk'], resp)
+    assert_true(len(resp) > 0, resp)
+    assert_true('riskAnalysis' in resp[0], resp)
+    assert_true('risk' in resp[0], resp)
+    assert_true('riskRecommendation' in resp[0], resp)
 
 
   def test_query_compatibility(self):
@@ -202,8 +164,6 @@ class TestOptimizerApi(object):
 
     resp = self.api.query_compatibility(source_platform=source_platform, target_platform=target_platform, query=query)
 
-    assert_equal('success', resp['status'], resp)
-
     assert_true('clauseName' in resp, resp)
     assert_true('clauseError' in resp, resp)
     assert_true('queryError' in resp, resp)
@@ -213,15 +173,14 @@ class TestOptimizerApi(object):
   def test_top_filters(self):  # Requires test_upload to run before
     resp = self.api.top_filters(db_tables='db1.Part')
 
-    assert_equal('success', resp['status'], resp)
-    assert_true('qids' in resp, resp)
-    assert_true('popularValues' in resp, resp)
+    assert_true(len(resp['results']) > 0, resp)
 
 
   def test_top_joins(self):
     resp = self.api.top_joins(db_tables='db1.Part')
 
-    assert_equal('success', resp['status'], resp)
+    assert_true(len(resp['results']) > 0, resp)
+
     assert_true('tables' in resp['results'][0], resp)
     assert_true('queryIds' in resp['results'][0], resp)
     assert_true('totalTableCount' in resp['results'][0], resp)
@@ -233,7 +192,8 @@ class TestOptimizerApi(object):
   def test_top_aggs(self):
     resp = self.api.top_aggs(db_tables='db1.Part')
 
-    assert_equal('success', resp['status'], resp)
+    assert_true(len(resp['results']) > 0, resp)
+
     assert_true('tables' in resp['results'][0], resp)
     assert_true('queryIds' in resp['results'][0], resp)
     assert_true('totalTableCount' in resp['results'][0], resp)
@@ -245,16 +205,23 @@ class TestOptimizerApi(object):
   def test_top_columns(self):
     resp = self.api.top_columns(db_tables='db1.Part')
 
-    assert_equal('success', resp['status'], resp)
-    assert_true('tables' in resp['results'][0], resp)
+    assert_true(len(resp.get('results', '')) > 0, resp)
+
+    assert_true('orderbyColumns' in resp, resp)
+    assert_true('selectColumns' in resp, resp)
+    assert_true('filterColumns' in resp, resp)
+    assert_true('joinColumns' in resp, resp)
+    assert_true('groupbyColumns' in resp, resp)
+    assert_true('groupbyColumns' in resp, resp)
 
 
   def test_top_databases(self):
     resp = self.api.top_databases()
 
-    assert_equal('success', resp['status'], resp)
-    assert_true('instanceCount' in resp['results'], resp)
-    assert_true('totalTableCount' in resp['results'], resp)
+    assert_true(len(resp['results']) > 0, resp)
+
+    assert_true('instanceCount' in resp['results'][0], resp)
+    assert_true('totalTableCount' in resp['results'][0], resp)
 
 
   def test_similar_queries(self):
@@ -263,7 +230,137 @@ class TestOptimizerApi(object):
 
     resp = self.api.similar_queries(source_platform=source_platform, query=query)
 
-    assert_equal('successs', resp['status'], resp)
-
     assert_true('querySignature' in resp, resp)
     assert_true('query' in resp, resp)
+
+
+
+class TestOptimizerRiskApi(BaseTestOptimizerApi):
+
+  def test_risk_10_views(self):
+    source_platform = 'hive'
+    query = '''SELECT code
+FROM
+  (SELECT code
+   FROM
+     (SELECT code
+      FROM
+        (SELECT code
+         FROM
+           (SELECT code
+            FROM
+              (SELECT code
+               FROM
+                 (SELECT code
+                  FROM
+                    (SELECT code
+                     FROM
+                       (SELECT code
+                        FROM
+                          (SELECT code
+                           FROM
+                             (SELECT code
+                              FROM
+                                (SELECT code
+                                 FROM
+                                   (SELECT code
+                                    FROM sample_01) t1) t2) t3) t4) t5) t6) t7) t8) t9) t10) t11) t12
+'''
+
+    resp = self.api.query_risk(query=query, source_platform=source_platform)
+    _assert_risks(['>=10 Inline Views present in query.'], resp)
+
+
+  def test_risk_cartesian_cross_join(self):
+    source_platform = 'hive'
+    query = '''SELECT s07.description, s07.total_emp, s08.total_emp, s07.salary
+FROM
+  sample_07 s07
+cross
+JOIN
+  sample_08 s08
+ON ( s07.code = s08.code )
+WHERE
+( s07.total_emp > s08.total_emp
+ AND s07.salary > 100000 )
+ORDER BY s07.salary DESC
+'''
+
+    resp = self.api.query_risk(query=query, source_platform=source_platform)
+    _assert_risks(['Cartesian or CROSS join found.'], resp)
+
+
+    source_platform = 'hive'
+    query = '''SELECT ID, NAME, AMOUNT, DATE FROM CUSTOMERS, ORDERS
+'''
+
+    resp = self.api.query_risk(query=query, source_platform=source_platform)
+    _assert_risks(['Cartesian or CROSS join found.'], resp)
+
+
+  def test_risk_5_joins(self):
+    source_platform = 'hive'
+    query = '''SELECT s07.description, s07.total_emp, s08.total_emp, s07.salary
+FROM
+  sample_07 s07
+JOIN
+  sample_08 s08
+ON ( s07.code = s08.code )
+JOIN
+  sample_06 s06
+ON ( s07.code = s06.code )
+JOIN
+  sample_05 s05
+ON ( s07.code = s05.code )
+JOIN
+  sample_04 s04
+ON ( s07.code = s04.code )
+JOIN
+  sample_03 s03
+ON ( s07.code = s03.code )
+JOIN
+  sample_02 s02
+ON ( s07.code = s02.code )
+JOIN
+  sample_01 s01
+ON ( s07.code = s01.code )
+
+WHERE
+( s07.total_emp > s08.total_emp
+ AND s07.salary > 100000 )
+ORDER BY s07.salary DESC
+LIMIT 1000
+'''
+
+    resp = self.api.query_risk(query=query, source_platform=source_platform)
+    _assert_risks(['>=5 table joins or >=10 join conditions found.'], resp)
+
+
+  def test_risk_10_group_by_columns(self):
+
+    source_platform = 'impala'
+    query = '''SELECT *
+FROM transactions
+GROUP BY account_client,
+         account_cty_code,
+         account_num,
+         allow_code,
+         ally_811,
+         anti_detect,
+         anti_transcode,
+         cc_fee,
+         auth_code,
+         cvv_eval,
+         cred_extract, denied_code
+         limit 5
+'''
+
+    resp = self.api.query_risk(query=query, source_platform=source_platform)
+    _assert_risks(['>=10 columns present in GROUP BY list.'], resp)
+
+
+def _assert_risks(risks, suggestions):
+  suggestion_names = [suggestion['riskAnalysis'] for suggestion in suggestions]
+
+  for risk in risks:
+    assert_true(risk in suggestion_names, suggestions)