Browse Source

HUE-7955 [importer] Backend clean-up of the creation of an index via a live query

Romain Rigaux 7 years ago
parent
commit
59363ee

+ 16 - 22
desktop/libs/indexer/src/indexer/api3.py

@@ -28,7 +28,7 @@ from desktop.models import Document2
 from librdbms.server import dbms as rdbms
 from notebook.connectors.base import get_api, Notebook
 from notebook.decorators import api_error_handler
-from notebook.models import make_notebook
+from notebook.models import make_notebook, MockedDjangoRequest
 
 from indexer.controller import CollectionManagerController
 from indexer.file_format import HiveFormat
@@ -156,7 +156,7 @@ def guess_field_types(request):
       columns = file_format.get('sampleCols')
       sample = file_format.get('sample')
     else:
-      snippet['query'] = snippet['statement'] #self._get_current_statement(db, snippet) # TODO multi statement
+      snippet['query'] = snippet['statement']
       try:
         sample = db.fetch_result(notebook, snippet, 4, start_over=True)['rows'][:4]
       except Exception, e:
@@ -170,20 +170,7 @@ def guess_field_types(request):
     format_ = {
         "sample": sample,
         "columns": columns,
-        "hs2_handle": None # HS2 there and valid? add sample
     }
-#     else:
-#       format_ = {'status': 3}
-#       sample = db.fetch_result(notebook, snippet, 4, start_over=True)
-# 
-#       format_ = {
-#           "sample": sample['rows'][:4],
-#           #"sample_cols": sample.meta,
-#           "columns": [
-#               Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
-#               for col in sample.meta
-#           ]
-#       }
   elif file_format['inputFormat'] == 'rdbms':
     query_server = rdbms.get_query_server_config(server=file_format['rdbmsType'])
     db = rdbms.get(request.user, query_server=query_server)
@@ -290,16 +277,23 @@ def _small_indexing(user, fs, client, source, destination, index_name):
   if source['inputFormat'] == 'file':
     data = fs.read(source["path"], 0, MAX_UPLOAD_SIZE)
 
+  if client.is_solr_six_or_more():
+    kwargs['processor'] = 'tolerant'
+
   try:
     if source['inputFormat'] == 'query':
-      #   elif file_format['inputFormat'] == 'hs2_handle':
+      query_id = source['query']['id'] if source['query'].get('id') else source['query']
+
+      notebook = Notebook(document=Document2.objects.document(user=user, doc_id=query_id)).get_data()
+      request = MockedDjangoRequest(user=user)
+      snippet = notebook['snippets'][0]
+
       searcher = CollectionManagerController(user)
-      columns = fields#['_uuid'] + [field['name'] for field in file_format['columns']]
-      return searcher.update_data_from_hive(index_name, columns, fetch_handle=file_format['fetch_handle'])
-      ## live HS2
-    else:      
-      if client.is_solr_six_or_more():
-        kwargs['processor'] = 'tolerant'
+      columns = [field['name'] for field in fields if field['name'] != 'hue_id']
+      fetch_handle = lambda rows, start_over: get_api(request, snippet).fetch_result(notebook, snippet, rows=rows, start_over=start_over) # Assumes handle still live
+      rows = searcher.update_data_from_hive(index_name, columns, fetch_handle=fetch_handle, indexing_options=kwargs)
+      # TODO if rows == MAX_ROWS truncation warning
+    else:
       response = client.index(name=index_name, data=data, **kwargs)
       errors = [error.get('message', '') for error in response['responseHeader'].get('errors', [])]
   except Exception, e:

+ 6 - 5
desktop/libs/indexer/src/indexer/controller.py

@@ -268,12 +268,14 @@ class CollectionManagerController(object):
     else:
       raise PopupException(_('Could not update index. Indexing strategy %s not supported.') % indexing_strategy)
 
-  def update_data_from_hive(self, collection_or_core_name, columns, fetch_handle):
+  def update_data_from_hive(self, collection_or_core_name, columns, fetch_handle, indexing_options=None):
     MAX_ROWS = 10000
     FETCH_BATCH = 1000
 
     row_count = 0
     has_more = True
+    if indexing_options is None:
+      indexing_options = {}
 
     client = SolrClient(self.user)
 
@@ -283,14 +285,13 @@ class CollectionManagerController(object):
         has_more = result['has_more']
 
         if result['data']:
-          kwargs = {}
           dataset = tablib.Dataset()
           dataset.append(columns)
           for i, row in enumerate(result['data']):
-            dataset.append([row_count + i] + [cell if cell else (0 if isinstance(cell, numbers.Number) else '') for cell in row])
+            dataset.append([cell if cell else (0 if isinstance(cell, numbers.Number) else '') for cell in row])
 
-          if not client.index(name=collection_or_core_name, data=dataset.csv, **kwargs):
-            raise PopupException(_('Could not update index. Check error logs for more info.'))
+          if not client.index(name=collection_or_core_name, data=dataset.csv, **indexing_options):
+            raise PopupException(_('Could not index the data. Check error logs for more info.'))
 
         row_count += len(dataset)
     except Exception, e:

+ 2 - 2
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -367,7 +367,7 @@ ${ assist.assistPanel() }
       <!-- /ko -->
       <!-- ko ifnot: createWizard.isGuessingFieldTypes -->
       <h4>${_('Preview')}</h4>
-      <!-- ko if: createWizard.source.inputFormat() == 'query' -->
+      <!-- ko if: createWizard.source.inputFormat() == 'query' && createWizard.source.sample().length == 0 -->
         ${ _('Add sample data') } <i class="fa fa-fw fa-play"></i>
       <!-- /ko -->
       <div class="card-body">
@@ -582,7 +582,7 @@ ${ assist.assistPanel() }
           <div class="card-body">
             % if ENABLE_SCALABLE_INDEXER.get():
             <div class="control-group">
-              <label class="checkbox inline-block" title="${ _('Execute a cluster job to index a large dataset.') }" data-bind="visible: $root.createWizard.source.inputFormat() != 'manual'">
+              <label class="checkbox inline-block" title="${ _('Execute a cluster job to index a large dataset.') }" data-bind="visible: $root.createWizard.source.inputFormat() == 'file'">
                 <input type="checkbox" data-bind="checked: indexerRunJob"> ${_('Index with a job')}
               </label>
 

+ 15 - 54
desktop/libs/notebook/src/notebook/api.py

@@ -29,6 +29,8 @@ from desktop.api2 import __paginate
 from desktop.lib.i18n import smart_str
 from desktop.lib.django_util import JsonResponse
 from desktop.models import Document2, Document
+from indexer.file_format import HiveFormat
+from indexer.fields import Field
 
 from notebook.connectors.base import get_api, Notebook, QueryExpired, SessionExpired, QueryError, _get_snippet_name
 from notebook.connectors.dataeng import DataEngApi
@@ -704,68 +706,27 @@ def export_result(request):
       'allowed': True
     }
   elif data_format == 'search-index':
-    # Open the result in the Dashboard via a SQL sub-query or the Import wizard (and its quick  or scalable indexer)
-
+    # Open the result in the Dashboard via a SQL sub-query or the Import wizard (quick vs scalable)
     if is_embedded:
       notebook_id = notebook['id'] or request.GET.get('editor', request.GET.get('notebook'))
 
-      if destination == '__hue__':        
+      if destination == '__hue__':
         engine = notebook['type'].replace('query-', '')
         response['watch_url'] = reverse('dashboard:browse', kwargs={'name': notebook_id}) + '?source=query&engine=%(engine)s' % {'engine': engine}
         response['status'] = 0
-        return JsonResponse(response) # Currently do not live index into Solr, but uses a SQL sub-query
-
-        destination = _get_snippet_name(notebook, unique=True, table_format=True)
-#         live_indexing = True
       else:
-        live_indexing = False
-
-
-      ## TODO
-      ## Add to response query id and sample data
-
-      sample = get_api(request, snippet).fetch_result(notebook, snippet, rows=4, start_over=True)
-
-#       from indexer.api3 import _index # Will be moved to the lib
-      from indexer.file_format import HiveFormat
-      from indexer.fields import Field
-      format_ = {
-          "sample": list(sample['data']),
-          "columns": [
-              Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
-              for col in sample['meta']
-          ]
-      }
-
-      return JsonResponse({
-        'status': 0,
-        'id': notebook_id,
-        'source_type': 'query',
-        'target_type': 'index',
-        'target_path': destination,
-        'sample': format_['sample'],
-        'columns': format_['columns']
-      })
+        sample = get_api(request, snippet).fetch_result(notebook, snippet, rows=4, start_over=True)
 
-#       file_format = {
-#           'name': 'col',
-#           'inputFormat': 'query',
-#           'format': {'quoteChar': '"', 'recordSeparator': '\n', 'type': 'csv', 'hasHeader': False, 'fieldSeparator': '\u0001'},
-#           "sample": '',
-#           "columns": [
-#               Field(col['name'].rsplit('.')[-1], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
-#               for col in sample['meta']
-#           ]
-#       }
-
-#       if live_indexing:
-#         file_format['inputFormat'] = 'hs2_handle'
-#         file_format['fetch_handle'] = lambda rows, start_over: get_api(request, snippet).fetch_result(notebook, snippet, rows=rows, start_over=start_over)
-#         response['rowcount'] = _index(request, file_format, destination, query=notebook['uuid'], start_time=start_time)
-#         response['watch_url'] = reverse('search:browse', kwargs={'name': destination})
-#         response['status'] = 0
-#       else:
-#         response = _index(request, file_format, destination, query=notebook['uuid'], start_time=start_time)
+        response['status'] = 0
+        response['id'] = notebook_id
+        response['source_type'] = 'query'
+        response['target_type'] = 'index'
+        response['target_path'] = destination
+        response['sample'] = list(sample['data'])
+        response['columns'] = [
+            Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
+            for col in sample['meta']
+        ]
     else:
       notebook_id = notebook['id'] or request.GET.get('editor', request.GET.get('notebook'))
       response['watch_url'] = reverse('notebook:execute_and_watch') + '?action=index_query&notebook=' + str(notebook_id) + '&snippet=0&destination=' + destination