Эх сурвалжийг харах

HUE-3078 [impala] Implement Impala Invalidate Metadata and Refresh

Provides 2 endpoints to update Impala's metadata:

POST /impala/api/invalidate/<database>

Calls INVALIDATE METADATA on the given database, but by default checks HMS for any new or removed tables, and then calls INVALIDATE on each table.

If flush_all=true, then a full INVALIDATE METADATA on the entire DB will be called.

On success, returns:
{
	"status": 0,
	"message": "Successfully invalidated metadata for `default`"
}

POST /impala/api/refresh/<database>/<table>

Calls REFRESH `<database>`.`<table>` which updates the metadata for the given table.

On success, returns:
{
	"status": 0,
	"message": "Successfully refreshed metadata for `default`.`sample_08`"
}
Jenny Kim 10 жил өмнө
parent
commit
638b4d3ebc

+ 29 - 11
apps/impala/src/impala/api.py

@@ -18,9 +18,10 @@
 
 ## Main views are inherited from Beeswax.
 
-
 import logging
-import json
+
+from django.utils.translation import ugettext as _
+from django.views.decorators.http import require_POST
 
 from desktop.lib.django_util import JsonResponse
 
@@ -32,20 +33,37 @@ from impala import dbms
 LOG = logging.getLogger(__name__)
 
 
-def refresh_tables(request):
+@require_POST
+def invalidate(request, database):
   query_server = dbms.get_query_server_config()
   db = beeswax_dbms.get(request.user, query_server=query_server)
 
   response = {'status': 0, 'message': ''}
 
-  if request.method == "POST":
-    try:
-      database = json.loads(request.POST['database'])
-      added = json.loads(request.POST.get('added', []))
-      removed = json.loads(request.POST.get('removed', []))
+  try:
+    flush_all = request.POST.get('flush_all', 'false').lower() == 'true'
+    db.invalidate(database, flush_all=flush_all)
+    response['message'] = _('Successfully invalidated metadata for `%s`') % database
+  except Exception, e:
+    response['status'] = -1
+    response['message'] = _(str(e))
+
+  return JsonResponse(response)
+
+
+
+@require_POST
+def refresh_table(request, database, table):
+  query_server = dbms.get_query_server_config()
+  db = beeswax_dbms.get(request.user, query_server=query_server)
+
+  response = {'status': 0, 'message': ''}
 
-      db.invalidate_tables(database, added + removed)
-    except Exception, e:
-      response['message'] = str(e)
+  try:
+    db.refresh_table(database, table)
+    response['message'] = _('Successfully refreshed metadata for `%s`.`%s`') % (database, table)
+  except Exception, e:
+    response['status'] = -1
+    response['message'] = _(str(e))
 
   return JsonResponse(response)

+ 46 - 14
apps/impala/src/impala/dbms.py

@@ -22,7 +22,9 @@ from desktop.lib.i18n import smart_str
 from beeswax.conf import BROWSE_PARTITIONED_TABLE_LIMIT
 from beeswax.design import hql_query
 from beeswax.models import QUERY_TYPES
-from beeswax.server.dbms import HiveServer2Dbms
+from beeswax.server import dbms
+from beeswax.server.dbms import HiveServer2Dbms, QueryServerException, QueryServerTimeoutException,\
+  get_query_server_config as beeswax_query_server_config
 
 from impala import conf
 
@@ -86,23 +88,42 @@ class ImpalaDbms(HiveServer2Dbms):
     return 'SELECT histogram(%s) FROM %s' % (select_clause, from_clause)
 
 
-  def invalidate_tables(self, database, tables=None):
+  def invalidate(self, database, flush_all=False):
     handle = None
-
     try:
-      if tables:
-        for table in tables:
-          hql = "INVALIDATE METADATA `%s`.`%s`" % (database, table,)
-          print hql
-          query = hql_query(hql, database, query_type=QUERY_TYPES[1])
-          handle = self.execute_and_wait(query, timeout_sec=10.0)
-      else:  # call INVALIDATE on entire DB to pick up newly created tables
-        hql = "INVALIDATE METADATA `%s`" % database
-        print hql
-        query = hql_query(hql, database, query_type=QUERY_TYPES[1])
+      if flush_all:
+        self.use(database)  # INVALIDATE does not accept database as a single parameter
+        hql = "INVALIDATE METADATA"
+        query = hql_query(hql, query_type=QUERY_TYPES[1])
         handle = self.execute_and_wait(query, timeout_sec=10.0)
+      else:
+        diff_tables = self._get_different_tables(database)
+        for table in diff_tables:
+          hql = "INVALIDATE METADATA `%s`.`%s`" % (database, table)
+          query = hql_query(hql, query_type=QUERY_TYPES[1])
+          handle = self.execute_and_wait(query, timeout_sec=10.0)
+    except QueryServerTimeoutException, e:
+      # Allow timeout exceptions to propagate
+      raise e
+    except Exception, e:
+      LOG.error('Failed to invalidate `%s`: %s' % (database, smart_str(e)))
+      msg = 'Failed to invalidate `%s`' % database
+      raise QueryServerException(msg)
+    finally:
+      if handle:
+        self.close(handle)
+
+
+  def refresh_table(self, database, table):
+    handle = None
+    try:
+      hql = "REFRESH `%s`.`%s`" % (database, table)
+      query = hql_query(hql, database, query_type=QUERY_TYPES[1])
+      handle = self.execute_and_wait(query, timeout_sec=10.0)
     except Exception, e:
-      LOG.warn('Refresh tables cache out of sync: %s' % smart_str(e))
+      LOG.error('Failed to refresh `%s`.`%s`: %s' % (database, table, smart_str(e)))
+      msg = 'Failed to refresh `%s`.`%s`' % (database, table)
+      raise QueryServerException(msg)
     finally:
       if handle:
         self.close(handle)
@@ -159,3 +180,14 @@ class ImpalaDbms(HiveServer2Dbms):
         self.close(handle)
 
     return results
+
+
+  def _get_beeswax_tables(self, database):
+    beeswax_query_server = dbms.get(user=self.client.user, query_server=beeswax_query_server_config(name='beeswax'))
+    return beeswax_query_server.get_tables(database=database)
+
+
+  def _get_different_tables(self, database):
+    beeswax_tables = self._get_beeswax_tables(database)
+    impala_tables = self.get_tables(database=database)
+    return set(beeswax_tables).symmetric_difference(impala_tables)

+ 65 - 2
apps/impala/src/impala/tests.py

@@ -119,7 +119,7 @@ class TestImpalaIntegration:
     hql = """
       USE default;
       DROP TABLE IF EXISTS %(db)s.tweets;
-      DROP DATABASE IF EXISTS %(db)s;
+      DROP DATABASE IF EXISTS %(db)s CASCADE;
       CREATE DATABASE %(db)s;
 
       USE %(db)s;
@@ -153,7 +153,7 @@ class TestImpalaIntegration:
     hql = """
     USE default;
     DROP TABLE IF EXISTS %(db)s.tweets;
-    DROP DATABASE %(db)s;
+    DROP DATABASE %(db)s CASCADE;
     """ % {'db': cls.DATABASE}
     resp = _make_query(cls.client, hql, database='default', local=False, server_name='impala')
     resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
@@ -231,6 +231,69 @@ class TestImpalaIntegration:
     assert_true(data['properties'].get('http_addr'))
 
 
+  def test_invalidate(self):
+    # Helper function to get Impala and Beeswax (HMS) tables
+    def get_impala_beeswax_tables():
+      impala_resp = self.client.get(reverse('impala:api_autocomplete_tables', kwargs={'database': self.DATABASE}))
+      impala_tables_meta = json.loads(impala_resp.content)['tables_meta']
+      impala_tables = [table['name'] for table in impala_tables_meta]
+      beeswax_resp = self.client.get(reverse('beeswax:api_autocomplete_tables', kwargs={'database': self.DATABASE}))
+      beeswax_tables_meta = json.loads(beeswax_resp.content)['tables_meta']
+      beeswax_tables = [table['name'] for table in beeswax_tables_meta]
+      return impala_tables, beeswax_tables
+
+    impala_tables, beeswax_tables = get_impala_beeswax_tables()
+    assert_equal(impala_tables, beeswax_tables,
+      "\ntest_invalidate: `%s`\nImpala Tables: %s\nBeeswax Tables: %s" % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
+
+    hql = """
+      CREATE TABLE new_table (a INT);
+    """
+    resp = _make_query(self.client, hql, wait=True, local=False, max=180.0, database=self.DATABASE)
+
+    impala_tables, beeswax_tables = get_impala_beeswax_tables()
+    # New table is not found by Impala
+    assert_true('new_table' in beeswax_tables, beeswax_tables)
+    assert_false('new_table' in impala_tables, impala_tables)
+
+    resp = self.client.post(reverse('impala:invalidate', kwargs={'database': self.DATABASE}))
+
+    impala_tables, beeswax_tables = get_impala_beeswax_tables()
+    # Invalidate picks up new table
+    assert_equal(impala_tables, beeswax_tables,
+      "\ntest_invalidate: `%s`\nImpala Tables: %s\nBeeswax Tables: %s" % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
+
+
+  def test_refresh_table(self):
+    # Helper function to get Impala and Beeswax (HMS) columns
+    def get_impala_beeswax_columns():
+      impala_resp = self.client.get(reverse('impala:api_autocomplete_columns', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
+      impala_columns = json.loads(impala_resp.content)['columns']
+      beeswax_resp = self.client.get(reverse('beeswax:api_autocomplete_columns', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
+      beeswax_columns = json.loads(beeswax_resp.content)['columns']
+      return impala_columns, beeswax_columns
+
+    impala_columns, beeswax_columns = get_impala_beeswax_columns()
+    assert_equal(impala_columns, beeswax_columns,
+      "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s" % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
+
+    hql = """
+      ALTER TABLE tweets ADD COLUMNS (new_column INT);
+    """
+    resp = _make_query(self.client, hql, wait=True, local=False, max=180.0, database=self.DATABASE)
+
+    impala_columns, beeswax_columns = get_impala_beeswax_columns()
+    # New column is not found by Impala
+    assert_true('new_column' in beeswax_columns, beeswax_columns)
+    assert_false('new_column' in impala_columns, impala_columns)
+
+    resp = self.client.post(reverse('impala:refresh_table', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
+
+    impala_columns, beeswax_columns = get_impala_beeswax_columns()
+    # Invalidate picks up new column
+    assert_equal(impala_columns, beeswax_columns,
+      "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s" % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
+
 
 # Could be refactored with SavedQuery.create_empty()
 def create_saved_query(app_name, owner):

+ 2 - 1
apps/impala/src/impala/urls.py

@@ -21,7 +21,8 @@ from beeswax.urls import urlpatterns as beeswax_urls
 
 
 urlpatterns = patterns('impala.api',
-  url(r'^api/refresh_tables$', 'refresh_tables', name='refresh_tables'),
+  url(r'^api/invalidate/(?P<database>\w+)$', 'invalidate', name='invalidate'),
+  url(r'^api/refresh/(?P<database>\w+)/(?P<table>\w+)$', 'refresh_table', name='refresh_table'),
 )
 
 urlpatterns += patterns('impala.dashboards',