Kaynağa Gözat

[importer] Fixing only headers CSV file table creation in Local File importer on Impala (#2967)

Ayush Goyal 3 yıl önce
ebeveyn
işleme
42994728a5

+ 1 - 0
apps/beeswax/data/tables/onlyheader.csv

@@ -0,0 +1 @@
+"date(1)","hour.1","minute","dep","arr","dep_delay","arr_delay","carrier","flight","dest","plane","cancelled","time","dist"

+ 15 - 10
desktop/libs/indexer/src/indexer/indexers/sql.py

@@ -358,7 +358,7 @@ class SQLIndexer(object):
             row = self.nomalize_booleans(row, columns)
           _csv_rows.append(tuple(row))
 
-        if _csv_rows:
+        if _csv_rows:  #sql for data insertion
           csv_rows = str(_csv_rows)[1:-1]
 
           if dialect in ('hive', 'mysql'):
@@ -368,18 +368,23 @@ class SQLIndexer(object):
               'csv_rows': csv_rows
             }
           elif dialect == 'impala':
-             # casting from string to boolean is not allowed in impala so string -> int -> bool
-            sql_ = ',\n'.join([
-              '  CAST ( `%(name)s` AS %(type)s ) `%(name)s`' % col if col['type'] != 'boolean' \
-              else '  CAST ( CAST ( `%(name)s` AS TINYINT ) AS boolean ) `%(name)s`' % col for col in columns
-            ])
-
-            sql += '''\nINSERT INTO %(database)s.%(table_name)s_tmp VALUES %(csv_rows)s;\n
-CREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s
-AS SELECT\n%(sql_)s\nFROM  %(database)s.%(table_name)s_tmp;\n\nDROP TABLE IF EXISTS %(database)s.%(table_name)s_tmp;'''% {
+            sql += '''\nINSERT INTO %(database)s.%(table_name)s_tmp VALUES %(csv_rows)s;\n'''% {
               'database': database,
               'table_name': table_name,
               'csv_rows': csv_rows,
+            }
+
+        if dialect == 'impala':
+          # casting from string to boolean is not allowed in impala so string -> int -> bool
+          sql_ = ',\n'.join([
+            '  CAST ( `%(name)s` AS %(type)s ) `%(name)s`' % col if col['type'] != 'boolean' \
+            else '  CAST ( CAST ( `%(name)s` AS TINYINT ) AS boolean ) `%(name)s`' % col for col in columns
+          ])
+
+          sql += '''\nCREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s
+AS SELECT\n%(sql_)s\nFROM  %(database)s.%(table_name)s_tmp;\n\nDROP TABLE IF EXISTS %(database)s.%(table_name)s_tmp;'''% {
+              'database': database,
+              'table_name': table_name,
               'sql_': sql_
             }
 

+ 72 - 2
desktop/libs/indexer/src/indexer/indexers/sql_tests.py

@@ -900,8 +900,7 @@ def test_create_table_from_local_impala():
       ],
       'sourceType': 'impala'
     }
-    request = MockRequest(fs=MockFs())
-    sql = SQLIndexer(user=request.user, fs=request.fs).create_table_from_local_file(source, destination).get_str()
+    sql = SQLIndexer(user=Mock(), fs=Mock()).create_table_from_local_file(source, destination).get_str()
 
     statement = '''USE default;
 
@@ -954,6 +953,77 @@ DROP TABLE IF EXISTS default.test1_tmp;'''
     assert_equal(statement, sql)
 
 
+def test_create_table_only_header_file_local_impala():
+  with patch('indexer.indexers.sql.get_interpreter') as get_interpreter:
+    get_interpreter.return_value = {'Name': 'Impala', 'dialect': 'impala'}
+    source = {
+      'path': BASE_DIR + '/apps/beeswax/data/tables/onlyheader.csv',
+      'sourceType': 'impala',
+      'format': {'hasHeader': True}
+    }
+    destination = {
+      'name': 'default.test1',
+      'columns': [
+        {'name': 'date', 'type': 'timestamp', 'keep': True},
+        {'name': 'hour', 'type': 'bigint', 'keep': True},
+        {'name': 'minute', 'type': 'bigint', 'keep': True},
+        {'name': 'dep', 'type': 'bigint', 'keep': True},
+        {'name': 'arr', 'type': 'bigint', 'keep': True},
+        {'name': 'dep_delay', 'type': 'bigint', 'keep': True},
+        {'name': 'arr_delay', 'type': 'bigint', 'keep': True},
+        {'name': 'carrier', 'type': 'string', 'keep': True},
+        {'name': 'flight', 'type': 'bigint', 'keep': True},
+        {'name': 'dest', 'type': 'string', 'keep': True},
+        {'name': 'plane', 'type': 'string', 'keep': True},
+        {'name': 'cancelled', 'type': 'boolean', 'keep': True},
+        {'name': 'time', 'type': 'bigint', 'keep': True},
+        {'name': 'dist', 'type': 'bigint', 'keep': True},
+      ],
+      'sourceType': 'impala'
+    }
+    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_tmp (
+  `date` string,
+  `hour` string,
+  `minute` string,
+  `dep` string,
+  `arr` string,
+  `dep_delay` string,
+  `arr_delay` string,
+  `carrier` string,
+  `flight` string,
+  `dest` string,
+  `plane` string,
+  `cancelled` string,
+  `time` string,
+  `dist` string);
+
+CREATE TABLE IF NOT EXISTS default.test1
+AS SELECT
+  CAST ( `date` AS timestamp ) `date`,
+  CAST ( `hour` AS bigint ) `hour`,
+  CAST ( `minute` AS bigint ) `minute`,
+  CAST ( `dep` AS bigint ) `dep`,
+  CAST ( `arr` AS bigint ) `arr`,
+  CAST ( `dep_delay` AS bigint ) `dep_delay`,
+  CAST ( `arr_delay` AS bigint ) `arr_delay`,
+  CAST ( `carrier` AS string ) `carrier`,
+  CAST ( `flight` AS bigint ) `flight`,
+  CAST ( `dest` AS string ) `dest`,
+  CAST ( `plane` AS string ) `plane`,
+  CAST ( CAST ( `cancelled` AS TINYINT ) AS boolean ) `cancelled`,
+  CAST ( `time` AS bigint ) `time`,
+  CAST ( `dist` AS bigint ) `dist`
+FROM  default.test1_tmp;
+
+DROP TABLE IF EXISTS default.test1_tmp;'''
+
+    assert_equal(statement, sql)
+
+
 def test_create_table_with_drop_column_from_local():
   with patch('indexer.indexers.sql.get_interpreter') as get_interpreter:
     get_interpreter.return_value = {'Name': 'Hive', 'dialect': 'hive'}