浏览代码

HUE-5129 [metastore] Add Kudu HASH partitioning SQL generation

Romain Rigaux 9 年之前
父节点
当前提交
f777306

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

@@ -214,7 +214,8 @@ def _create_table_from_a_file(request, source, destination):
 
   columns = destination['columns']
   partition_columns = destination['partitionColumns']
-
+  kudu_partition_columns = destination['kuduPartitionColumns']
+  print kudu_partition_columns
   comment = destination['description']
 
   source_path = source['path']
@@ -255,16 +256,22 @@ def _create_table_from_a_file(request, source, destination):
    "escapeChar"    = "\\\\"
    '''
 
-  if load_data:
-    if table_format in ('parquet', 'kudu'):
+  if table_format in ('parquet', 'kudu'):
+    if load_data:
       table_name, final_table_name = 'hue__tmp_%s' % table_name, table_name
-
+  
       sql += '\n\nDROP TABLE IF EXISTS `%(database)s`.`%(table_name)s`;\n' % {
           'database': database,
           'table_name': table_name
       }
+    else:
+      row_format = ''
+      file_format = table_format
+      skip_header = False
+      if table_format == 'kudu':
+        columns = [col for col in columns if col['name'] in primary_keys] + [col for col in columns if col['name'] not in primary_keys]      
 
-  if external or load_data and table_format in ('parquet', 'kudu'):
+  if external or (load_data and table_format in ('parquet', 'kudu')):
     if not request.fs.isdir(external_path): # File selected
       external_path, external_file_name = request.fs.split(external_path)
 
@@ -287,10 +294,11 @@ def _create_table_from_a_file(request, source, destination):
           'external': external or load_data and table_format in ('parquet', 'kudu'),
           'path': external_path,
           'skip_header': skip_header,
-          'primary_keys': primary_keys if table_format == 'kudu' and not load_data else []
+          'primary_keys': primary_keys if table_format == 'kudu' and not load_data else [],
        },
       'columns': columns,
       'partition_columns': partition_columns,
+      'kudu_partition_columns': kudu_partition_columns,
       'database': database
     }
   )
@@ -298,10 +306,10 @@ def _create_table_from_a_file(request, source, destination):
   if table_format == 'text' and not external and load_data:
     sql += "\n\nLOAD DATA INPATH '%s' INTO TABLE `%s`.`%s`;" % (source_path, database, table_name)
 
-  if table_format in ('parquet', 'kudu'):
+  if load_data and table_format in ('parquet', 'kudu'):
     file_format = table_format
     if table_format == 'kudu':
-      columns_list = ['`%s`' % col for col in primary_keys + [col['name'] for col in destination['columns'] if col['name'] not in primary_keys]]
+      columns_list = ['`%s`' % col for col in primary_keys] + [col['name'] for col in destination['columns'] if col['name'] not in primary_keys]
       extra_create_properties = """PRIMARY KEY (%(primary_keys)s)
       DISTRIBUTE BY HASH INTO 16 BUCKETS
       STORED AS %(file_format)s

+ 23 - 4
desktop/libs/indexer/src/indexer/templates/gen/create_table_statement.mako

@@ -48,23 +48,42 @@ COMMENT "${col["comment"]|n}" \
 ) \
 </%def>\
 
+<%def name="kudu_partition(partition)">
+% if partition['name'] == 'HASH':
+  HASH (${ ', '.join(partition['columns']) }) PARTITIONS ${ partition['int_val'] }
+% elif partition['name'] == 'RANGE BY':
+  RANGE BY (${ ', '.join(partition['columns']) }) (${ ', '.join([kudu_range_partition(range_partition) for range_partition in partition['range_partitions']]) })
+% endif
+</%def>
+
+<%def name="kudu_range_partition(partition)">
+% if partition['name'] == 'HASH':
+  HASH (${ ', '.join(partition['columns']) }) PARTITIONS ${ partition['int_val'] }
+% elif partition['name'] == 'RANGE BY':
+  RANGE BY (${ ', '.join(partition['columns']) }) (${ ', '.join(partition['range_partitions']) })
+% endif
+</%def>
+
 
 CREATE \
 % if table.get("external", False):
 EXTERNAL \
 % endif
 TABLE ${ '`%s`.`%s`' % (database, table["name"]) | n }
-${ column_list(table, columns) | n }
+${ column_list(table, columns) | n } \
 % if table["comment"]:
 COMMENT "${table["comment"] | n }"
 % endif
-% if len(partition_columns) > 0:
+% if partition_columns and table.get('file_format') != 'kudu':
 PARTITIONED BY ${ column_list(table, partition_columns) | n }
 % endif
+% if kudu_partition_columns  and table.get('file_format') == 'kudu':
+PARTITION BY ${ ', '.join([kudu_partition(partition) for partition in kudu_partition_columns]) | n }
+% endif
 ## TODO: CLUSTERED BY here
 ## TODO: SORTED BY...INTO...BUCKETS here
-ROW FORMAT \
 % if table.get('row_format'):
+ROW FORMAT \
 %   if table["row_format"] == "Delimited":
   DELIMITED
 %     if table.has_key('field_terminator'):
@@ -84,7 +103,7 @@ ROW FORMAT \
 %   endif
 % endif
 % if table.has_key('file_format'):
-  STORED AS ${table["file_format"] | n} \
+  STORED AS ${ table["file_format"] | n } \
 % endif
 % if table.get("file_format") == "InputFormat":
 INPUTFORMAT ${table["input_format_class"] | n} OUTPUTFORMAT ${table["output_format_class"] | n}

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

@@ -971,14 +971,13 @@ ${ assist.assistPanel() }
             name = val.name();
           }
         } else if (self.inputFormat() == 'manual') {
-          name = '.';
+          name = wizard.prefill.target_path ? wizard.prefill.target_path() + '.' : '';
         }
 
         return name.replace(' ', '_');
       });
       self.defaultName.subscribe(function(newVal) {
-        var prefix = wizard.prefill.target_path ? wizard.prefill.target_path() + (newVal == '.' ? '' : '.') : '';
-        vm.createWizard.destination.name(prefix + newVal);
+        vm.createWizard.destination.name(newVal);
       });
     };
 
@@ -1024,7 +1023,9 @@ ${ assist.assistPanel() }
           {'value': 'json', 'name': 'Json'},
           {'value': 'kudu', 'name': 'Kudu'},
           {'value': 'orc', 'name': 'ORC'},
-          {'value': 'avro', 'name': 'Avro'}
+          {'value': 'avro', 'name': 'Avro'},
+          {'value': 'rcfile', 'name': 'RCFile'},
+          {'value': 'sequencefile', 'name': 'SequenceFile'}
       ]);
 
       self.partitionColumns = ko.observableArray();
@@ -1225,10 +1226,10 @@ ${ assist.assistPanel() }
                 var snippet = self.editorVM.selectedNotebook().snippets()[0]; // Could be native to editor at some point
                 if (! snippet.result.handle().has_more_statements) {
                   if (self.editorVM.selectedNotebook().onSuccessUrl()) {
-                    window.location.href = self.editorVM.selectedNotebook().onSuccessUrl();
+                  //  window.location.href = self.editorVM.selectedNotebook().onSuccessUrl();
                   }
                 } else { // Perform last DROP statement execute
-                  //snippet.execute();
+                  snippet.execute();
                 }
               }
             });