Browse Source

combining the code for localfile and remote file for phoenix

ayush.goyal 4 years ago
parent
commit
18995d5705

+ 6 - 0
desktop/core/src/desktop/js/ko/components/assist/assistDbEntry.js

@@ -20,6 +20,7 @@ import * as ko from 'knockout';
 
 
 import huePubSub from 'utils/huePubSub';
 import huePubSub from 'utils/huePubSub';
 import sqlUtils from 'sql/sqlUtils';
 import sqlUtils from 'sql/sqlUtils';
+import { findEditorConnector } from 'config/hueConfig';
 
 
 const findNameInHierarchy = async (entry, searchCondition) => {
 const findNameInHierarchy = async (entry, searchCondition) => {
   while (entry && !searchCondition(entry)) {
   while (entry && !searchCondition(entry)) {
@@ -55,6 +56,11 @@ class AssistDbEntry {
     self.navigationSettings = navigationSettings;
     self.navigationSettings = navigationSettings;
 
 
     self.sourceType = assistDbNamespace.sourceType;
     self.sourceType = assistDbNamespace.sourceType;
+    self.phoenix_connector = findEditorConnector(connector => connector.dialect === 'phoenix');
+    self.importer_url =
+      self.phoenix_connector && self.sourceType === self.phoenix_connector.id
+        ? window.HUE_URLS.IMPORTER_CREATE_PHOENIX_TABLE
+        : window.HUE_URLS.IMPORTER_CREATE_TABLE;
 
 
     self.expandable = self.catalogEntry.hasPossibleChildren();
     self.expandable = self.catalogEntry.hasPossibleChildren();
 
 

+ 1 - 1
desktop/core/src/desktop/js/ko/components/assist/ko.assistDbPanel.js

@@ -365,7 +365,7 @@ const TEMPLATE =
       <span class="assist-tables-counter">(<span data-bind="text: filteredEntries().length"></span>)</span>
       <span class="assist-tables-counter">(<span data-bind="text: filteredEntries().length"></span>)</span>
       <!-- ko if: sourceType !== 'solr' && $component.showImporter() -->
       <!-- ko if: sourceType !== 'solr' && $component.showImporter() -->
       <!-- ko if: typeof databaseName !== 'undefined' -->
       <!-- ko if: typeof databaseName !== 'undefined' -->
-        <a class="inactive-action" data-bind="hueLink: window.HUE_URLS.IMPORTER_CREATE_TABLE + databaseName + '/?sourceType=' + sourceType + '&namespace=' + assistDbNamespace.namespace.id" title="${I18n(
+        <a class="inactive-action" data-bind="hueLink: importer_url + databaseName + '/?sourceType=' + sourceType + '&namespace=' + assistDbNamespace.namespace.id" title="${I18n(
           'Create table'
           'Create table'
         )}" href="javascript:void(0)">
         )}" href="javascript:void(0)">
           <i class="pointer fa fa-plus" title="${I18n('Create table')}"></i>
           <i class="pointer fa fa-plus" title="${I18n('Create table')}"></i>

+ 1 - 0
desktop/core/src/desktop/templates/global_js_constants.mako

@@ -139,6 +139,7 @@
 
 
   window.HUE_URLS = {
   window.HUE_URLS = {
     IMPORTER_CREATE_TABLE: '${ 'indexer' in apps and url('indexer:importer_prefill', source_type = 'all', target_type = 'table')}',
     IMPORTER_CREATE_TABLE: '${ 'indexer' in apps and url('indexer:importer_prefill', source_type = 'all', target_type = 'table')}',
+    IMPORTER_CREATE_PHOENIX_TABLE: '${ 'indexer' in apps and url('indexer:importer_prefill', source_type = 'all', target_type = 'big-table')}',
     IMPORTER_CREATE_DATABASE: '${ 'indexer' in apps and url('indexer:importer_prefill', source_type = 'manual', target_type = 'database')}',
     IMPORTER_CREATE_DATABASE: '${ 'indexer' in apps and url('indexer:importer_prefill', source_type = 'manual', target_type = 'database')}',
     NOTEBOOK_INDEX: '${url('notebook:index')}',
     NOTEBOOK_INDEX: '${url('notebook:index')}',
     % if 'pig' in apps:
     % if 'pig' in apps:

+ 10 - 6
desktop/libs/indexer/src/indexer/indexers/phoenix_sql.py

@@ -59,7 +59,7 @@ class PhoenixIndexer():
     # Until we have proper type convertion
     # Until we have proper type convertion
     for col in columns:
     for col in columns:
       if col['type'] == 'string':
       if col['type'] == 'string':
-        col['type'] = 'VARCHAR'
+        col['type'] = 'varchar'
 
 
     sql = '''CREATE TABLE IF NOT EXISTS %(table_name)s (
     sql = '''CREATE TABLE IF NOT EXISTS %(table_name)s (
 %(columns)s
 %(columns)s
@@ -73,17 +73,21 @@ CONSTRAINT my_pk PRIMARY KEY (%(primary_keys)s)
       }
       }
 
 
     source_path = urllib_unquote(source['path'])
     source_path = urllib_unquote(source['path'])
-    file_obj = request.fs.open(source_path)
-    content = file_obj.read().decode("utf-8")
-    csvfile = string_io(content)
-    reader = csv.reader(csvfile)
+    if source['inputFormat'] == 'file':
+      file_obj = request.fs.open(source_path)
+      content = file_obj.read().decode("utf-8")
+      csvfile = string_io(content)
+      reader = csv.reader(csvfile)
+    else:
+      local_file = open(source_path, 'r')
+      reader = csv.reader(local_file)
 
 
     if destination['indexerRunJob']:
     if destination['indexerRunJob']:
       for count, csv_row in enumerate(reader):
       for count, csv_row in enumerate(reader):
         if (source['format']['hasHeader'] and count == 0) or not csv_row:
         if (source['format']['hasHeader'] and count == 0) or not csv_row:
             continue
             continue
         else:
         else:
-          _sql = ', '.join([ "'{0}'".format(col_val) if columns[count]['type'] in ('VARCHAR', 'timestamp') \
+          _sql = ', '.join([ "'{0}'".format(col_val) if columns[count]['type'] in ('varchar', 'timestamp') \
             else '{0}'.format(col_val) for count, col_val in enumerate(csv_row)])
             else '{0}'.format(col_val) for count, col_val in enumerate(csv_row)])
 
 
           sql += '''\nUPSERT INTO %(table_name)s VALUES (%(csv_row)s);\n''' % {
           sql += '''\nUPSERT INTO %(table_name)s VALUES (%(csv_row)s);\n''' % {

+ 83 - 0
desktop/libs/indexer/src/indexer/indexers/phoenix_sql_tests.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.from indexer.indexers.phoenix_sql import PhoenixIndexer
+
+import sys
+from nose.tools import assert_equal
+
+from desktop.settings import BASE_DIR
+from indexer.indexers.phoenix_sql import PhoenixIndexer
+
+if sys.version_info[0] > 2:
+  from unittest.mock import patch, Mock, MagicMock
+else:
+  from mock import patch, Mock, MagicMock
+
+
+def test_create_table_phoenix():
+  with patch('indexer.indexers.phoenix_sql.get_ordered_interpreters') as get_ordered_interpreters:
+    get_ordered_interpreters.return_value = [{'Name': 'Phoenix', 'dialect': 'phoenix', 'type': 'phoenix'}]
+    source = {
+      'inputFormat': 'localfile',
+      'path': BASE_DIR + '/apps/beeswax/data/tables/us_population.csv',
+      'sourceType': 'phoenix',
+      'format': {'hasHeader': False}
+    }
+    destination = {
+      'name': 'default.test1',
+      'columns': [
+        {'name': 'field_1', 'type': 'string'},
+        {'name': 'field_2', 'type': 'string'},
+        {'name': 'field_3', 'type': 'bigint'},
+      ],
+      'sourceType': 'phoenix',
+      'indexerPrimaryKey': ['field_3'],
+      'indexerRunJob': True
+    }
+    request = Mock()
+    sql = PhoenixIndexer(user=Mock(), fs=Mock()).create_table_from_file(request, source, destination).get_str()
+
+    statement = '''USE default;
+
+CREATE TABLE IF NOT EXISTS test1 (
+  field_1 varchar,
+  field_2 varchar,
+  field_3 bigint
+CONSTRAINT my_pk PRIMARY KEY (field_3)
+);
+
+UPSERT INTO test1 VALUES ('NY', 'New York', 8143197);
+
+UPSERT INTO test1 VALUES ('CA', 'Los Angeles', 3844829);
+
+UPSERT INTO test1 VALUES ('IL', 'Chicago', 2842518);
+
+UPSERT INTO test1 VALUES ('TX', 'Houston', 2016582);
+
+UPSERT INTO test1 VALUES ('PA', 'Philadelphia', 1463281);
+
+UPSERT INTO test1 VALUES ('AZ', 'Phoenix', 1461575);
+
+UPSERT INTO test1 VALUES ('TX', 'San Antonio', 1256509);
+
+UPSERT INTO test1 VALUES ('CA', 'San Diego', 1255540);
+
+UPSERT INTO test1 VALUES ('TX', 'Dallas', 1213825);
+
+UPSERT INTO test1 VALUES ('CA', 'San Jose', 912332);'''
+
+    assert_equal(statement, sql)

+ 0 - 25
desktop/libs/indexer/src/indexer/indexers/sql.py

@@ -326,21 +326,6 @@ class SQLIndexer(object):
         'columns': ',\n'.join(['  `%(name)s` %(type)s' % col for col in columns]),
         'columns': ',\n'.join(['  `%(name)s` %(type)s' % col for col in columns]),
       }
       }
 
 
-    elif dialect == 'phoenix':
-
-      for col in columns:
-        if col['type'] == 'string':
-          col['type'] = 'CHAR(255)'
-
-      sql = '''CREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s (
-%(columns)s
-CONSTRAINT my_pk PRIMARY KEY (%(primary_keys)s));\n''' % {
-          'database': database,
-          'table_name': table_name,
-          'columns': ',\n'.join(['  %(name)s %(type)s' % col for col in columns]),
-          'primary_keys': ', '.join(destination.get('primaryKeys'))
-      }
-
     elif dialect == 'impala':
     elif dialect == 'impala':
       sql = '''CREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s_tmp (
       sql = '''CREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s_tmp (
 %(columns)s);\n''' % {
 %(columns)s);\n''' % {
@@ -372,16 +357,6 @@ CONSTRAINT my_pk PRIMARY KEY (%(primary_keys)s));\n''' % {
               'table_name': table_name,
               'table_name': table_name,
               'csv_rows': csv_rows
               'csv_rows': csv_rows
             }
             }
-          elif dialect == 'phoenix':
-            for csv_row in _csv_rows:
-              _sql = ', '.join([ "'{0}'".format(col_val) if columns[count]['type'] in ('CHAR(255)', 'timestamp') \
-                else '{0}'.format(col_val) for count, col_val in enumerate(csv_row)])
-
-              sql += '''\nUPSERT INTO %(database)s.%(table_name)s VALUES (%(csv_row)s);\n''' % {
-                'database': database,
-                'table_name': table_name,
-                'csv_row': _sql
-              }
           elif dialect == 'impala':
           elif dialect == 'impala':
              # casting from string to boolean is not allowed in impala so string -> int -> bool
              # casting from string to boolean is not allowed in impala so string -> int -> bool
             sql_ = ',\n'.join([
             sql_ = ',\n'.join([

+ 0 - 51
desktop/libs/indexer/src/indexer/indexers/sql_tests.py

@@ -872,57 +872,6 @@ INSERT INTO default.test1 VALUES ('NY', 'New York', '8143197'), ('CA', 'Los Ange
     assert_equal(statement, sql)
     assert_equal(statement, sql)
 
 
 
 
-def test_create_table_from_local_phoenix():
-  with patch('indexer.indexers.sql.get_interpreter') as get_interpreter:
-    get_interpreter.return_value = {'Name': 'Phoenix', 'dialect': 'phoenix'}
-    source = {
-      'path': BASE_DIR + '/apps/beeswax/data/tables/us_population.csv',
-      'sourceType': 'phoenix',
-      'format': {'hasHeader': False}
-    }
-    destination = {
-      'name': 'default.test1',
-      'columns': [
-        {'name': 'field_1', 'type': 'string'},
-        {'name': 'field_2', 'type': 'string'},
-        {'name': 'field_3', 'type': 'bigint'},
-      ],
-      'sourceType': 'phoenix',
-      'primaryKeys': ['field_3']
-    }
-    sql = SQLIndexer(user=Mock(), fs=Mock()).create_table_from_local_file(source, destination).get_str()
-
-    statement = '''USE default;
-
-CREATE TABLE IF NOT EXISTS default.test1 (
-  field_1 CHAR(255),
-  field_2 CHAR(255),
-  field_3 bigint
-CONSTRAINT my_pk PRIMARY KEY (field_3));
-
-UPSERT INTO default.test1 VALUES ('NY', 'New York', 8143197);
-
-UPSERT INTO default.test1 VALUES ('CA', 'Los Angeles', 3844829);
-
-UPSERT INTO default.test1 VALUES ('IL', 'Chicago', 2842518);
-
-UPSERT INTO default.test1 VALUES ('TX', 'Houston', 2016582);
-
-UPSERT INTO default.test1 VALUES ('PA', 'Philadelphia', 1463281);
-
-UPSERT INTO default.test1 VALUES ('AZ', 'Phoenix', 1461575);
-
-UPSERT INTO default.test1 VALUES ('TX', 'San Antonio', 1256509);
-
-UPSERT INTO default.test1 VALUES ('CA', 'San Diego', 1255540);
-
-UPSERT INTO default.test1 VALUES ('TX', 'Dallas', 1213825);
-
-UPSERT INTO default.test1 VALUES ('CA', 'San Jose', 912332);'''
-
-    assert_equal(statement, sql)
-
-
 def test_create_table_from_local_impala():
 def test_create_table_from_local_impala():
   with patch('indexer.indexers.sql.get_interpreter') as get_interpreter:
   with patch('indexer.indexers.sql.get_interpreter') as get_interpreter:
     get_interpreter.return_value = {'Name': 'Impala', 'dialect': 'impala'}
     get_interpreter.return_value = {'Name': 'Impala', 'dialect': 'impala'}

+ 16 - 7
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -781,7 +781,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
           <div class="card-body">
           <div class="card-body">
             % if ENABLE_SCALABLE_INDEXER.get():
             % if ENABLE_SCALABLE_INDEXER.get():
             <div class="control-group">
             <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() == 'file'">
+              <label class="checkbox inline-block" title="${ _('Execute a cluster job to index a large dataset.') }" data-bind="visible: (['file', 'localfile'].indexOf($root.createWizard.source.inputFormat()) != -1)">
                 <input type="checkbox" data-bind="checked: indexerRunJob">
                 <input type="checkbox" data-bind="checked: indexerRunJob">
                   <!-- ko if: outputFormat() == 'index' -->
                   <!-- ko if: outputFormat() == 'index' -->
                     ${_('Index with a job')}
                     ${_('Index with a job')}
@@ -1794,7 +1794,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
       });
       });
 
 
       self.interpreters = ko.pureComputed(function() {
       self.interpreters = ko.pureComputed(function() {
-        return window.getLastKnownConfig().app_config.editor.interpreters.filter(function (interpreter) { return interpreter.is_sql });
+        return window.getLastKnownConfig().app_config.editor.interpreters.filter(function (interpreter) { return interpreter.is_sql && interpreter.dialect != 'phoenix' });
       });
       });
       self.interpreter = ko.observable(vm.sourceType);
       self.interpreter = ko.observable(vm.sourceType);
       self.interpreter.subscribe(function(val) {
       self.interpreter.subscribe(function(val) {
@@ -2346,7 +2346,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
           if (format.value === 'stream-table' && ['stream'].indexOf(wizard.source.inputFormat()) === -1) {
           if (format.value === 'stream-table' && ['stream'].indexOf(wizard.source.inputFormat()) === -1) {
             return false;
             return false;
           }
           }
-          if (format.value === 'big-table' && ['file'].indexOf(wizard.source.inputFormat()) === -1) {
+          if (format.value === 'big-table' && ['file', 'localfile'].indexOf(wizard.source.inputFormat()) === -1) {
             return false;
             return false;
           }
           }
           if (format.value === 'hbase' && (wizard.source.inputFormat() !== 'rdbms' || wizard.source.rdbmsAllTablesSelected())) {
           if (format.value === 'hbase' && (wizard.source.inputFormat() !== 'rdbms' || wizard.source.rdbmsAllTablesSelected())) {
@@ -2374,8 +2374,8 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
       self.defaultName = ko.pureComputed(function() {
       self.defaultName = ko.pureComputed(function() {
         var name = '';
         var name = '';
 
 
-        if (wizard.source.inputFormat() === 'file' || wizard.source.inputFormat() === 'stream') {
-          if (self.outputFormat() === 'table') {
+        if (['file', 'stream', 'localfile'].indexOf(wizard.source.inputFormat()) != -1) {
+          if (['table', 'big-table'].indexOf(self.outputFormat()) != -1) {
             name = wizard.prefill.target_path().length > 0 ? wizard.prefill.target_path() : 'default';
             name = wizard.prefill.target_path().length > 0 ? wizard.prefill.target_path() : 'default';
 
 
             if (wizard.source.inputFormat() === 'stream') {
             if (wizard.source.inputFormat() === 'stream') {
@@ -2385,7 +2385,16 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
                 name += '.' + wizard.source.streamObject();
                 name += '.' + wizard.source.streamObject();
               }
               }
             } else if (wizard.source.path()) {
             } else if (wizard.source.path()) {
-              name += '.' + wizard.source.path().split('/').pop().split('.')[0];
+              const source_path = wizard.source.path();
+              var database_name = name += '.';
+              if (self.outputFormat() === 'big-table' && wizard.prefill.target_path().length === 0) {
+                database_name = '';
+              }
+              if (wizard.source.inputFormat() === 'localfile') {
+                name = database_name + source_path.substring(source_path.lastIndexOf(':') + 1, source_path.lastIndexOf(';')).split('.')[0];
+              } else {
+                name = database_name + source_path.split('/').pop().split('.')[0];
+              }
             }
             }
           } else { // Index
           } else { // Index
             name = wizard.prefill.target_path().length > 0 ? wizard.prefill.target_path() : wizard.source.path().split('/').pop().split('.')[0];
             name = wizard.prefill.target_path().length > 0 ? wizard.prefill.target_path() : wizard.source.path().split('/').pop().split('.')[0];
@@ -2840,7 +2849,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
       };
       };
       self.loadSampleData = function(resp) {
       self.loadSampleData = function(resp) {
         resp.columns.forEach(function (entry, i, arr) {
         resp.columns.forEach(function (entry, i, arr) {
-          if (self.destination.outputFormat() === 'table' && self.source.inputFormat() != 'rdbms') {
+          if (['table', 'big-table'].indexOf(self.destination.outputFormat()) != -1 && self.source.inputFormat() != 'rdbms') {
             entry.type = MAPPINGS.get(MAPPINGS.SOLR_TO_HIVE, entry.type, 'string');
             entry.type = MAPPINGS.get(MAPPINGS.SOLR_TO_HIVE, entry.type, 'string');
           } else if (self.destination.outputFormat() === 'index') {
           } else if (self.destination.outputFormat() === 'index') {
             entry.type = MAPPINGS.get(MAPPINGS.HIVE_TO_SOLR, entry.type, entry.type);
             entry.type = MAPPINGS.get(MAPPINGS.HIVE_TO_SOLR, entry.type, entry.type);