فهرست منبع

HUE-4266 [indexer] Add geo, host, grok, split operations

Aaron Peddle 9 سال پیش
والد
کامیت
1e15498716

+ 1 - 1
apps/oozie/src/oozie/models2.py

@@ -68,7 +68,7 @@ class Job(object):
 
   @classmethod
   def get_workspace(cls, user):
-    if type(user) is User:
+    if not isinstance(user, basestring):
       user = user.username
     return (REMOTE_SAMPLE_DIR.get() + '/hue-oozie-$TIME').replace('$USER', user).replace('$TIME', str(time.time()))
 

+ 124 - 0
desktop/libs/indexer/src/data/oozie_workspace/clean_to_match_schema.conf

@@ -0,0 +1,124 @@
+# 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 nose.tools import assert_equal
+<%page args="fields, get_regex"/>
+
+
+# require that all kept fields are present
+% for field in fields:
+  % if field["keep"] and field["required"]:
+    {
+      if {
+        conditions : [
+          { equals { "${field['name']}" : [] } }
+        ]
+        then : [
+          { logError { format : "Ignoring record because it has no ${field['name']}: {}", args : ["@{}"] } }
+          { dropRecord {} }
+        ]
+      }
+    }
+  % endif
+%endfor
+
+# convert common date formats to solr date format
+% for field in fields:
+  %if field["type"] == 'date':
+    {
+      convertTimestamp {
+        field : ${field['name']}
+        outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
+        outputTimezone : UTC
+      }
+    }
+  %endif
+%endfor
+
+# require that all kept fields match their expected type
+% for field in fields:
+  %if field["keep"]:
+    {
+      if {
+        conditions : [
+          {
+            grok {
+              expressions : {
+                "${field['name']}" : "${get_regex(field['type'])}"
+              }
+              extract : false
+
+            }
+          }
+        ]
+        then : []
+        else : [
+          %if not field["required"]:
+            # if didn't match because the field isn't present then we keep the record
+            {
+              if {
+                conditions : [
+                    { equals { "${field['name']}" : [] } }
+                ]
+                then : []
+                else : [
+                  { logError { format : "Ignoring record due to incorrect type for ${field['name']}: {}", args : ["@{}"] } }
+                  { dropRecord {} }
+                ]
+              }
+            }
+          %else:
+            { logError { format : "Ignoring record due to incorrect type for ${field['name']}: {}", args : ["@{}"] } }
+            { dropRecord {} }
+          %endif
+        ]
+      }
+    }
+  %endif
+%endfor
+
+# treat empty strings as if the field wasn't in the tuple unless the type is string
+% for field in fields:
+  % if field["type"] != "string":
+    {
+      if {
+        conditions : [
+          {
+            grok {
+              expressions : {
+                "${field['name']}" : "^$"
+              }
+              extract : false
+            }
+          }
+        ]
+        then : [
+          {
+            removeFields {
+              blacklist : ["literal:${field['name']}"]
+            }
+          }
+        ]
+      }
+    }
+  % endif
+%endfor
+
+# remove excess fields
+{
+  sanitizeUnknownSolrFields {
+    # Location from which to fetch Solr schema
+    solrLocator : <%text>${SOLR_LOCATOR}</%text>
+  }
+}

+ 1 - 1
desktop/libs/indexer/src/data/oozie_workspace/convert_date_operation.conf

@@ -24,4 +24,4 @@
     outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
     outputTimezone : UTC
   }
-}
+}

+ 29 - 0
desktop/libs/indexer/src/data/oozie_workspace/extract_uri_components_operation.conf

@@ -0,0 +1,29 @@
+# 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 nose.tools import assert_equal
+<%page args="field, operation"/>
+
+# field
+# operation
+%for i, component in enumerate(get_kept_args(operation)):
+
+  {
+    extractURIComponent {
+      inputField : "${field['name']}"
+      outputField : "${operation['fields'][i]['name']}"
+      component : ${component.name}
+    }
+  }
+%endfor

+ 29 - 0
desktop/libs/indexer/src/data/oozie_workspace/find_replace_operation.conf

@@ -0,0 +1,29 @@
+# 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 nose.tools import assert_equal
+<%page args="field, operation"/>
+
+# field
+# operation
+{
+  findReplace {
+    field : "${field['name']}"
+    dictionaryFiles : ["grok_dictionaries"]
+    pattern : """${operation['settings']['find']}"""
+    isRegex : true
+    replacement : """${operation['settings']['replace']}"""
+    replaceFirst : false
+  }
+}

+ 39 - 0
desktop/libs/indexer/src/data/oozie_workspace/geo_ip_operation.conf

@@ -0,0 +1,39 @@
+# 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 nose.tools import assert_equal
+<%page args="field, operation"/>
+
+# field
+# operation
+{
+  geoIP {
+    inputField : "${field['name']}"
+    database : "GeoLite2-City.mmdb"
+  }
+}
+
+# extract parts of the geolocation info from the Jackson JsonNode Java
+# object contained in the _attachment_body field and store the parts in
+# the given record output fields:
+{
+  extractJsonPaths {
+    flatten : false
+    paths : {
+      %for i, component in enumerate(get_kept_args(operation)):
+        "${operation['fields'][i]['name']}" : "${component.name}"
+      %endfor
+    }
+  }
+}

+ 1 - 1
desktop/libs/indexer/src/data/oozie_workspace/grok_operation.conf

@@ -24,4 +24,4 @@
       ${field['name']}: """${operation['settings']['regexp']}"""
     }
   }
-}
+}

+ 13 - 81
desktop/libs/indexer/src/data/oozie_workspace/morphline_template.conf

@@ -13,6 +13,7 @@
 # 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.
+
 SOLR_LOCATOR : {
   # Name of solr collection
   collection : "${collection_name}"
@@ -26,26 +27,23 @@ morphlines : [
     id : ${collection_name}
     importCommands : ["org.kitesdk.**", "org.apache.solr.**"]
     commands : [
-      {
-        readCSV {
-          separator : "${format_character(format['fieldSeparator'])}"
-          columns : [
-            % for field in fields[:num_base_fields]:
-            "${field['name']}"
-            %endfor
-          ]
-          quoteChar : "${format_character(format['quoteChar'])}"
-          ignoreFirstLine : "${'true' if format['hasHeader'] else 'false'}"
-          charset : UTF-8
-        }
-      }
 
+      # download required libs
       {
         downloadHdfsFile{
-          inputFiles:["${grok_dictionaries_location}"]
+          inputFiles:[
+            %if grok_dictionaries_location:
+              "${grok_dictionaries_location}"
+            %endif
+            %if geolite_db_location:
+              "${geolite_db_location}"
+            %endif
+          ]
         }
       }
 
+      <%include file="${'parse_%s.conf' % format['type']}", args="format=format, format_character=format_character"/>
+
 
       {
         generateUUID {
@@ -61,73 +59,7 @@ morphlines : [
         %endfor
       %endfor
 
-
-      # require that all kept fields are present
-      % for field in fields:
-        % if field["keep"] and field["required"]:
-          {
-            if {
-              conditions : [
-                { equals { "${field['name']}" : [] } }
-              ]
-              then : [
-                { logError { format : "Ignoring record because it has no ${field['name']}: {}", args : ["@{}"] } }
-                { dropRecord {} }
-              ]
-            }
-          }
-        % endif
-      %endfor
-
-      # require that all kept fields match their expected type
-      % for field in fields:
-        %if field["keep"]:
-          {
-            if {
-              conditions : [
-                {
-                  grok {
-                    expressions : {
-                      "${field['name']}" : "${get_regex(field['type'])}"
-                    }
-                    extract : false
-
-                  }
-                }
-              ]
-              then : []
-              else : [
-                %if not field["required"]:
-                  # if didn't match because the field isn't present then we keep the record
-                  {
-                    if {
-                      conditions : [
-                          { equals { "${field['name']}" : [] } }
-                      ]
-                      then : []
-                      else : [
-                        { logError { format : "Ignoring record due to incorrect type for ${field['name']}: {}", args : ["@{}"] } }
-                        { dropRecord {} }
-                      ]
-                    }
-                  }
-                %else:
-                  { logError { format : "Ignoring record due to incorrect type for ${field['name']}: {}", args : ["@{}"] } }
-                  { dropRecord {} }
-                %endif
-              ]
-            }
-          }
-        %endif
-      %endfor
-
-      # remove excess fields
-      {
-        sanitizeUnknownSolrFields {
-          # Location from which to fetch Solr schema
-          solrLocator : <%text>${SOLR_LOCATOR}</%text>
-        }
-      }
+      <%include file="clean_to_match_schema.conf" args="fields=fields, get_regex=get_regex"/>
 
       # log the record at DEBUG level to SLF4J
       { logDebug { format : "output record: {}", args : ["@{}"] } }

+ 30 - 0
desktop/libs/indexer/src/data/oozie_workspace/parse_csv.conf

@@ -0,0 +1,30 @@
+# 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 nose.tools import assert_equal
+<%page args="format, format_character"/>
+
+{
+  readCSV {
+    separator : "${format_character(format['fieldSeparator'])}"
+    columns : [
+      % for field in fields[:num_base_fields]:
+      "${field['name']}"
+      %endfor
+    ]
+    quoteChar : "${format_character(format['quoteChar'])}"
+    ignoreFirstLine : "${'true' if format['hasHeader'] else 'false'}"
+    charset : UTF-8
+  }
+}

+ 1 - 1
desktop/libs/indexer/src/data/oozie_workspace/split_operation.conf

@@ -31,4 +31,4 @@
     #isRegex : true
     addEmptyStrings : false
   }
-}
+}

+ 30 - 0
desktop/libs/indexer/src/data/oozie_workspace/translate_operation.conf

@@ -0,0 +1,30 @@
+# 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 nose.tools import assert_equal
+<%page args="field, operation"/>
+
+# field
+# operation
+{
+  translate{
+    field: "${field['name']}"
+    dictionary: {
+      %for pair in operation['settings']['mapping']:
+        "${pair['key']}" : "${pair['value']}"
+      %endfor
+    }
+    fallback: ${operation['settings']['default']}
+  }
+}

+ 3 - 2
desktop/libs/indexer/src/indexer/api3.py

@@ -60,11 +60,12 @@ def _convert_format(format_dict, inverse=False):
 
 def guess_format(request):
   file_format = json.loads(request.POST.get('fileFormat', '{}'))
+
   indexer = Indexer(request.user, request.fs)
   stream = request.fs.open(file_format["path"])
   format_ = indexer.guess_format({"file":stream})
   _convert_format(format_)
-  
+
   return JsonResponse(format_)
 
 def guess_field_types(request):
@@ -87,7 +88,7 @@ def index_file(request):
 
   morphline = indexer.generate_morphline_config(collection_name, file_format, unique_field)
 
-  collection_manager = CollectionManagerController(request.user.username)
+  collection_manager = CollectionManagerController(request.user)
   if not collection_manager.collection_exists(collection_name):
     collection_manager.create_collection(collection_name, schema_fields, unique_key_field=unique_field)
 

+ 78 - 0
desktop/libs/indexer/src/indexer/fields.py

@@ -0,0 +1,78 @@
+# 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.import logging
+import re
+
+class FieldType():
+  def __init__(self, name, regex):
+    self._name = name
+    self._regex = regex
+
+  @property
+  def name(self):
+    return self._name
+
+  @property
+  def regex(self):
+    return self._regex
+
+  def matches(self, field):
+    pattern = re.compile(self._regex)
+
+    return pattern.match(field)
+
+class Field(object):
+  def __init__(self, name, field_type):
+    self.name = name
+    self.field_type = field_type
+    self.keep = True
+    self.operations = []
+    self.required = True
+
+  def to_dict(self):
+    return {'name': self.name,
+    'type': self.field_type,
+    'keep': self.keep,
+    'operations': self.operations,
+    'required': self.required}
+
+FIELD_TYPES = [
+  FieldType('text', "^.{100,}$"),
+  FieldType('string', "^.*$"),
+  FieldType('double', "^([+-]?[0-9]+\\.?[0-9]+)?$"),
+  FieldType('long', "^(?:[+-]?(?:[0-9]+))?$"),
+  FieldType('date', "^([0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+(\\.[0-9]*)?Z)?$")
+]
+
+def guess_field_type_from_samples(samples):
+  guesses = [_guess_field_type(sample) for sample in samples]
+
+  return _pick_best_field(guesses)
+
+def _guess_field_type(field_val):
+  if field_val == "":
+    return None
+
+  for field_type in FIELD_TYPES[::-1]:
+    if field_type.matches(field_val):
+      return field_type.name
+
+def _pick_best_field(types):
+  types = set(types)
+
+  for field in FIELD_TYPES:
+    if field.name in types:
+      return field.name
+  return "string"

+ 206 - 0
desktop/libs/indexer/src/indexer/file_format.py

@@ -0,0 +1,206 @@
+# 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.import logging
+import csv
+import operator
+import itertools
+import logging
+
+from indexer.fields import Field, guess_field_type_from_samples
+
+LOG = logging.getLogger(__name__)
+
+
+def _valid_csv_format(format_):
+  valid_field_separator = "fieldSeparator" in format_ and len(format_["fieldSeparator"]) == 1
+  valid_record_separator = "recordSeparator" in format_ and len(format_["recordSeparator"]) == 1
+  valid_quote_char = "quoteChar" in format_ and len(format_["quoteChar"]) == 1
+  valid_has_header = "recordSeparator" in format_
+
+  return valid_has_header and valid_quote_char and valid_record_separator and valid_field_separator
+
+def get_file_format_instance(file_stream, format_=None):
+  if not format_ or _valid_csv_format(format_):
+    return CSVFormat(file_stream, format_)
+  return None
+
+class FileFormat(object):
+  def __init__(self):
+    pass
+
+  @property
+  def format_(self):
+    pass
+
+  @property
+  def sample(self):
+    pass
+
+  @property
+  def fields(self):
+    return []
+
+  def get_format(self):
+    return self.format_
+
+  def get_fields(self):
+    obj = {}
+
+    obj['columns'] = [field.to_dict() for field in self.fields]
+    obj['sample'] = self.sample
+
+    return obj
+
+  def to_dict(self):
+    obj = {}
+
+    obj['format'] = self.format_
+    obj['columns'] = [field.to_dict() for field in self.fields]
+    obj['sample'] = self.sample
+
+    return obj
+
+class CSVFormat(FileFormat):
+  def __init__(self, file_stream, format_=None):
+    file_stream.seek(0)
+    sample = '\n'.join(file_stream.read(1024*1024*5).splitlines())
+    file_stream.seek(0)
+
+    if format_:
+      self._delimiter = format_["fieldSeparator"].encode('utf-8')
+      self._line_terminator = format_["recordSeparator"].encode('utf-8')
+      self._quote_char = format_["quoteChar"].encode('utf-8')
+      self._has_header = format_["hasHeader"]
+    else:
+      dialect, self._has_header = self._guess_dialect(sample)
+      self._delimiter = dialect.delimiter
+      self._line_terminator = dialect.lineterminator
+      self._quote_char = dialect.quotechar
+
+    # sniffer insists on \r\n even when \n. This is safer and good enough for a preview
+    self._line_terminator = self._line_terminator.replace("\r\n", "\n")
+
+    self._sample_rows = self._get_sample_rows(sample)
+
+    self._num_columns = self._guess_num_columns(self._sample_rows)
+
+    self._fields = self._guess_fields(sample)
+
+    super(CSVFormat, self).__init__()
+
+  @property
+  def sample(self):
+    return self._sample_rows
+
+  @property
+  def fields(self):
+    return self._fields
+
+  @property
+  def delimiter(self):
+    return self._delimiter
+
+  @property
+  def line_terminator(self):
+    return self._line_terminator
+
+  @property
+  def quote_char(self):
+    return self._quote_char
+
+  @property
+  def format_(self):
+    return {
+      "type":"csv",
+      "fieldSeparator":self.delimiter,
+      "recordSeparator":self.line_terminator,
+      "quoteChar":self.quote_char,
+      "hasHeader":self._has_header
+    }
+
+  def _guess_dialect(self, sample):
+    sniffer = csv.Sniffer()
+    dialect = sniffer.sniff(sample)
+    has_header = sniffer.has_header(sample)
+    return dialect, has_header
+
+  def _guess_num_columns(self, sample_rows):
+    counts = {}
+
+    for row in sample_rows:
+      num_columns = len(row)
+
+      if num_columns not in counts:
+        counts[num_columns] = 0
+      counts[num_columns] += 1
+
+    if counts:
+      num_columns_guess = max(counts.iteritems(), key=operator.itemgetter(1))[0]
+    else:
+      num_columns_guess = 0
+    return num_columns_guess
+
+  def _guess_field_types(self, sample_rows):
+    field_type_guesses = []
+
+    num_columns = self._num_columns
+
+    for col in range(num_columns):
+      column_samples = [sample_row[col] for sample_row in sample_rows if len(sample_row) > col]
+
+      field_type_guess = guess_field_type_from_samples(column_samples)
+      field_type_guesses.append(field_type_guess)
+
+    return field_type_guesses
+
+  def _get_sample_reader(self, sample):
+    if self.line_terminator != '\n':
+      sample = sample.replace('\n', '\\n')
+    return csv.reader(sample.split(self.line_terminator), delimiter=self.delimiter, quotechar=self.quote_char)
+
+  def _guess_field_names(self, sample):
+    reader = self._get_sample_reader(sample)
+
+    first_row = reader.next()
+
+    if self._has_header:
+      header = first_row
+    else:
+      header = ["field_%d" % (i+1) for i in range(self._num_columns)]
+
+    return header
+
+  def _get_sample_rows(self, sample):
+    NUM_SAMPLES = 5
+
+    header_offset = 1 if self._has_header else 0
+    reader = itertools.islice(self._get_sample_reader(sample), header_offset, NUM_SAMPLES + 1)
+
+    sample_rows = list(reader)
+    return sample_rows
+
+  def _guess_fields(self, sample):
+    header = self._guess_field_names(sample)
+    types = self._guess_field_types(self._sample_rows)
+
+    if len(header) == len(types):
+      # create the fields
+      fields = [Field(header[i], types[i]) for i in range(len(header))]
+    else:
+      # likely failed to guess correctly
+      LOG.warn("Guess field types failed - number of headers didn't match number of predicted types.")
+      fields = []
+
+    return fields

+ 131 - 0
desktop/libs/indexer/src/indexer/operations.py

@@ -0,0 +1,131 @@
+# 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.import logging
+
+class Argument():
+  def __init__(self, type_, name):
+    self._name = name
+    self._type = type_
+
+  @property
+  def name(self):
+    return self._name
+
+  @property
+  def type(self):
+    return self._type
+
+  def to_dict(self):
+    return {"name": self._name, "type": self._type}
+
+class Operator():
+  def __init__(self, name, args, output_type):
+    self._name = name
+    self._args = args
+    self._output_type = output_type
+
+  @property
+  def name(self):
+    return self._name
+
+  @property
+  def args(self):
+    return self._args
+
+  def to_dict(self):
+    return {
+      "name": self._name,
+      "args": [arg.to_dict() for arg in self._args],
+      "outputType": self._output_type
+    }
+
+OPERATORS = [
+  Operator(
+    name="split",
+    args=[
+      Argument("text", "splitChar")
+    ],
+    output_type="custom_fields"
+  ),
+  Operator(
+    name="grok",
+    args=[
+      Argument("text", "regexp")
+    ],
+    output_type="custom_fields"
+  ),
+  Operator(
+    name="convert_date",
+    args=[
+      Argument("text", "format")
+    ],
+    output_type="inplace"
+  ),
+  Operator(
+    name="extract_uri_components",
+    args=[
+      Argument("checkbox", "authority"),
+      Argument("checkbox", "fragment"),
+      Argument("checkbox", "host"),
+      Argument("checkbox", "path"),
+      Argument("checkbox", "port"),
+      Argument("checkbox", "query"),
+      Argument("checkbox", "scheme"),
+      Argument("checkbox", "scheme_specific_path"),
+      Argument("checkbox", "user_info")
+    ],
+    output_type="checkbox_fields"
+  ),
+  Operator(
+    name="geo_ip",
+    args=[
+      Argument("checkbox", "/country/iso_code"),
+      Argument("checkbox", "/country/names/en"),
+      Argument("checkbox", "/subdivisions[]/names/en"),
+      Argument("checkbox", "/subdivisions[]/iso_code"),
+      Argument("checkbox", "/city/names/en"),
+      Argument("checkbox", "/postal/code"),
+      Argument("checkbox", "/location/latitude"),
+      Argument("checkbox", "/location/longitude"),
+    ],
+    output_type="checkbox_fields"
+  ),
+  Operator(
+    name="translate",
+    args=[
+      Argument("text", "default"),
+      Argument("mapping", "mapping")
+    ],
+    output_type="inplace"
+  ),
+  Operator(
+    name="find_replace",
+    args=[
+      Argument("text", "find"),
+      Argument("text", "replace")
+    ],
+    output_type="inplace"
+  ),
+]
+
+def _get_operator(operation_name):
+  return [operation for operation in OPERATORS if operation.name == operation_name][0]
+
+def get_checked_args(operation):
+  operation_args = _get_operator(operation["type"]).args
+
+  kept_args = [arg for arg in operation_args if operation['settings'][arg.name]]
+
+  return kept_args

+ 13 - 288
desktop/libs/indexer/src/indexer/smart_indexer.py

@@ -14,10 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.import logging
 import os
-import csv
-import operator
-import itertools
-import re
 import logging
 
 from mako.lookup import TemplateLookup
@@ -27,6 +23,9 @@ from liboozie.oozie_api import get_oozie
 from oozie.models2 import Job
 from liboozie.submission2 import Submission
 
+from indexer.fields import Field, FIELD_TYPES
+from indexer.operations import get_checked_args
+from indexer.file_format import get_file_format_instance
 from indexer.conf import CONFIG_INDEXING_TEMPLATES_PATH
 from indexer.conf import CONFIG_INDEXER_LIBS_PATH
 from indexer.conf import zkensemble
@@ -108,12 +107,12 @@ class Indexer(object):
           ]
     }
     """
-    file_format = FileFormat.get_instance(data['file'])
+    file_format = get_file_format_instance(data['file'])
     return file_format.get_format()
 
   def guess_field_types(self, data):
-    file_format = FileFormat.get_instance(data['file'], data['format'])
-    return file_format.get_fields()
+    file_format = get_file_format_instance(data['file'], data['format'])
+    return file_format.get_fields() if file_format else {'columns':[]}
 
   # Breadth first ordering of fields
   def get_field_list(self, field_data):
@@ -155,7 +154,7 @@ class Indexer(object):
 
   @staticmethod
   def _get_regex_for_type(type_):
-    matches = filter(lambda field_type: field_type.name == type_, Field.TYPES)
+    matches = filter(lambda field_type: field_type.name == type_, FIELD_TYPES)
 
     return matches[0].regex.replace('\\', '\\\\')
 
@@ -173,6 +172,9 @@ class Indexer(object):
     Morphline content 'SOLR_LOCATOR : { ...}'
     """
 
+    geolite_loc = os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "GeoLite2-City.mmdb")
+    grok_dicts_loc = os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "grok_dictionaries")
+
     properties = {
       "collection_name":collection_name,
       "fields":self.get_field_list(data['columns']),
@@ -181,7 +183,9 @@ class Indexer(object):
       "uuid_name" : uuid_name,
       "get_regex":Indexer._get_regex_for_type,
       "format":data['format'],
-      "grok_dictionaries_location" : os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "grok_dictionaries"),
+      "get_kept_args": get_checked_args,
+      "grok_dictionaries_location" : grok_dicts_loc if self.fs.exists(geolite_loc) else None,
+      "geolite_db_location" : geolite_loc if self.fs.exists(geolite_loc) else None,
       "zk_host": zkensemble()
     }
 
@@ -192,282 +196,3 @@ class Indexer(object):
 
     return morphline
 
-class FieldType():
-  def __init__(self, name, regex):
-    self._name = name
-    self._regex = regex
-
-  @property
-  def name(self):
-    return self._name
-
-  @property
-  def regex(self):
-    return self._regex
-  
- 
-  def matches(self, field):
-    pattern = re.compile(self._regex)
-
-    return pattern.match(field)
-
-class Operator():
-  def __init__(self, name, args):
-    self._name = name
-    self._args = args
-
-  def to_dict(self):
-    return {
-      "name": self._name,
-      "args": self._args
-    }
-
-class Field(object):
-  TYPES = [
-    FieldType('text', "^.{100,}$"),
-    FieldType('string', "^.*$"),
-    FieldType('double', "^[+-]?[0-9]+\\.?[0-9]+$"),
-    FieldType('long', "^(?:[+-]?(?:[0-9]+))$"),
-    FieldType('date', "[0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+(\\.[0-9]*)?Z")
-  ]
-
-  OPERATORS = [
-    Operator(
-      name="split",
-      args=["splitChar"]
-      ),
-    Operator(
-      name="grok",
-      args=["regexp"]
-      ),
-    Operator(
-      name="convert_date",
-      args=["format"]
-      ),
-  ]
-
-  def __init__(self, name, field_type):
-    self._name = name
-    self._field_type = field_type
-    self._keep = True
-    self._operations = []
-    self._required = True
-
-  @staticmethod
-  def guess_type(samples):
-    guesses = [Field._guess_field_type(sample) for sample in samples]
-
-    return Field._pick_best(guesses)
-
-  @staticmethod
-  def _guess_field_type(field):
-    for field_type in Field.TYPES[::-1]:
-      if field_type.matches(field):
-        return field_type.name
-
-  @staticmethod
-  def _pick_best(types):
-    types = set(types)
-
-    for field in Field.TYPES:
-      if field.name in types:
-        return field.name
-    return "string"
-
-  @property
-  def required(self):
-    return self._required
-
-  @property
-  def name(self):
-    return self._name
-
-  @property
-  def field_type(self):
-    return self._field_type
-
-  @property
-  def keep(self):
-    return self._keep
-
-  @property
-  def operations(self):
-    return self._operations
-
-  def to_dict(self):
-    return {'name': self.name,
-    'type': self.field_type,
-    'keep': self.keep,
-    'operations': self.operations,
-    'required': self.required}
-
-class FileFormat(object):
-  @staticmethod
-  def get_instance(file_stream, format_=None):
-    return CSVFormat(file_stream, format_)
-
-  def __init__(self):
-    pass
-
-  @property
-  def format_(self):
-    pass
-
-  @property
-  def sample(self):
-    pass
-
-  @property
-  def fields(self):
-    return []
-
-  def get_format(self):
-    return self.format_
-
-  def get_fields(self):
-    obj = {}
-
-    obj['columns'] = [field.to_dict() for field in self.fields]
-    obj['sample'] = self.sample
-
-    return obj
-
-  def to_dict(self):
-    obj = {}
-
-    obj['format'] = self.format_
-    obj['columns'] = [field.to_dict() for field in self.fields]
-    obj['sample'] = self.sample
-
-    return obj
-
-class CSVFormat(FileFormat):
-  def __init__(self, file_stream, format_=None):
-    file_stream.seek(0)
-    sample = file_stream.read(1024*1024*5)
-    file_stream.seek(0)
-
-    if format_:
-      self._delimiter = format_["fieldSeparator"].encode('utf-8')
-      self._line_terminator = format_["recordSeparator"].encode('utf-8')
-      self._quote_char = format_["quoteChar"].encode('utf-8')
-      self._has_header = format_["hasHeader"]
-    else:
-      dialect, self._has_header = self._guess_dialect(sample)
-      self._delimiter = dialect.delimiter
-      self._line_terminator = dialect.lineterminator
-      self._quote_char = dialect.quotechar
-
-    # sniffer insists on \r\n even when \n. This is safer and good enough for a preview
-    self._line_terminator = self._line_terminator.replace("\r\n", "\n")
-
-    self._sample_rows = self._get_sample_rows(sample)
-    self._num_columns = self._guess_num_columns(self._sample_rows)
-
-    self._fields = self._guess_fields(sample)
-
-    super(CSVFormat, self).__init__()
-
-  @property
-  def sample(self):
-    return self._sample_rows
-
-  @property
-  def fields(self):
-    return self._fields
-
-  @property
-  def delimiter(self):
-    return self._delimiter
-
-  @property
-  def line_terminator(self):
-    return self._line_terminator
-
-  @property
-  def quote_char(self):
-    return self._quote_char
-
-  @property
-  def format_(self):
-    return {
-      "type":"csv",
-      "fieldSeparator":self.delimiter,
-      "recordSeparator":self.line_terminator,
-      "quoteChar":self.quote_char,
-      "hasHeader":self._has_header
-    }
-
-  def _guess_dialect(self, sample):
-    sniffer = csv.Sniffer()
-    dialect = sniffer.sniff(sample)
-    has_header = sniffer.has_header(sample)
-    return dialect, has_header
-
-  def _guess_num_columns(self, sample_rows):
-    counts = {}
-
-    for row in sample_rows:
-      num_columns = len(row)
-
-      if num_columns not in counts:
-        counts[num_columns] = 0
-      counts[num_columns] += 1
-
-    if counts:
-      num_columns_guess = max(counts.iteritems(), key=operator.itemgetter(1))[0]
-    else:
-      num_columns_guess = 0
-    return num_columns_guess
-
-  def _guess_field_types(self, sample_rows):
-    field_type_guesses = []
-
-    num_columns = self._num_columns
-
-    for col in range(num_columns):
-      column_samples = [sample_row[col] for sample_row in sample_rows if len(sample_row) > col]
-
-      field_type_guess = Field.guess_type(column_samples)
-      field_type_guesses.append(field_type_guess)
-
-    return field_type_guesses
-
-  def _get_sample_reader(self, sample):
-    if self.line_terminator != '\n':
-      sample = sample.replace('\n', '\\n')
-    return csv.reader(sample.split(self.line_terminator), delimiter=self.delimiter, quotechar=self.quote_char)
-
-  def _guess_field_names(self, sample):
-    reader = self._get_sample_reader(sample)
-
-    first_row = reader.next()
-
-    if self._has_header:
-      header = first_row
-    else:
-      header = ["field_%d" % (i+1) for i in range(self._num_columns)]
-
-    return header
-
-  def _get_sample_rows(self, sample):
-    NUM_SAMPLES = 5
-
-    header_offset = 1 if self._has_header else 0
-    reader = itertools.islice(self._get_sample_reader(sample), header_offset, NUM_SAMPLES + 1)
-
-    sample_rows = list(reader)
-    return sample_rows
-
-  def _guess_fields(self, sample):
-    header = self._guess_field_names(sample)
-    types = self._guess_field_types(self._sample_rows)
-
-    if len(header) == len(types):
-      fields = [Field(header[i], types[i]) for i in range(len(header))]
-    else:
-      # likely failed to guess correctly
-      LOG.warn("Guess field types failed - number of headers didn't match number of predicted types.")
-      fields = []
-
-    return fields

+ 122 - 37
desktop/libs/indexer/src/indexer/templates/indexer.mako

@@ -36,12 +36,13 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
   <div class="snippet-settings" data-bind="visible: createWizard.show" style="
   text-align: center;">
 
-    <div class="control-group" data-bind="css: { error: createWizard.validName() === false, success: createWizard.validName()}">
+    <div class="control-group" data-bind="css: { error: createWizard.isNameAvailable() === false, success: createWizard.isNameAvailable()}">
       <label for="collectionName" class="control-label">${ _('Name') }</label>
       <div class="controls">
         <input type="text" class="form-control" id = "collectionName" data-bind="value: createWizard.fileFormat().name, valueUpdate: 'afterkeydown'">
-        <span class="help-block" data-bind="visible: createWizard.validName() === true">${ _('Collection name available') }</span>
-        <span class="help-block" data-bind="visible: createWizard.validName() === false">${_('This collection already exists') }</span>
+        <span class="help-block" data-bind="visible: createWizard.isNameAvailable() === true">${ _('Collection name available') }</span>
+        <span class="help-block" data-bind="visible: createWizard.isNameAvailable() === false && createWizard.fileFormat().name().length > 0">${_('This collection already exists') }</span>
+        <span class="help-block" data-bind="visible: createWizard.isNameAvailable() === false && createWizard.fileFormat().name().length == 0">${_('This collection needs a name') }</span>
       </div>
     </div>
 
@@ -82,21 +83,25 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
           </thead>
           <tbody data-bind="foreach: createWizard.sample">
             <tr data-bind="foreach: $data">
-              <td data-bind="visible: $root.createWizard.fileFormat().columns()[$index()].keep, text: $data">
-              </td>
+              <!-- ko if: $index() < $root.createWizard.fileFormat().columns().length -->
+                <td data-bind="visible: $root.createWizard.fileFormat().columns()[$index()].keep, text: $data">
+                </td>
 
                 <!-- ko with: $root.createWizard.fileFormat().columns()[$index()] -->
                   <!-- ko template: 'output-generated-field-data-template' --> <!-- /ko -->
                 <!-- /ko -->
+              <!-- /ko -->
             </tr>
           </tbody>
         </table>
 
         <br><hr><br>
 
-        <a href="javascript:void(0)" class="btn" data-bind="visible: !createWizard.indexingStarted() , click: createWizard.indexFile, css: {disabled : !createWizard.validName()}">${_('Index File!')}</a>
+        <a href="javascript:void(0)" class="btn" data-bind="visible: !createWizard.indexingStarted() , click: createWizard.indexFile, css: {disabled : !createWizard.readyToIndex()}">${_('Index File!')}</a>
+
+        <h4 class="error" data-bind="visible: !createWizard.isNameAvailable() && createWizard.fileFormat().name().length > 0">${_('Collection needs a unique name')}</h4>
+        <h4 class="error" data-bind="visible: !createWizard.isNameAvailable() && createWizard.fileFormat().name().length == 0">${_('Collection needs a name')}</h4>
 
-        <h4 class="error" data-bind="visible: !createWizard.validName()">${_('Collection needs a unique name')}</h4>
 
         <a href="javascript:void(0)" class="btn btn-success" data-bind="visible: createWizard.jobId, attr: {           href: '/oozie/list_oozie_workflow/' + createWizard.jobId() }" target="_blank" title="${ _('Open') }">
           ${_('View Indexing Status')}
@@ -123,7 +128,9 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
 <script type="text/html" id="operation-template">
   <div><select data-bind="options: $root.createWizard.operationTypes.map(function(o){return o.name});, value: operation.type"></select>
   <!-- ko template: "operation-args-template" --><!-- /ko -->
-    <input type="number" data-bind="value: operation.numExpectedFields">
+    <!-- ko if: operation.settings().outputType() == "custom_fields" -->
+      <input type="number" data-bind="value: operation.numExpectedFields">
+    <!-- /ko -->
     <button class="btn" data-bind="click: function(){$root.createWizard.removeOperation(operation, list)}">${_('remove')}</button>
     <div style="padding-left:50px" data-bind="foreach: operation.fields">
       <div data-bind="template: { name:'field-template',data:$data}"></div>
@@ -153,11 +160,32 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
 </script>
 
 <script type="text/html" id="operation-args-template">
-  <!-- ko foreach: Object.keys(operation.settings()) -->
-  <input type="text" data-bind="value: $parent.operation.settings()[$data]">
+  <!-- ko foreach: {data: operation.settings().getArguments(), as: 'argument'} -->
+    <!-- ko template: {name: 'operation-arg-'+argument.type, data:{operation: $parent.operation, argVal: $parent.operation.settings()[argument.name]}}--><!-- /ko -->
   <!-- /ko -->
+
 </script>
 
+<script type="text/html" id="operation-arg-text">
+  <input type="text" data-bind="attr: {placeholder: argument.name}, value: argVal">
+</script>
+
+<script type="text/html" id="operation-arg-checkbox">
+  <h4 data-bind="text: argument.name"></h4>
+  <input type="checkbox" data-bind="checked: argVal">
+</script>
+
+<script type="text/html" id="operation-arg-mapping">
+  <!-- ko foreach: argVal-->
+    <div>
+      <input type="text" data-bind="value: key, attr: {placeholder: 'key'}">
+      <input type="text" data-bind="value: value, attr: {placeholder: 'value'}">
+      <button class="btn" data-bind="click: function(){$parent.operation.settings().mapping.remove($data)}">${_('Remove Pair')}</button>
+    </div>
+  <!-- /ko -->
+  <button class="btn" data-bind="click: operation.addPair">${_('Add Pair')}</button>
+  <br>
+</script>
 
 <div class="hueOverlay" data-bind="visible: isLoading">
   <!--[if lte IE 9]>
@@ -175,9 +203,16 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
 
 
 <script type="text/javascript" charset="utf-8">
+var fieldNum = 0;
+
+var getNewFieldName = function(){
+  fieldNum++;
+  return "new_field_" + fieldNum
+}
+
   var createDefaultField = function(){
     return {
-      name: ko.observable("new_field"),
+      name: ko.observable(getNewFieldName()),
       type: ko.observable("string"),
       keep: ko.observable(true),
       required: ko.observable(true),
@@ -186,43 +221,89 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
   };
 
   var Operation = function(type){
+    var self = this;
+
+    var createArgumentValue = function(arg){
+      if(arg.type == "mapping"){
+        return ko.observableArray([]);
+      }
+      else if(arg.type =="checkbox"){
+        return ko.observable(false);
+      }
+      else{
+        return ko.observable("");
+      }
+    }
+
     var constructSettings = function(type){
       var settings = {};
+
       var operation = viewModel.createWizard.operationTypes.find(function(currOperation){
         return currOperation.name == type;
       });
 
       for(var i = 0; i < operation.args.length; i++){
-        settings[operation.args[i]] = ko.observable(operation.args[i]);
+        argVal = createArgumentValue(operation.args[i]);
+
+        if(operation.args[i].type == "checkbox" && operation.outputType == "checkbox_fields"){
+          argVal.subscribe(function(newVal){
+            if(newVal){
+              self.fields.push(createDefaultField());
+            }
+            else{
+              self.fields.pop();
+            }
+          });
+        }
+
+        settings[operation.args[i].name] = argVal;
+      }
+
+      settings.getArguments = function(){
+        return operation.args
+      };
+
+      settings.outputType= function(){
+        return operation.outputType;
       }
+
       return settings;
     };
 
-    var self = this;
+    var init = function(){
+      self.fields([]);
+      self.numExpectedFields(0);
 
-    self.type = ko.observable(type);
-    self.fields = ko.observableArray();
+      self.numExpectedFields.subscribe(function(numExpectedFields){
+        if(numExpectedFields < self.fields().length){
+          self.fields(self.fields().slice(0,numExpectedFields));
+        }
+        else if (numExpectedFields > self.fields().length){
+          difference = numExpectedFields - self.fields().length;
 
-    self.numExpectedFields = ko.observable(0);
+          for(var i = 0; i < difference; i++){
+            self.fields.push(createDefaultField());
+          }
+        }
+      });
 
-    self.numExpectedFields.subscribe(function(numExpectedFields){
-      if(numExpectedFields < self.fields().length){
-        self.fields(self.fields().slice(0,numExpectedFields));
-      }
-      else if (numExpectedFields > self.fields().length){
-        difference = numExpectedFields - self.fields().length;
+      self.settings(constructSettings(self.type()));
+    }
 
-        for(var i = 0; i < difference; i++){
-          self.fields.push(createDefaultField());
-        }
-      }
-    });
+    self.type = ko.observable(type);
+    self.fields = ko.observableArray();
+    self.numExpectedFields = ko.observable();
+    self.settings = ko.observable();
 
-    self.settings = ko.observable(constructSettings(type));
+    init();
 
     self.type.subscribe(function(newType){
-      self.settings(constructSettings(newType));
+      init();
     });
+
+    self.addPair = function(){
+      self.settings().mapping.push({key: ko.observable(""), value: ko.observable("")});
+    }
   }
 
   var File_Format = function (vm) {
@@ -230,7 +311,6 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
 
 
     self.name = ko.observable('');
-    self.sample = ko.observableArray();
     self.show = ko.observable(false);
 
     self.path = ko.observable('/tmp/test.csv');
@@ -246,8 +326,6 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
 
     self.fieldTypes = ko.observableArray(${fields_json | n});
 
-    self.validName = ko.observable(null);
-
     self.show = ko.observable(true);
     self.showCreate = ko.observable(false);
 
@@ -259,6 +337,17 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
 
     self.indexingStarted = ko.observable(false);
 
+    self.isNameAvailable = ko.computed(function(){
+      var name = self.fileFormat().name();
+      return viewModel && viewModel.collectionNameAvailable(name) && name.length > 0;
+    });
+
+    self.readyToIndex = ko.computed(function(){
+      var validFields = self.fileFormat().columns().length
+
+      return self.isNameAvailable() && validFields;
+    });
+
     self.fileFormat().format.subscribe(function(){
       self.fileFormat().format().quoteChar.subscribe(self.guessFieldTypes);
       self.fileFormat().format().recordSeparator.subscribe(self.guessFieldTypes);
@@ -269,10 +358,6 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
       self.guessFieldTypes();
     });
 
-    self.fileFormat().name.subscribe(function(newName){
-      self.validName(viewModel.collectionNameAvailable(newName));
-    })
-
     self.guessFormat = function() {
       viewModel.isLoading(true);
       $.post("${ url('indexer:guess_format') }", {
@@ -310,7 +395,7 @@ ${ commonheader(_("Solr Indexes"), "search", user, "60px") | n,unicode }
     };
 
     self.indexFile = function() {
-      if(!self.validName()) return;
+      if(!self.readyToIndex()) return;
 
       self.indexingStarted(true);
 

+ 29 - 5
desktop/libs/indexer/src/indexer/tests_indexer.py

@@ -40,7 +40,7 @@ class IndexerTest():
 
   def test_guess_format(self):
     stream = StringIO.StringIO(IndexerTest.simpleCSVString)
-    indexer = Indexer("hue", None)
+    indexer = Indexer("test", None)
 
     guessed_format = indexer.guess_format({'file': stream})
 
@@ -74,13 +74,37 @@ class IndexerTest():
       }
     ]
 
-    for i in range(len(expected_fields)):
-      expected = expected_fields[i]
-      actual = fields[i]
-
+    for expected, actual in zip(expected_fields, fields):
       for key in ("name", "type"):
         assert_equal(expected[key], actual[key])
 
+  def test_guess_format_invalid_csv_format(self):
+    indexer = Indexer("test", None)
+    stream = StringIO.StringIO(IndexerTest.simpleCSVString)
+
+    guessed_format = indexer.guess_format({'file': stream})
+
+    guessed_format["fieldSeparator"] = "invalid separator"
+
+    fields = indexer.guess_field_types({"file":stream, "format": guessed_format})['columns']
+    assert_equal(fields, [])
+
+    stream.seek(0)
+    guessed_format = indexer.guess_format({'file': stream})
+
+    guessed_format["recordSeparator"] = "invalid separator"
+
+    fields = indexer.guess_field_types({"file":stream, "format": guessed_format})['columns']
+    assert_equal(fields, [])
+
+    stream.seek(0)
+    guessed_format = indexer.guess_format({'file': stream})
+
+    guessed_format["quoteChar"] = "invalid quoteChar"
+
+    fields = indexer.guess_field_types({"file":stream, "format": guessed_format})['columns']
+    assert_equal(fields, [])
+
   def test_end_to_end(self):
     fs = cluster.get_hdfs()
     collection_name = "test_collection"

+ 4 - 3
desktop/libs/indexer/src/indexer/views.py

@@ -24,7 +24,8 @@ from desktop.lib.django_util import JsonResponse, render
 
 from indexer.controller2 import IndexController
 from indexer.management.commands import indexer_setup
-from indexer.smart_indexer import Field
+from indexer.fields import FIELD_TYPES
+from indexer.operations import OPERATORS
 
 LOG = logging.getLogger(__name__)
 
@@ -53,8 +54,8 @@ def indexer(request):
 
   return render('indexer.mako', request, {
       'indexes_json': json.dumps(indexes),
-      'fields_json' : json.dumps([field.name for field in Field.TYPES]),
-      'operators_json' : json.dumps([operator.to_dict() for operator in Field.OPERATORS])
+      'fields_json' : json.dumps([field.name for field in FIELD_TYPES]),
+      'operators_json' : json.dumps([operator.to_dict() for operator in OPERATORS])
   })
 
 def install_examples(request, is_redirect=False):