diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/bro_index.template b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/bro_index.template
index 7db006e..3a68d75 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/bro_index.template
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/bro_index.template
@@ -98,7 +98,7 @@
           "mapping": {
             "type": "float"
           },
-          "match": "threat.triage.rules:*:score",
+          "match": "threat:triage:*score",
           "match_mapping_type": "*"
         }
       },
@@ -107,7 +107,7 @@
           "mapping": {
             "type": "string"
           },
-          "match": "threat.triage.rules:*:reason",
+          "match": "threat:triage:rules:*:reason",
           "match_mapping_type": "*"
         }
       },
@@ -116,9 +116,9 @@
           "mapping": {
             "type": "string"
           },
-          "match": "threat.triage.rules:*:name",
+          "match": "threat:triage:rules:*:name",
           "match_mapping_type": "*"
-        }
+      }
       }
       ],
       "properties": {
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/snort_index.template b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/snort_index.template
index f13a9ee..7c6b401 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/snort_index.template
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/snort_index.template
@@ -98,28 +98,28 @@
           "mapping": {
             "type": "float"
           },
-          "match": "threat.triage.rules:*:score",
+          "match": "threat:triage:*score",
           "match_mapping_type": "*"
         }
       },
-        {
-          "threat_triage_reason": {
-            "mapping": {
-              "type": "string"
-            },
-            "match": "threat.triage.rules:*:reason",
-            "match_mapping_type": "*"
-          }
-        },
-        {
-          "threat_triage_name": {
-            "mapping": {
-              "type": "string"
-            },
-            "match": "threat.triage.rules:*:name",
-            "match_mapping_type": "*"
-          }
+      {
+        "threat_triage_reason": {
+          "mapping": {
+            "type": "string"
+          },
+          "match": "threat:triage:rules:*:reason",
+          "match_mapping_type": "*"
         }
+      },
+      {
+        "threat_triage_name": {
+          "mapping": {
+            "type": "string"
+          },
+          "match": "threat:triage:rules:*:name",
+          "match_mapping_type": "*"
+        }
+      }
       ],
       "properties": {
         "timestamp": {
@@ -195,9 +195,6 @@
         "tcpwindow": {
           "type": "string"
         },
-        "threat:triage:level": {
-          "type": "double"
-        },
         "tos": {
           "type": "integer"
         },
diff --git a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/yaf_index.template b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/yaf_index.template
index d84235d..d100eb0 100644
--- a/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/yaf_index.template
+++ b/metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/files/yaf_index.template
@@ -98,7 +98,7 @@
           "mapping": {
             "type": "float"
           },
-          "match": "threat.triage.rules:*:score",
+          "match": "threat:triage:*score",
           "match_mapping_type": "*"
         }
       },
@@ -107,7 +107,7 @@
           "mapping": {
             "type": "string"
           },
-          "match": "threat.triage.rules:*:reason",
+          "match": "threat:triage:rules:*:reason",
           "match_mapping_type": "*"
         }
       },
@@ -116,7 +116,7 @@
           "mapping": {
             "type": "string"
           },
-          "match": "threat.triage.rules:*:name",
+          "match": "threat:triage:rules:*:name",
           "match_mapping_type": "*"
         }
       }
diff --git a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java
index 4ce9644..25bb809 100644
--- a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java
+++ b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/config/IndexConfig.java
@@ -34,6 +34,10 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
 
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
 @Configuration
 public class IndexConfig {
 
@@ -57,6 +61,7 @@
       int searchMaxGroups = environment.getProperty(MetronRestConstants.SEARCH_MAX_GROUPS, Integer.class, 1000);
       String metaDaoImpl = environment.getProperty(MetronRestConstants.META_DAO_IMPL, String.class, null);
       String metaDaoSort = environment.getProperty(MetronRestConstants.META_DAO_SORT, String.class, null);
+
       AccessConfig config = new AccessConfig();
       config.setMaxSearchResults(searchMaxResults);
       config.setMaxSearchGroups(searchMaxGroups);
@@ -84,6 +89,7 @@
       MetaAlertDao ret = (MetaAlertDao) IndexDaoFactory.create(metaDaoImpl, config).get(0);
       ret.init(indexDao, Optional.ofNullable(metaDaoSort));
       return ret;
+
     }
     catch(RuntimeException re) {
       throw re;
diff --git a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/controller/SearchControllerIntegrationTest.java b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/controller/SearchControllerIntegrationTest.java
index 3673654..78a1e20 100644
--- a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/controller/SearchControllerIntegrationTest.java
+++ b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/controller/SearchControllerIntegrationTest.java
@@ -89,8 +89,8 @@
   public void setup() throws Exception {
     this.mockMvc = MockMvcBuilders.webAppContextSetup(this.wac).apply(springSecurity()).build();
     ImmutableMap<String, String> testData = ImmutableMap.of(
-        "bro_index_2017.01.01.01", SearchIntegrationTest.broData,
-        "snort_index_2017.01.01.01", SearchIntegrationTest.snortData
+            "bro_index_2017.01.01.01", SearchIntegrationTest.broData,
+            "snort_index_2017.01.01.01", SearchIntegrationTest.snortData
     );
     loadTestData(testData);
     loadColumnTypes();
@@ -114,19 +114,19 @@
     }});
 
     assertEventually(() -> this.mockMvc.perform(post(searchUrl + "/search").with(httpBasic(user, password)).with(csrf()).contentType(MediaType.parseMediaType("application/json;charset=UTF-8")).content(defaultQuery))
-        .andExpect(status().isOk())
-        .andExpect(content().contentType(MediaType.parseMediaType("application/json;charset=UTF-8")))
-        .andExpect(jsonPath("$.total").value(5))
-        .andExpect(jsonPath("$.results[0].source.source:type").value("bro"))
-        .andExpect(jsonPath("$.results[0].source.timestamp").value(5))
-        .andExpect(jsonPath("$.results[1].source.source:type").value("bro"))
-        .andExpect(jsonPath("$.results[1].source.timestamp").value(4))
-        .andExpect(jsonPath("$.results[2].source.source:type").value("bro"))
-        .andExpect(jsonPath("$.results[2].source.timestamp").value(3))
-        .andExpect(jsonPath("$.results[3].source.source:type").value("bro"))
-        .andExpect(jsonPath("$.results[3].source.timestamp").value(2))
-        .andExpect(jsonPath("$.results[4].source.source:type").value("bro"))
-        .andExpect(jsonPath("$.results[4].source.timestamp").value(1))
+            .andExpect(status().isOk())
+            .andExpect(content().contentType(MediaType.parseMediaType("application/json;charset=UTF-8")))
+            .andExpect(jsonPath("$.total").value(5))
+            .andExpect(jsonPath("$.results[0].source.source:type").value("bro"))
+            .andExpect(jsonPath("$.results[0].source.timestamp").value(5))
+            .andExpect(jsonPath("$.results[1].source.source:type").value("bro"))
+            .andExpect(jsonPath("$.results[1].source.timestamp").value(4))
+            .andExpect(jsonPath("$.results[2].source.source:type").value("bro"))
+            .andExpect(jsonPath("$.results[2].source.timestamp").value(3))
+            .andExpect(jsonPath("$.results[3].source.source:type").value("bro"))
+            .andExpect(jsonPath("$.results[3].source.timestamp").value(2))
+            .andExpect(jsonPath("$.results[4].source.source:type").value("bro"))
+            .andExpect(jsonPath("$.results[4].source.timestamp").value(1))
     );
 
     sensorIndexingConfigService.delete("bro");
@@ -288,4 +288,4 @@
     columnTypes.put("snort", snortTypes);
     InMemoryDao.setColumnMetadata(columnTypes);
   }
-}
+}
\ No newline at end of file
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ColumnMetadataDao.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ColumnMetadataDao.java
new file mode 100644
index 0000000..0393629
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ColumnMetadataDao.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.metron.elasticsearch.dao;
+
+import org.apache.metron.indexing.dao.search.FieldType;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Responsible for retrieving column-level metadata about search indices.
+ */
+public interface ColumnMetadataDao {
+
+  /**
+   * Retrieves column metadata for one or more search indices.
+   * @param indices The search indices to retrieve column metadata for.
+   * @return The column metadata, one set for each search index.
+   * @throws IOException
+   */
+  Map<String, FieldType> getColumnMetadata(List<String> indices) throws IOException;
+
+  /**
+   * Finds the latest version of a set of base indices.  This can be used to find
+   * the latest 'bro' index, for example.
+   *
+   * Assuming the following indices exist...
+   *
+   *    [
+   *      'bro_index_2017.10.03.19'
+   *      'bro_index_2017.10.03.20',
+   *      'bro_index_2017.10.03.21',
+   *      'snort_index_2017.10.03.19',
+   *      'snort_index_2017.10.03.20',
+   *      'snort_index_2017.10.03.21'
+   *    ]
+   *
+   *  And the include indices are given as...
+   *
+   *    ['bro', 'snort']
+   *
+   * Then the latest indices are...
+   *
+   *    ['bro_index_2017.10.03.21', 'snort_index_2017.10.03.21']
+   *
+   * @param includeIndices The base names of the indices to include
+   * @return The latest version of a set of indices.
+   */
+  String[] getLatestIndices(List<String> includeIndices);
+}
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchColumnMetadataDao.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchColumnMetadataDao.java
new file mode 100644
index 0000000..8e210b4
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchColumnMetadataDao.java
@@ -0,0 +1,179 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.metron.elasticsearch.dao;
+
+import org.apache.metron.indexing.dao.search.FieldType;
+import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
+import org.elasticsearch.client.AdminClient;
+import org.elasticsearch.cluster.metadata.MappingMetaData;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.metron.elasticsearch.utils.ElasticsearchUtils.INDEX_NAME_DELIMITER;
+
+/**
+ * Responsible for retrieving column-level metadata for Elasticsearch search indices.
+ */
+public class ElasticsearchColumnMetadataDao implements ColumnMetadataDao {
+
+  private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+  private static Map<String, FieldType> elasticsearchTypeMap;
+  static {
+    Map<String, FieldType> fieldTypeMap = new HashMap<>();
+    fieldTypeMap.put("string", FieldType.STRING);
+    fieldTypeMap.put("ip", FieldType.IP);
+    fieldTypeMap.put("integer", FieldType.INTEGER);
+    fieldTypeMap.put("long", FieldType.LONG);
+    fieldTypeMap.put("date", FieldType.DATE);
+    fieldTypeMap.put("float", FieldType.FLOAT);
+    fieldTypeMap.put("double", FieldType.DOUBLE);
+    fieldTypeMap.put("boolean", FieldType.BOOLEAN);
+    elasticsearchTypeMap = Collections.unmodifiableMap(fieldTypeMap);
+  }
+
+  /**
+   * An Elasticsearch administrative client.
+   */
+  private transient AdminClient adminClient;
+
+  /**
+   * @param adminClient The Elasticsearch admin client.
+   */
+  public ElasticsearchColumnMetadataDao(AdminClient adminClient) {
+    this.adminClient = adminClient;
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public Map<String, FieldType> getColumnMetadata(List<String> indices) throws IOException {
+    Map<String, FieldType> indexColumnMetadata = new HashMap<>();
+    Map<String, String> previousIndices = new HashMap<>();
+    Set<String> fieldBlackList = new HashSet<>();
+
+    String[] latestIndices = getLatestIndices(indices);
+    if (latestIndices.length > 0) {
+      ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings = adminClient
+              .indices()
+              .getMappings(new GetMappingsRequest().indices(latestIndices))
+              .actionGet()
+              .getMappings();
+
+      // for each index
+      for (Object key : mappings.keys().toArray()) {
+        String indexName = key.toString();
+        ImmutableOpenMap<String, MappingMetaData> mapping = mappings.get(indexName);
+
+        // for each mapping in the index
+        Iterator<String> mappingIterator = mapping.keysIt();
+        while (mappingIterator.hasNext()) {
+          MappingMetaData mappingMetaData = mapping.get(mappingIterator.next());
+          Map<String, Map<String, String>> map = (Map<String, Map<String, String>>) mappingMetaData
+                  .getSourceAsMap().get("properties");
+
+          // for each field in the mapping
+          for (String field : map.keySet()) {
+            if (!fieldBlackList.contains(field)) {
+              FieldType type = toFieldType(map.get(field).get("type"));
+
+              if(!indexColumnMetadata.containsKey(field)) {
+                indexColumnMetadata.put(field, type);
+
+                // record the last index in which a field exists, to be able to print helpful error message on type mismatch
+                previousIndices.put(field, indexName);
+
+              } else {
+                FieldType previousType = indexColumnMetadata.get(field);
+                if (!type.equals(previousType)) {
+                  String previousIndexName = previousIndices.get(field);
+                  LOG.error(String.format(
+                          "Field type mismatch: %s.%s has type %s while %s.%s has type %s.  Defaulting type to %s.",
+                          indexName, field, type.getFieldType(),
+                          previousIndexName, field, previousType.getFieldType(),
+                          FieldType.OTHER.getFieldType()));
+                  indexColumnMetadata.put(field, FieldType.OTHER);
+
+                  // the field is defined in multiple indices with different types; ignore the field as type has been set to OTHER
+                  fieldBlackList.add(field);
+                }
+              }
+            }
+          }
+        }
+      }
+    } else {
+      LOG.info(String.format("Unable to find any latest indices; indices=%s", indices));
+    }
+
+    return indexColumnMetadata;
+
+  }
+
+  /**
+   * Retrieves the latest indices.
+   * @param includeIndices
+   * @return
+   */
+  @Override
+  public String[] getLatestIndices(List<String> includeIndices) {
+    LOG.debug("Getting latest indices; indices={}", includeIndices);
+    Map<String, String> latestIndices = new HashMap<>();
+    String[] indices = adminClient
+            .indices()
+            .prepareGetIndex()
+            .setFeatures()
+            .get()
+            .getIndices();
+
+    for (String index : indices) {
+      int prefixEnd = index.indexOf(INDEX_NAME_DELIMITER);
+      if (prefixEnd != -1) {
+        String prefix = index.substring(0, prefixEnd);
+        if (includeIndices.contains(prefix)) {
+          String latestIndex = latestIndices.get(prefix);
+          if (latestIndex == null || index.compareTo(latestIndex) > 0) {
+            latestIndices.put(prefix, index);
+          }
+        }
+      }
+    }
+
+    return latestIndices.values().toArray(new String[latestIndices.size()]);
+  }
+
+  /**
+   * Converts a string type to the corresponding FieldType.
+   * @param type The type to convert.
+   * @return The corresponding FieldType or FieldType.OTHER, if no match.
+   */
+  private FieldType toFieldType(String type) {
+    return elasticsearchTypeMap.getOrDefault(type, FieldType.OTHER);
+  }
+}
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
index 87ad7f7..910c09b 100644
--- a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchDao.java
@@ -17,26 +17,8 @@
  */
 package org.apache.metron.elasticsearch.dao;
 
-import static org.apache.metron.elasticsearch.utils.ElasticsearchUtils.INDEX_NAME_DELIMITER;
-
 import com.google.common.base.Splitter;
 import com.google.common.collect.Iterables;
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.function.Function;
-import java.util.stream.Collectors;
 import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
 import org.apache.metron.indexing.dao.AccessConfig;
 import org.apache.metron.indexing.dao.IndexDao;
@@ -52,19 +34,16 @@
 import org.apache.metron.indexing.dao.search.SearchRequest;
 import org.apache.metron.indexing.dao.search.SearchResponse;
 import org.apache.metron.indexing.dao.search.SearchResult;
+import org.apache.metron.indexing.dao.search.SortField;
 import org.apache.metron.indexing.dao.search.SortOrder;
 import org.apache.metron.indexing.dao.update.Document;
 import org.elasticsearch.action.ActionWriteResponse.ShardInfo;
-import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.action.search.SearchPhaseExecutionException;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.client.transport.TransportClient;
-import org.elasticsearch.cluster.metadata.MappingMetaData;
-import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.index.mapper.ip.IpFieldMapper;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -80,18 +59,64 @@
 import org.elasticsearch.search.aggregations.metrics.sum.Sum;
 import org.elasticsearch.search.aggregations.metrics.sum.SumBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+import static org.apache.metron.elasticsearch.utils.ElasticsearchUtils.INDEX_NAME_DELIMITER;
 
 public class ElasticsearchDao implements IndexDao {
 
   private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+  /**
+   * The value required to ensure that Elasticsearch sorts missing values last.
+   */
+  private static final String SORT_MISSING_LAST = "_last";
+
+  /**
+   * The value required to ensure that Elasticsearch sorts missing values last.
+   */
+  private static final String SORT_MISSING_FIRST = "_first";
+
+  /**
+   * The Elasticsearch client.
+   */
   private transient TransportClient client;
+
+  /**
+   * Retrieves column metadata about search indices.
+   */
+  private ColumnMetadataDao columnMetadataDao;
+
+  /**
+   * Handles the submission of search requests to Elasticsearch.
+   */
+  private ElasticsearchRequestSubmitter requestSubmitter;
+
   private AccessConfig accessConfig;
 
-  protected ElasticsearchDao(TransportClient client, AccessConfig config) {
+  protected ElasticsearchDao(TransportClient client,
+                             ColumnMetadataDao columnMetadataDao,
+                             ElasticsearchRequestSubmitter requestSubmitter,
+                             AccessConfig config) {
     this.client = client;
+    this.columnMetadataDao = columnMetadataDao;
+    this.requestSubmitter = requestSubmitter;
     this.accessConfig = config;
   }
 
@@ -99,21 +124,6 @@
     //uninitialized.
   }
 
-  private static Map<String, FieldType> elasticsearchSearchTypeMap;
-
-  static {
-    Map<String, FieldType> fieldTypeMap = new HashMap<>();
-    fieldTypeMap.put("string", FieldType.STRING);
-    fieldTypeMap.put("ip", FieldType.IP);
-    fieldTypeMap.put("integer", FieldType.INTEGER);
-    fieldTypeMap.put("long", FieldType.LONG);
-    fieldTypeMap.put("date", FieldType.DATE);
-    fieldTypeMap.put("float", FieldType.FLOAT);
-    fieldTypeMap.put("double", FieldType.DOUBLE);
-    fieldTypeMap.put("boolean", FieldType.BOOLEAN);
-    elasticsearchSearchTypeMap = Collections.unmodifiableMap(fieldTypeMap);
-  }
-
   @Override
   public SearchResponse search(SearchRequest searchRequest) throws InvalidSearchException {
     return search(searchRequest, new QueryStringQueryBuilder(searchRequest.getQuery()));
@@ -121,56 +131,139 @@
 
   /**
    * Defers to a provided {@link org.elasticsearch.index.query.QueryBuilder} for the query.
-   * @param searchRequest The request defining the parameters of the search
+   * @param request The request defining the parameters of the search
    * @param queryBuilder The actual query to be run. Intended for if the SearchRequest requires wrapping
    * @return The results of the query
    * @throws InvalidSearchException When the query is malformed or the current state doesn't allow search
    */
-  protected SearchResponse search(SearchRequest searchRequest, QueryBuilder queryBuilder) throws InvalidSearchException {
+  protected SearchResponse search(SearchRequest request, QueryBuilder queryBuilder) throws InvalidSearchException {
+    org.elasticsearch.action.search.SearchRequest esRequest;
+    org.elasticsearch.action.search.SearchResponse esResponse;
+
     if(client == null) {
       throw new InvalidSearchException("Uninitialized Dao!  You must call init() prior to use.");
     }
-    if (searchRequest.getSize() > accessConfig.getMaxSearchResults()) {
+
+    if (request.getSize() > accessConfig.getMaxSearchResults()) {
       throw new InvalidSearchException("Search result size must be less than " + accessConfig.getMaxSearchResults());
     }
-    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
+
+    esRequest = buildSearchRequest(request, queryBuilder);
+    esResponse = requestSubmitter.submitSearch(esRequest);
+    return buildSearchResponse(request, esResponse);
+  }
+
+  /**
+   * Builds an Elasticsearch search request.
+   * @param searchRequest The Metron search request.
+   * @param queryBuilder
+   * @return An Elasticsearch search request.
+   */
+  private org.elasticsearch.action.search.SearchRequest buildSearchRequest(
+          SearchRequest searchRequest,
+          QueryBuilder queryBuilder) throws InvalidSearchException {
+
+    LOG.debug("Got search request; request={}", ElasticsearchUtils.toJSON(searchRequest).orElse("???"));
+    SearchSourceBuilder searchBuilder = new SearchSourceBuilder()
             .size(searchRequest.getSize())
             .from(searchRequest.getFrom())
             .query(queryBuilder)
             .trackScores(true);
-    searchRequest.getSort().forEach(sortField -> searchSourceBuilder.sort(sortField.getField(), getElasticsearchSortOrder(sortField.getSortOrder())));
-    Optional<List<String>> fields = searchRequest.getFields();
-    if (fields.isPresent()) {
-      searchSourceBuilder.fields(fields.get());
-    } else {
-      searchSourceBuilder.fetchSource(true);
-    }
-    Optional<List<String>> facetFields = searchRequest.getFacetFields();
-    if (facetFields.isPresent()) {
-      facetFields.get().forEach(field -> searchSourceBuilder.aggregation(new TermsBuilder(getFacentAggregationName(field)).field(field)));
-    }
-    String[] wildcardIndices = wildcardIndices(searchRequest.getIndices());
-    org.elasticsearch.action.search.SearchResponse elasticsearchResponse;
+
+    // column metadata needed to understand the type of each sort field
+    Map<String, FieldType> meta;
     try {
-      elasticsearchResponse = client.search(new org.elasticsearch.action.search.SearchRequest(wildcardIndices)
-              .source(searchSourceBuilder)).actionGet();
-    } catch (SearchPhaseExecutionException e) {
-      LOG.error("Could not execute search", e);
-      throw new InvalidSearchException("Could not execute search", e);
+      meta = getColumnMetadata(searchRequest.getIndices());
+    } catch(IOException e) {
+      throw new InvalidSearchException("Unable to get column metadata", e);
     }
+
+    // handle sort fields
+    for(SortField sortField : searchRequest.getSort()) {
+
+      // what type is the sort field?
+      FieldType sortFieldType = meta.getOrDefault(sortField.getField(), FieldType.OTHER);
+
+      // sort order - if ascending missing values sorted last. otherwise, missing values sorted first
+      org.elasticsearch.search.sort.SortOrder sortOrder = getElasticsearchSortOrder(sortField.getSortOrder());
+      String missingSortOrder;
+      if(sortOrder == org.elasticsearch.search.sort.SortOrder.DESC) {
+        missingSortOrder = SORT_MISSING_LAST;
+      } else {
+        missingSortOrder = SORT_MISSING_FIRST;
+      }
+
+      // sort by the field - missing fields always last
+      FieldSortBuilder sortBy = new FieldSortBuilder(sortField.getField())
+              .order(sortOrder)
+              .missing(missingSortOrder)
+              .unmappedType(sortFieldType.getFieldType());
+      searchBuilder.sort(sortBy);
+    }
+
+    // handle search fields
+    if (searchRequest.getFields().isPresent()) {
+      searchBuilder.fields(searchRequest.getFields().get());
+    } else {
+      searchBuilder.fetchSource(true);
+    }
+
+    // handle facet fields
+    if (searchRequest.getFacetFields().isPresent()) {
+      for(String field : searchRequest.getFacetFields().get()) {
+        String name = getFacentAggregationName(field);
+        TermsBuilder terms = new TermsBuilder(name).field(field);
+        searchBuilder.aggregation(terms);
+      }
+    }
+
+    // return the search request
+    String[] indices = wildcardIndices(searchRequest.getIndices());
+    LOG.debug("Built Elasticsearch request; indices={}, request={}", indices, searchBuilder.toString());
+    return new org.elasticsearch.action.search.SearchRequest()
+            .indices(indices)
+            .source(searchBuilder);
+  }
+
+  /**
+   * Builds a search response.
+   *
+   * This effectively transforms an Elasticsearch search response into a Metron search response.
+   *
+   * @param searchRequest The Metron search request.
+   * @param esResponse The Elasticsearch search response.
+   * @return A Metron search response.
+   * @throws InvalidSearchException
+   */
+  private SearchResponse buildSearchResponse(
+          SearchRequest searchRequest,
+          org.elasticsearch.action.search.SearchResponse esResponse) throws InvalidSearchException {
+
     SearchResponse searchResponse = new SearchResponse();
-    searchResponse.setTotal(elasticsearchResponse.getHits().getTotalHits());
-    searchResponse.setResults(Arrays.stream(elasticsearchResponse.getHits().getHits()).map(searchHit ->
-        getSearchResult(searchHit, fields.isPresent())).collect(Collectors.toList()));
-    if (facetFields.isPresent()) {
+    searchResponse.setTotal(esResponse.getHits().getTotalHits());
+
+    // search hits --> search results
+    List<SearchResult> results = new ArrayList<>();
+    for(SearchHit hit: esResponse.getHits().getHits()) {
+      results.add(getSearchResult(hit, searchRequest.getFields().isPresent()));
+    }
+    searchResponse.setResults(results);
+
+    // handle facet fields
+    if (searchRequest.getFacetFields().isPresent()) {
+      List<String> facetFields = searchRequest.getFacetFields().get();
       Map<String, FieldType> commonColumnMetadata;
       try {
         commonColumnMetadata = getColumnMetadata(searchRequest.getIndices());
       } catch (IOException e) {
-        throw new InvalidSearchException(String.format("Could not get common column metadata for indices %s", Arrays.toString(searchRequest.getIndices().toArray())));
+        throw new InvalidSearchException(String.format(
+                "Could not get common column metadata for indices %s",
+                Arrays.toString(searchRequest.getIndices().toArray())));
       }
-      searchResponse.setFacetCounts(getFacetCounts(facetFields.get(), elasticsearchResponse.getAggregations(), commonColumnMetadata ));
+      searchResponse.setFacetCounts(getFacetCounts(facetFields, esResponse.getAggregations(), commonColumnMetadata ));
     }
+
+    LOG.debug("Built search response; response={}", ElasticsearchUtils.toJSON(searchResponse).orElse("???"));
     return searchResponse;
   }
 
@@ -188,42 +281,76 @@
    */
   protected GroupResponse group(GroupRequest groupRequest, QueryBuilder queryBuilder)
       throws InvalidSearchException {
+    org.elasticsearch.action.search.SearchRequest esRequest;
+    org.elasticsearch.action.search.SearchResponse esResponse;
+
     if (client == null) {
       throw new InvalidSearchException("Uninitialized Dao!  You must call init() prior to use.");
     }
     if (groupRequest.getGroups() == null || groupRequest.getGroups().size() == 0) {
       throw new InvalidSearchException("At least 1 group must be provided.");
     }
-    final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
-    searchSourceBuilder.query(queryBuilder);
-    searchSourceBuilder.aggregation(getGroupsTermBuilder(groupRequest, 0));
-    String[] wildcardIndices = wildcardIndices(groupRequest.getIndices());
-    org.elasticsearch.action.search.SearchRequest request;
-    org.elasticsearch.action.search.SearchResponse response;
 
-    try {
-      request = new org.elasticsearch.action.search.SearchRequest(wildcardIndices)
-          .source(searchSourceBuilder);
-      response = client.search(request).actionGet();
-    } catch (SearchPhaseExecutionException e) {
-      throw new InvalidSearchException("Could not execute search", e);
-    }
+    esRequest = buildGroupRequest(groupRequest, queryBuilder);
+    esResponse = requestSubmitter.submitSearch(esRequest);
+    GroupResponse response = buildGroupResponse(groupRequest, esResponse);
+
+    return response;
+  }
+
+  /**
+   * Builds a group search request.
+   * @param groupRequest The Metron group request.
+   * @param queryBuilder The search query.
+   * @return An Elasticsearch search request.
+   */
+  private org.elasticsearch.action.search.SearchRequest buildGroupRequest(
+          GroupRequest groupRequest,
+          QueryBuilder queryBuilder) {
+
+    // handle groups
+    TermsBuilder groups = getGroupsTermBuilder(groupRequest, 0);
+    final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
+            .query(queryBuilder)
+            .aggregation(groups);
+
+    // return the search request
+    String[] indices = wildcardIndices(groupRequest.getIndices());
+    return new org.elasticsearch.action.search.SearchRequest()
+            .indices(indices)
+            .source(searchSourceBuilder);
+  }
+
+  /**
+   * Build a group response.
+   * @param groupRequest The original group request.
+   * @param response The search response.
+   * @return A group response.
+   * @throws InvalidSearchException
+   */
+  private GroupResponse buildGroupResponse(
+          GroupRequest groupRequest,
+          org.elasticsearch.action.search.SearchResponse response) throws InvalidSearchException {
+
+    // build the search response
     Map<String, FieldType> commonColumnMetadata;
     try {
       commonColumnMetadata = getColumnMetadata(groupRequest.getIndices());
     } catch (IOException e) {
-      throw new InvalidSearchException(String
-          .format("Could not get common column metadata for indices %s",
+      throw new InvalidSearchException(String.format("Could not get common column metadata for indices %s",
               Arrays.toString(groupRequest.getIndices().toArray())));
     }
+
     GroupResponse groupResponse = new GroupResponse();
     groupResponse.setGroupedBy(groupRequest.getGroups().get(0).getField());
-    groupResponse.setGroupResults(
-        getGroupResults(groupRequest, 0, response.getAggregations(), commonColumnMetadata));
+    groupResponse.setGroupResults(getGroupResults(groupRequest, 0, response.getAggregations(), commonColumnMetadata));
     return groupResponse;
   }
 
   private String[] wildcardIndices(List<String> indices) {
+    if(indices == null)
+      return new String[] {};
+
     return indices
             .stream()
             .map(index -> String.format("%s%s*", index, INDEX_NAME_DELIMITER))
@@ -235,26 +362,43 @@
     if(this.client == null) {
       this.client = ElasticsearchUtils.getClient(config.getGlobalConfigSupplier().get(), config.getOptionalSettings());
       this.accessConfig = config;
+      this.columnMetadataDao = new ElasticsearchColumnMetadataDao(this.client.admin());
+      this.requestSubmitter = new ElasticsearchRequestSubmitter(this.client);
+    }
+
+    if(columnMetadataDao == null) {
+      throw new IllegalArgumentException("No ColumnMetadataDao available");
+    }
+
+    if(requestSubmitter == null) {
+      throw new IllegalArgumentException("No ElasticsearchRequestSubmitter available");
     }
   }
 
   @Override
   public Document getLatest(final String guid, final String sensorType) throws IOException {
-    Optional<Document> ret = searchByGuid(
-            guid
-            , sensorType
-            , hit -> {
-              Long ts = 0L;
-              String doc = hit.getSourceAsString();
-              String sourceType = Iterables.getFirst(Splitter.on("_doc").split(hit.getType()), null);
-              try {
-                return Optional.of(new Document(doc, guid, sourceType, ts));
-              } catch (IOException e) {
-                throw new IllegalStateException("Unable to retrieve latest: " + e.getMessage(), e);
-              }
-            }
-            );
-    return ret.orElse(null);
+    Optional<Document> doc = searchByGuid(guid, sensorType, hit -> toDocument(guid, hit));
+    return doc.orElse(null);
+  }
+
+  private Optional<Document> toDocument(final String guid, SearchHit hit) {
+    Long ts = 0L;
+    String doc = hit.getSourceAsString();
+    String sourceType = toSourceType(hit.getType());
+    try {
+      return Optional.of(new Document(doc, guid, sourceType, ts));
+    } catch (IOException e) {
+      throw new IllegalStateException("Unable to retrieve latest: " + e.getMessage(), e);
+    }
+  }
+
+  /**
+   * Returns the source type based on a given doc type.
+   * @param docType The document type.
+   * @return The source type.
+   */
+  private String toSourceType(String docType) {
+    return Iterables.getFirst(Splitter.on("_doc").split(docType), null);
   }
 
   @Override
@@ -394,8 +538,7 @@
     String type = sensorType + "_doc";
     Object ts = update.getTimestamp();
     IndexRequest indexRequest = new IndexRequest(indexName, type, update.getGuid())
-        .source(update.getDocument())
-        ;
+        .source(update.getDocument());
     if(ts != null) {
       indexRequest = indexRequest.timestamp(ts.toString());
     }
@@ -403,77 +546,9 @@
     return indexRequest;
   }
 
-  @SuppressWarnings("unchecked")
   @Override
   public Map<String, FieldType> getColumnMetadata(List<String> indices) throws IOException {
-    Map<String, FieldType> indexColumnMetadata = new HashMap<>();
-
-    // Keep track of the last index used to inspect a field type so we can print a helpful error message on type mismatch
-    Map<String, String> previousIndices = new HashMap<>();
-    // If we have detected a field type mismatch, ignore the field going forward since the type has been set to OTHER
-    Set<String> fieldBlackList = new HashSet<>();
-
-    String[] latestIndices = getLatestIndices(indices);
-    if (latestIndices.length > 0) {
-      ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings = client
-          .admin()
-          .indices()
-          .getMappings(new GetMappingsRequest().indices(latestIndices))
-          .actionGet()
-          .getMappings();
-      for (Object key : mappings.keys().toArray()) {
-        String indexName = key.toString();
-        ImmutableOpenMap<String, MappingMetaData> mapping = mappings.get(indexName);
-        Iterator<String> mappingIterator = mapping.keysIt();
-        while (mappingIterator.hasNext()) {
-          MappingMetaData mappingMetaData = mapping.get(mappingIterator.next());
-          Map<String, Map<String, String>> map = (Map<String, Map<String, String>>) mappingMetaData
-              .getSourceAsMap().get("properties");
-          for (String field : map.keySet()) {
-            if (!fieldBlackList.contains(field)) {
-              FieldType type = elasticsearchSearchTypeMap
-                  .getOrDefault(map.get(field).get("type"), FieldType.OTHER);
-              if (indexColumnMetadata.containsKey(field)) {
-                FieldType previousType = indexColumnMetadata.get(field);
-                if (!type.equals(previousType)) {
-                  String previousIndexName = previousIndices.get(field);
-                  LOG.error(String.format(
-                      "Field type mismatch: %s.%s has type %s while %s.%s has type %s.  Defaulting type to %s.",
-                      indexName, field, type.getFieldType(),
-                      previousIndexName, field, previousType.getFieldType(),
-                      FieldType.OTHER.getFieldType()));
-                  indexColumnMetadata.put(field, FieldType.OTHER);
-                  // Detected a type mismatch so ignore the field from now on
-                  fieldBlackList.add(field);
-                }
-              } else {
-                indexColumnMetadata.put(field, type);
-                previousIndices.put(field, indexName);
-              }
-            }
-          }
-        }
-      }
-    }
-    return indexColumnMetadata;
-  }
-
-  protected String[] getLatestIndices(List<String> includeIndices) {
-    Map<String, String> latestIndices = new HashMap<>();
-    String[] indices = client.admin().indices().prepareGetIndex().setFeatures().get().getIndices();
-    for (String index : indices) {
-      int prefixEnd = index.indexOf(INDEX_NAME_DELIMITER);
-      if (prefixEnd != -1) {
-        String prefix = index.substring(0, prefixEnd);
-        if (includeIndices.contains(prefix)) {
-          String latestIndex = latestIndices.get(prefix);
-          if (latestIndex == null || index.compareTo(latestIndex) > 0) {
-            latestIndices.put(prefix, index);
-          }
-        }
-      }
-    }
-    return latestIndices.values().toArray(new String[latestIndices.size()]);
+    return columnMetadataDao.getColumnMetadata(indices);
   }
 
   private org.elasticsearch.search.sort.SortOrder getElasticsearchSortOrder(
@@ -588,4 +663,19 @@
   private String getSumAggregationName(String field) {
     return String.format("%s_score", field);
   }
+
+  public ElasticsearchDao client(TransportClient client) {
+    this.client = client;
+    return this;
+  }
+
+  public ElasticsearchDao columnMetadataDao(ColumnMetadataDao columnMetadataDao) {
+    this.columnMetadataDao = columnMetadataDao;
+    return this;
+  }
+
+  public ElasticsearchDao accessConfig(AccessConfig accessConfig) {
+    this.accessConfig = accessConfig;
+    return this;
+  }
 }
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchRequestSubmitter.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchRequestSubmitter.java
new file mode 100644
index 0000000..0e0df21
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/dao/ElasticsearchRequestSubmitter.java
@@ -0,0 +1,138 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.metron.elasticsearch.dao;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
+import org.apache.metron.indexing.dao.search.InvalidSearchException;
+import org.elasticsearch.action.search.SearchPhaseExecutionException;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.rest.RestStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+
+/**
+ * Responsible for submitting requests to Elasticsearch.
+ */
+public class ElasticsearchRequestSubmitter {
+
+  private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+  /**
+   * The Elasticsearch client.
+   */
+  private TransportClient client;
+
+  public ElasticsearchRequestSubmitter(TransportClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Submit a search to Elasticsearch.
+   * @param request A search request.
+   * @return The search response.
+   */
+  public SearchResponse submitSearch(SearchRequest request) throws InvalidSearchException {
+    LOG.debug("About to submit a search; request={}", ElasticsearchUtils.toJSON(request).orElse("???"));
+
+    // submit the search request
+    org.elasticsearch.action.search.SearchResponse esResponse;
+    try {
+      esResponse = client
+              .search(request)
+              .actionGet();
+      LOG.debug("Got Elasticsearch response; response={}", esResponse.toString());
+
+    } catch (SearchPhaseExecutionException e) {
+      String msg = String.format(
+              "Failed to execute search; error='%s', search='%s'",
+              ExceptionUtils.getRootCauseMessage(e),
+              ElasticsearchUtils.toJSON(request).orElse("???"));
+      LOG.error(msg, e);
+      throw new InvalidSearchException(msg, e);
+    }
+
+    // check for shard failures
+    if(esResponse.getFailedShards() > 0) {
+      handleShardFailures(request, esResponse);
+    }
+
+    // validate the response status
+    if(RestStatus.OK == esResponse.status()) {
+      return esResponse;
+
+    } else {
+      // the search was not successful
+      String msg = String.format(
+              "Bad search response; status=%s, timeout=%s, terminatedEarly=%s",
+              esResponse.status(), esResponse.isTimedOut(), esResponse.isTerminatedEarly());
+      LOG.error(msg);
+      throw new InvalidSearchException(msg);
+    }
+  }
+
+  /**
+   * Handle individual shard failures that can occur even when the response is OK.  These
+   * can indicate misconfiguration of the search indices.
+   * @param request The search request.
+   * @param response  The search response.
+   */
+  private void handleShardFailures(
+          org.elasticsearch.action.search.SearchRequest request,
+          org.elasticsearch.action.search.SearchResponse response) {
+    /*
+     * shard failures are only logged.  the search itself is not failed.  this approach
+     * assumes that a user is interested in partial search results, even if the
+     * entire search result set cannot be produced.
+     *
+     * for example, assume the user adds an additional sensor and the telemetry
+     * is indexed into a new search index.  if that search index is misconfigured,
+     * it can result in partial shard failures.  rather than failing the entire search,
+     * we log the error and allow the results to be returned from shards that
+     * are correctly configured.
+     */
+    int errors = ArrayUtils.getLength(response.getShardFailures());
+    LOG.error("Search resulted in {}/{} shards failing; errors={}, search={}",
+            response.getFailedShards(),
+            response.getTotalShards(),
+            errors,
+            ElasticsearchUtils.toJSON(request).orElse("???"));
+
+    // log each reported failure
+    int failureCount=1;
+    for(ShardSearchFailure fail: response.getShardFailures()) {
+      String msg = String.format(
+              "Shard search failure [%s/%s]; reason=%s, index=%s, shard=%s, status=%s, nodeId=%s",
+              failureCount,
+              errors,
+              ExceptionUtils.getRootCauseMessage(fail.getCause()),
+              fail.index(),
+              fail.shardId(),
+              fail.status(),
+              fail.shard().getNodeId());
+      LOG.error(msg, fail.getCause());
+    }
+  }
+}
diff --git a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/utils/ElasticsearchUtils.java b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/utils/ElasticsearchUtils.java
index 4c9933b..f29012a 100644
--- a/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/utils/ElasticsearchUtils.java
+++ b/metron-platform/metron-elasticsearch/src/main/java/org/apache/metron/elasticsearch/utils/ElasticsearchUtils.java
@@ -22,10 +22,15 @@
 import com.google.common.collect.Iterables;
 import org.apache.commons.lang.StringUtils;
 import org.apache.metron.common.configuration.writer.WriterConfiguration;
+import org.codehaus.jackson.map.ObjectMapper;
 import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.lang.invoke.MethodHandles;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.text.SimpleDateFormat;
@@ -35,11 +40,14 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static java.lang.String.format;
 
 public class ElasticsearchUtils {
 
+  private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
   private static ThreadLocal<Map<String, SimpleDateFormat>> DATE_FORMAT_CACHE
           = ThreadLocal.withInitial(() -> new HashMap<>());
 
@@ -179,4 +187,48 @@
     }
     throw new IllegalStateException("Unable to read the elasticsearch ips, expected es.ip to be either a list of strings, a string hostname or a host:port string");
   }
+
+  /**
+   * Converts an Elasticsearch SearchRequest to JSON.
+   * @param esRequest The search request.
+   * @return The JSON representation of the SearchRequest.
+   */
+  public static Optional<String> toJSON(org.elasticsearch.action.search.SearchRequest esRequest) {
+    Optional<String> json = Optional.empty();
+
+    if(esRequest != null) {
+      try {
+        json = Optional.of(XContentHelper.convertToJson(esRequest.source(), true));
+
+      } catch (Throwable t) {
+        LOG.error("Failed to convert search request to JSON", t);
+      }
+    }
+
+    return json;
+  }
+
+  /**
+   * Convert a SearchRequest to JSON.
+   * @param request The search request.
+   * @return The JSON representation of the SearchRequest.
+   */
+  public static Optional<String> toJSON(Object request) {
+    Optional<String> json = Optional.empty();
+
+    if(request != null) {
+      try {
+        json = Optional.of(
+                new ObjectMapper()
+                        .writer()
+                        .withDefaultPrettyPrinter()
+                        .writeValueAsString(request));
+
+      } catch (Throwable t) {
+        LOG.error("Failed to convert request to JSON", t);
+      }
+    }
+
+    return json;
+  }
 }
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchColumnMetadataDaoTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchColumnMetadataDaoTest.java
new file mode 100644
index 0000000..0a83ee0
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchColumnMetadataDaoTest.java
@@ -0,0 +1,144 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.metron.elasticsearch.dao;
+
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.admin.indices.get.GetIndexRequestBuilder;
+import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
+import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
+import org.elasticsearch.client.AdminClient;
+import org.elasticsearch.client.IndicesAdminClient;
+import org.elasticsearch.cluster.metadata.MappingMetaData;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests the ElasticsearchColumnMetadata class.
+ */
+public class ElasticsearchColumnMetadataDaoTest {
+
+  /**
+   * @param indices The names of all indices that will exist.
+   * @return An object to test.
+   */
+  public ElasticsearchColumnMetadataDao setup(String[] indices) {
+    return setup(indices, ImmutableOpenMap.of());
+  }
+
+  /**
+   * @param indices The names of all indices that will exist.
+   * @param mappings The index mappings.
+   * @return An object to test.
+   */
+  public ElasticsearchColumnMetadataDao setup(
+          String[] indices,
+          ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings) {
+
+    AdminClient adminClient = mock(AdminClient.class);
+    IndicesAdminClient indicesAdminClient = mock(IndicesAdminClient.class);
+    GetIndexRequestBuilder getIndexRequestBuilder = mock(GetIndexRequestBuilder.class);
+    GetIndexResponse getIndexResponse = mock(GetIndexResponse.class);
+    ActionFuture getMappingsActionFuture = mock(ActionFuture.class);
+    GetMappingsResponse getMappingsResponse = mock(GetMappingsResponse.class);
+
+    // setup the mocks so that a set of indices are available to the DAO
+    when(adminClient.indices()).thenReturn(indicesAdminClient);
+    when(indicesAdminClient.prepareGetIndex()).thenReturn(getIndexRequestBuilder);
+    when(getIndexRequestBuilder.setFeatures()).thenReturn(getIndexRequestBuilder);
+    when(getIndexRequestBuilder.get()).thenReturn(getIndexResponse);
+    when(getIndexResponse.getIndices()).thenReturn(indices);
+
+    // setup the mocks so that a set of mappings are available to the DAO
+    when(indicesAdminClient.getMappings(any())).thenReturn(getMappingsActionFuture);
+    when(getMappingsActionFuture.actionGet()).thenReturn(getMappingsResponse);
+    when(getMappingsResponse.getMappings()).thenReturn(mappings);
+
+    return new ElasticsearchColumnMetadataDao(adminClient);
+  }
+
+  @Test
+  public void testGetOneLatestIndex() {
+
+    // setup
+    String[] existingIndices = new String[] {
+            "bro_index_2017.10.03.19",
+            "bro_index_2017.10.03.20",
+            "bro_index_2017.10.03.21",
+            "snort_index_2017.10.03.19",
+            "snort_index_2017.10.03.20",
+            "snort_index_2017.10.03.21"
+    };
+    ElasticsearchColumnMetadataDao dao = setup(existingIndices);
+
+    // get the latest indices
+    List<String> args = Collections.singletonList("bro");
+    String[] actual = dao.getLatestIndices(args);
+
+    // validation
+    String [] expected = new String[] { "bro_index_2017.10.03.21" };
+    assertArrayEquals(expected, actual);
+  }
+
+  @Test
+  public void testGetLatestIndices() {
+    // setup
+    String[] existingIndices = new String[] {
+            "bro_index_2017.10.03.19",
+            "bro_index_2017.10.03.20",
+            "bro_index_2017.10.03.21",
+            "snort_index_2017.10.03.19",
+            "snort_index_2017.10.03.19",
+            "snort_index_2017.10.03.21"
+    };
+    ElasticsearchColumnMetadataDao dao = setup(existingIndices);
+
+    // get the latest indices
+    List<String> args = Arrays.asList("bro", "snort");
+    String[] actual = dao.getLatestIndices(args);
+
+    // validation
+    String [] expected = new String[] { "bro_index_2017.10.03.21", "snort_index_2017.10.03.21" };
+    assertArrayEquals(expected, actual);
+  }
+
+  @Test
+  public void testLatestIndicesWhereNoneExist() {
+
+    // setup - there are no existing indices
+    String[] existingIndices = new String[] {};
+    ElasticsearchColumnMetadataDao dao = setup(existingIndices);
+
+    // get the latest indices
+    List<String> args = Arrays.asList("bro", "snort");
+    String[] actual = dao.getLatestIndices(args);
+
+    // validation
+    String [] expected = new String[] {};
+    assertArrayEquals(expected, actual);
+  }
+}
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchDaoTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchDaoTest.java
index 7c33018..a6c0aa6 100644
--- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchDaoTest.java
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchDaoTest.java
@@ -17,98 +17,209 @@
  */
 package org.apache.metron.elasticsearch.dao;
 
-import org.apache.metron.elasticsearch.matcher.SearchRequestMatcher;
+import org.apache.metron.elasticsearch.utils.ElasticsearchUtils;
 import org.apache.metron.indexing.dao.AccessConfig;
-import org.apache.metron.indexing.dao.IndexDao;
-import org.apache.metron.indexing.dao.search.*;
-import org.elasticsearch.action.ActionFuture;
+import org.apache.metron.indexing.dao.search.FieldType;
+import org.apache.metron.indexing.dao.search.InvalidSearchException;
+import org.apache.metron.indexing.dao.search.SearchRequest;
+import org.apache.metron.indexing.dao.search.SearchResponse;
+import org.apache.metron.indexing.dao.search.SortField;
+import org.apache.metron.indexing.dao.search.SortOrder;
 import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
-import org.junit.Assert;
-import org.junit.Before;
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
 import org.junit.Test;
-import org.mockito.Mock;
+import org.mockito.ArgumentCaptor;
 
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.argThat;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class ElasticsearchDaoTest {
 
-  private IndexDao searchService;
+  private ElasticsearchDao dao;
+  private ElasticsearchRequestSubmitter requestSubmitter;
 
-  @Mock
-  TransportClient client;
+  private void setup(RestStatus status, int maxSearchResults, Map<String, FieldType> metadata) throws Exception {
 
-  @Before
-  public void setUp() throws Exception {
-    client = mock(TransportClient.class);
+    // setup the mock search hits
+    SearchHit hit1 = mock(SearchHit.class);
+    when(hit1.getId()).thenReturn("id1");
+    when(hit1.getSource()).thenReturn(new HashMap<String, Object>(){{ put("field", "value1"); }});
+    when(hit1.getScore()).thenReturn(0.1f);
+
+    SearchHit hit2 = mock(SearchHit.class);
+    when(hit2.getId()).thenReturn("id2");
+    when(hit2.getSource()).thenReturn(new HashMap<String, Object>(){{ put("field", "value2"); }});
+    when(hit2.getScore()).thenReturn(0.2f);
+
+    // search hits
+    SearchHit[] hits = { hit1, hit2 };
+    SearchHits searchHits = mock(SearchHits.class);
+    when(searchHits.getHits()).thenReturn(hits);
+    when(searchHits.getTotalHits()).thenReturn(Integer.toUnsignedLong(hits.length));
+
+    // search response which returns the search hits
+    org.elasticsearch.action.search.SearchResponse response = mock(org.elasticsearch.action.search.SearchResponse.class);
+    when(response.status()).thenReturn(status);
+    when(response.getHits()).thenReturn(searchHits);
+
+    // provides column metadata
+    ColumnMetadataDao columnMetadataDao = mock(ColumnMetadataDao.class);
+    when(columnMetadataDao.getColumnMetadata(any())).thenReturn(metadata);
+
+    // returns the search response
+    requestSubmitter = mock(ElasticsearchRequestSubmitter.class);
+    when(requestSubmitter.submitSearch(any())).thenReturn(response);
+
+    TransportClient client = mock(TransportClient.class);
+
+    // provides configuration
     AccessConfig config = mock(AccessConfig.class);
-    when(config.getMaxSearchResults()).thenReturn(50);
-    searchService = new ElasticsearchDao(client, config);
+    when(config.getMaxSearchResults()).thenReturn(maxSearchResults);
+
+    dao = new ElasticsearchDao(client, columnMetadataDao, requestSubmitter, config);
+  }
+
+  private void setup(RestStatus status, int maxSearchResults) throws Exception {
+    setup(status, maxSearchResults, new HashMap<>());
   }
 
   @Test
-  public void searchShouldProperlyBuildSearchRequest() throws Exception {
+  public void searchShouldSortByGivenFields() throws Exception {
 
-    // setup the mock client
-    SearchHit searchHit1 = mock(SearchHit.class);
-    when(searchHit1.getId()).thenReturn("id1");
-    when(searchHit1.getSource()).thenReturn(new HashMap<String, Object>(){{ put("field", "value1"); }});
-    when(searchHit1.getScore()).thenReturn(0.1f);
+    // setup the column metadata
+    Map<String, FieldType> columnMetadata = new HashMap<>();
+    columnMetadata.put("sortByStringDesc", FieldType.STRING);
+    columnMetadata.put("sortByIntAsc", FieldType.INTEGER);
 
-    SearchHit searchHit2 = mock(SearchHit.class);
-    when(searchHit2.getId()).thenReturn("id2");
-    when(searchHit2.getSource()).thenReturn(new HashMap<String, Object>(){{ put("field", "value2"); }});
-    when(searchHit2.getScore()).thenReturn(0.2f);
-
-    SearchHits searchHits = mock(SearchHits.class);
-    when(searchHits.getHits()).thenReturn(new SearchHit[]{searchHit1, searchHit2});
-    when(searchHits.getTotalHits()).thenReturn(2L);
-
-    org.elasticsearch.action.search.SearchResponse elasticsearchResponse = mock(org.elasticsearch.action.search.SearchResponse.class);
-    when(elasticsearchResponse.getHits()).thenReturn(searchHits);
-
-    ActionFuture actionFuture = mock(ActionFuture.class);
-    when(actionFuture.actionGet()).thenReturn(elasticsearchResponse);
-    when(client.search(any())).thenReturn(actionFuture);
+    // setup the dao
+    setup(RestStatus.OK, 25, columnMetadata);
 
     // "sort by" fields for the search request
-    SortField[] sortFields = {
-            sortBy("sortField1", SortOrder.DESC),
-            sortBy("sortField2", SortOrder.ASC)
+    SortField[] expectedSortFields = {
+            sortBy("sortByStringDesc", SortOrder.DESC),
+            sortBy("sortByIntAsc", SortOrder.ASC),
+            sortBy("sortByUndefinedDesc", SortOrder.DESC)
     };
 
-    // create a search request
+    // create a metron search request
+    final List<String> indices = Arrays.asList("bro", "snort");
     SearchRequest searchRequest = new SearchRequest();
     searchRequest.setSize(2);
-    searchRequest.setIndices(Arrays.asList("bro", "snort"));
+    searchRequest.setIndices(indices);
     searchRequest.setFrom(5);
-    searchRequest.setSort(Arrays.asList(sortFields));
+    searchRequest.setSort(Arrays.asList(expectedSortFields));
     searchRequest.setQuery("some query");
 
-    // submit the search request
-    SearchResponse searchResponse = searchService.search(searchRequest);
+    // submit the metron search request
+    SearchResponse searchResponse = dao.search(searchRequest);
+    assertNotNull(searchResponse);
 
-    // validate
-    String[] expectedIndices = {"bro_index*", "snort_index*"};
-    verify(client).search(argThat(new SearchRequestMatcher(expectedIndices, "some query", 2, 5, sortFields)));
-    assertEquals(2, searchResponse.getTotal());
-    List<SearchResult> actualSearchResults = searchResponse.getResults();
-    assertEquals(2, actualSearchResults.size());
-    assertEquals("id1", actualSearchResults.get(0).getId());
-    assertEquals("value1", actualSearchResults.get(0).getSource().get("field"));
-    assertEquals(0.1f, actualSearchResults.get(0).getScore(), 0.0f);
-    assertEquals("id2", actualSearchResults.get(1).getId());
-    assertEquals("value2", actualSearchResults.get(1).getSource().get("field"));
-    assertEquals(0.2f, actualSearchResults.get(1).getScore(), 0.0f);
-    verifyNoMoreInteractions(client);
+    // capture the elasticsearch search request that was created
+    ArgumentCaptor<org.elasticsearch.action.search.SearchRequest> argument = ArgumentCaptor.forClass(org.elasticsearch.action.search.SearchRequest.class);
+    verify(requestSubmitter).submitSearch(argument.capture());
+    org.elasticsearch.action.search.SearchRequest request = argument.getValue();
+
+    // transform the request to JSON for validation
+    JSONParser parser = new JSONParser();
+    JSONObject json = (JSONObject) parser.parse(ElasticsearchUtils.toJSON(request).orElse("???"));
+
+    // validate the sort fields
+    JSONArray sortFields = (JSONArray) json.get("sort");
+    assertEquals(3, sortFields.size());
+
+    {
+      // sort by string descending
+      JSONObject aSortField = (JSONObject) sortFields.get(0);
+      JSONObject sortBy = (JSONObject) aSortField.get("sortByStringDesc");
+      assertEquals("desc", sortBy.get("order"));
+      assertEquals("_last", sortBy.get("missing"));
+      assertEquals("string", sortBy.get("unmapped_type"));
+    }
+    {
+      // sort by integer ascending
+      JSONObject aSortField = (JSONObject) sortFields.get(1);
+      JSONObject sortByIntAsc = (JSONObject) aSortField.get("sortByIntAsc");
+      assertEquals("asc", sortByIntAsc.get("order"));
+      assertEquals("_first", sortByIntAsc.get("missing"));
+      assertEquals("integer", sortByIntAsc.get("unmapped_type"));
+    }
+    {
+      // sort by unknown type
+      JSONObject aSortField = (JSONObject) sortFields.get(2);
+      JSONObject sortByUndefinedDesc = (JSONObject) aSortField.get("sortByUndefinedDesc");
+      assertEquals("desc", sortByUndefinedDesc.get("order"));
+      assertEquals("_last", sortByUndefinedDesc.get("missing"));
+      assertEquals("other", sortByUndefinedDesc.get("unmapped_type"));
+    }
+  }
+
+  @Test
+  public void searchShouldWildcardIndices() throws Exception {
+
+    // setup the dao
+    setup(RestStatus.OK, 25);
+
+    // "sort by" fields for the search request
+    SortField[] expectedSortFields = {
+            sortBy("sortByStringDesc", SortOrder.DESC),
+            sortBy("sortByIntAsc", SortOrder.ASC),
+            sortBy("sortByUndefinedDesc", SortOrder.DESC)
+    };
+
+    // create a metron search request
+    final List<String> indices = Arrays.asList("bro", "snort");
+    SearchRequest searchRequest = new SearchRequest();
+    searchRequest.setSize(2);
+    searchRequest.setIndices(indices);
+    searchRequest.setFrom(5);
+    searchRequest.setSort(Arrays.asList(expectedSortFields));
+    searchRequest.setQuery("some query");
+
+    // submit the metron search request
+    SearchResponse searchResponse = dao.search(searchRequest);
+    assertNotNull(searchResponse);
+
+    // capture the elasticsearch search request that was created
+    ArgumentCaptor<org.elasticsearch.action.search.SearchRequest> argument = ArgumentCaptor.forClass(org.elasticsearch.action.search.SearchRequest.class);
+    verify(requestSubmitter).submitSearch(argument.capture());
+    org.elasticsearch.action.search.SearchRequest request = argument.getValue();
+
+    // transform the request to JSON for validation
+    JSONParser parser = new JSONParser();
+    JSONObject json = (JSONObject) parser.parse(ElasticsearchUtils.toJSON(request).orElse("???"));
+
+    // ensure that the index names are 'wildcard-ed'
+    String[] expected = { "bro_index*", "snort_index*" };
+    assertArrayEquals(expected, request.indices());
+  }
+
+
+  @Test(expected = InvalidSearchException.class)
+  public void searchShouldThrowExceptionWhenMaxResultsAreExceeded() throws Exception {
+
+    int maxSearchResults = 20;
+    setup(RestStatus.OK, maxSearchResults);
+
+    SearchRequest searchRequest = new SearchRequest();
+    searchRequest.setSize(maxSearchResults+1);
+
+    dao.search(searchRequest);
+    // exception expected - size > max
   }
 
   private SortField sortBy(String field, SortOrder order) {
@@ -118,19 +229,4 @@
     return sortField;
   }
 
-  @Test
-  public void searchShouldThrowExceptionWhenMaxResultsAreExceeded() throws Exception {
-    SearchRequest searchRequest = new SearchRequest();
-    searchRequest.setSize(51);
-    try {
-      searchService.search(searchRequest);
-      Assert.fail("Did not throw expected exception");
-    }
-    catch(InvalidSearchException ise) {
-      Assert.assertEquals("Search result size must be less than 50", ise.getMessage());
-    }
-  }
-
-
-
 }
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchRequestSubmitterTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchRequestSubmitterTest.java
new file mode 100644
index 0000000..26f5fff
--- /dev/null
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/dao/ElasticsearchRequestSubmitterTest.java
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.metron.elasticsearch.dao;
+
+import org.apache.metron.indexing.dao.search.InvalidSearchException;
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchShardTarget;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ElasticsearchRequestSubmitterTest {
+
+  private ElasticsearchRequestSubmitter submitter;
+
+  public ElasticsearchRequestSubmitter setup(SearchResponse response) {
+
+    // mocks
+    TransportClient client = mock(TransportClient.class);
+    ActionFuture future = Mockito.mock(ActionFuture.class);
+
+    // the client should return the given search response
+    when(client.search(any())).thenReturn(future);
+    when(future.actionGet()).thenReturn(response);
+
+    return new ElasticsearchRequestSubmitter(client);
+  }
+
+  @Test
+  public void searchShouldSucceedWhenOK() throws InvalidSearchException {
+
+    // mocks
+    SearchResponse response = mock(SearchResponse.class);
+    SearchRequest request = mock(SearchRequest.class);
+
+    // response will have status of OK and no failed shards
+    when(response.status()).thenReturn(RestStatus.OK);
+    when(response.getFailedShards()).thenReturn(0);
+    when(response.getTotalShards()).thenReturn(2);
+
+    // search should succeed
+    ElasticsearchRequestSubmitter submitter = setup(response);
+    SearchResponse actual = submitter.submitSearch(request);
+    assertNotNull(actual);
+  }
+
+  @Test(expected = InvalidSearchException.class)
+  public void searchShouldFailWhenNotOK() throws InvalidSearchException {
+
+    // mocks
+    SearchResponse response = mock(SearchResponse.class);
+    SearchRequest request = mock(SearchRequest.class);
+
+    // response will have status of OK
+    when(response.status()).thenReturn(RestStatus.PARTIAL_CONTENT);
+    when(response.getFailedShards()).thenReturn(0);
+    when(response.getTotalShards()).thenReturn(2);
+
+    // search should succeed
+    ElasticsearchRequestSubmitter submitter = setup(response);
+    submitter.submitSearch(request);
+  }
+
+  @Test
+  public void searchShouldHandleShardFailure() throws InvalidSearchException {
+    // mocks
+    SearchResponse response = mock(SearchResponse.class);
+    SearchRequest request = mock(SearchRequest.class);
+    ShardSearchFailure fail = mock(ShardSearchFailure.class);
+    SearchShardTarget target = mock(SearchShardTarget.class);
+
+    // response will have status of OK
+    when(response.status()).thenReturn(RestStatus.OK);
+
+    // the response will report shard failures
+    when(response.getFailedShards()).thenReturn(1);
+    when(response.getTotalShards()).thenReturn(2);
+
+    // the response will return the failures
+    ShardSearchFailure[] failures = { fail };
+    when(response.getShardFailures()).thenReturn(failures);
+
+    // shard failure needs to report the node
+    when(fail.shard()).thenReturn(target);
+    when(target.getNodeId()).thenReturn("node1");
+
+    // shard failure needs to report details of failure
+    when(fail.index()).thenReturn("bro_index_2017-10-11");
+    when(fail.shardId()).thenReturn(1);
+
+    // search should succeed, even with failed shards
+    ElasticsearchRequestSubmitter submitter = setup(response);
+    SearchResponse actual = submitter.submitSearch(request);
+    assertNotNull(actual);
+  }
+}
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java
index 07cc708..3d50e99 100644
--- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java
+++ b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/integration/ElasticsearchSearchIntegrationTest.java
@@ -17,18 +17,11 @@
  */
 package org.apache.metron.elasticsearch.integration;
 
-
-import java.io.File;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.concurrent.ExecutionException;
 import org.adrianwalker.multilinestring.Multiline;
 import org.apache.metron.elasticsearch.dao.ElasticsearchDao;
-import org.apache.metron.elasticsearch.dao.ElasticsearchMetaAlertDao;
 import org.apache.metron.elasticsearch.integration.components.ElasticSearchComponent;
 import org.apache.metron.indexing.dao.AccessConfig;
 import org.apache.metron.indexing.dao.IndexDao;
-import org.apache.metron.indexing.dao.MetaAlertDao;
 import org.apache.metron.indexing.dao.SearchIntegrationTest;
 import org.apache.metron.integration.InMemoryComponent;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
@@ -43,7 +36,13 @@
 import org.json.simple.parser.JSONParser;
 import org.json.simple.parser.ParseException;
 
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.concurrent.ExecutionException;
+
 public class ElasticsearchSearchIntegrationTest extends SearchIntegrationTest {
+
   private static String indexDir = "target/elasticsearch_search";
   private static String dateFormat = "yyyy.MM.dd.HH";
   private static final int MAX_RETRIES = 10;
@@ -53,19 +52,46 @@
    * {
    * "bro_doc": {
    *   "properties": {
-   *     "source:type": { "type": "string" },
-   *     "ip_src_addr": { "type": "ip" },
-   *     "ip_src_port": { "type": "integer" },
-   *     "long_field": { "type": "long" },
-   *     "timestamp" : { "type": "date" },
-   *     "latitude" : { "type": "float" },
-   *     "score": { "type": "double" },
-   *     "is_alert": { "type": "boolean" },
-   *     "location_point": { "type": "geo_point" },
-   *     "bro_field": { "type": "string" },
-   *     "duplicate_name_field": { "type": "string" }
+   *     "source:type": {
+   *        "type": "string",
+   *        "index": "not_analyzed"
+   *     },
+   *     "ip_src_addr": {
+   *        "type": "ip"
+   *     },
+   *     "ip_src_port": {
+   *        "type": "integer"
+   *     },
+   *     "long_field": {
+   *        "type": "long"
+   *     },
+   *     "timestamp": {
+   *        "type": "date",
+   *        "format": "epoch_millis"
+   *      },
+   *     "latitude" : {
+   *        "type": "float"
+   *      },
+   *     "score": {
+   *        "type": "double"
+   *     },
+   *     "is_alert": {
+   *        "type": "boolean"
+   *     },
+   *     "location_point": {
+   *        "type": "geo_point"
+   *     },
+   *     "bro_field": {
+   *        "type": "string"
+   *     },
+   *     "duplicate_name_field": {
+   *        "type": "string"
+   *     },
+   *     "alert": {
+   *         "type": "nested"
+   *     }
    *   }
-   * }
+   *  }
    * }
    */
   @Multiline
@@ -73,21 +99,51 @@
 
   /**
    * {
-   * "snort_doc": {
-   *   "properties": {
-   *     "source:type": { "type": "string" },
-   *     "ip_src_addr": { "type": "ip" },
-   *     "ip_src_port": { "type": "integer" },
-   *     "long_field": { "type": "long" },
-   *     "timestamp" : { "type": "date" },
-   *     "latitude" : { "type": "float" },
-   *     "score": { "type": "double" },
-   *     "is_alert": { "type": "boolean" },
-   *     "location_point": { "type": "geo_point" },
-   *     "snort_field": { "type": "integer" },
-   *     "duplicate_name_field": { "type": "integer" }
-   *   }
-   * }
+   *  "snort_doc": {
+   *     "properties": {
+   *        "source:type": {
+   *          "type": "string",
+   *          "index": "not_analyzed"
+   *        },
+   *        "ip_src_addr": {
+   *          "type": "ip"
+   *        },
+   *        "ip_src_port": {
+   *          "type": "integer"
+   *        },
+   *        "long_field": {
+   *          "type": "long"
+   *        },
+   *        "timestamp": {
+   *          "type": "date",
+   *          "format": "epoch_millis"
+   *        },
+   *        "latitude" : {
+   *          "type": "float"
+   *        },
+   *        "score": {
+   *          "type": "double"
+   *        },
+   *        "is_alert": {
+   *          "type": "boolean"
+   *        },
+   *        "location_point": {
+   *          "type": "geo_point"
+   *        },
+   *        "snort_field": {
+   *          "type": "integer"
+   *        },
+   *        "duplicate_name_field": {
+   *          "type": "integer"
+   *        },
+   *        "alert": {
+   *           "type": "nested"
+   *        },
+   *        "threat:triage:score": {
+   *           "type": "float"
+   *        }
+   *      }
+   *    }
    * }
    */
   @Multiline
@@ -106,27 +162,23 @@
   @Multiline
   private static String metaAlertTypeMappings;
 
-
   @Override
   protected IndexDao createDao() throws Exception {
-    IndexDao elasticsearchDao = new ElasticsearchDao();
-    elasticsearchDao.init(
-            new AccessConfig() {{
-              setMaxSearchResults(100);
-              setMaxSearchGroups(100);
-              setGlobalConfigSupplier( () ->
-                new HashMap<String, Object>() {{
-                  put("es.clustername", "metron");
-                  put("es.port", "9300");
-                  put("es.ip", "localhost");
-                  put("es.date.format", dateFormat);
-                  }}
-              );
+    AccessConfig config = new AccessConfig();
+    config.setMaxSearchResults(100);
+    config.setMaxSearchGroups(100);
+    config.setGlobalConfigSupplier( () ->
+            new HashMap<String, Object>() {{
+              put("es.clustername", "metron");
+              put("es.port", "9300");
+              put("es.ip", "localhost");
+              put("es.date.format", dateFormat);
             }}
     );
-    MetaAlertDao ret = new ElasticsearchMetaAlertDao();
-    ret.init(elasticsearchDao);
-    return elasticsearchDao;
+
+    IndexDao dao = new ElasticsearchDao();
+    dao.init(config);
+    return dao;
   }
 
   @Override
diff --git a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/matcher/SearchRequestMatcher.java b/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/matcher/SearchRequestMatcher.java
deleted file mode 100644
index 417e48b..0000000
--- a/metron-platform/metron-elasticsearch/src/test/java/org/apache/metron/elasticsearch/matcher/SearchRequestMatcher.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF 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.
- */
-package org.apache.metron.elasticsearch.matcher;
-
-import org.apache.metron.indexing.dao.search.SortField;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.index.query.QueryStringQueryBuilder;
-import org.elasticsearch.search.builder.SearchSourceBuilder;
-import org.elasticsearch.search.sort.FieldSortBuilder;
-import org.elasticsearch.search.sort.SortOrder;
-import org.hamcrest.Description;
-import org.mockito.ArgumentMatcher;
-
-import java.util.Arrays;
-
-public class SearchRequestMatcher extends ArgumentMatcher<SearchRequest> {
-
-  private String[] expectedIndices;
-  private String[] actualIndices;
-
-  private BytesReference expectedSource;
-  private BytesReference actualSource;
-
-  private boolean indicesMatch;
-  private boolean sourcesMatch;
-
-  public SearchRequestMatcher(String[] indices, String query, int size, int from, SortField[] sortFields) {
-    expectedIndices = indices;
-    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
-            .size(size)
-            .from(from)
-            .query(new QueryStringQueryBuilder(query))
-            .fetchSource(true)
-            .trackScores(true);
-    for(SortField sortField: sortFields) {
-      FieldSortBuilder fieldSortBuilder = new FieldSortBuilder(sortField.getField());
-      fieldSortBuilder.order(sortField.getSortOrder() == org.apache.metron.indexing.dao.search.SortOrder.DESC ? SortOrder.DESC : SortOrder.ASC);
-      searchSourceBuilder = searchSourceBuilder.sort(fieldSortBuilder);
-    }
-    expectedSource = searchSourceBuilder.buildAsBytes(Requests.CONTENT_TYPE);
-  }
-
-  @Override
-  public boolean matches(Object o) {
-    SearchRequest searchRequest = (SearchRequest) o;
-
-    actualIndices = searchRequest.indices();
-    actualSource = searchRequest.source();
-
-    indicesMatch = Arrays.equals(expectedIndices, actualIndices);
-    sourcesMatch = expectedSource.equals(actualSource);
-
-    return indicesMatch && sourcesMatch;
-  }
-
-  @Override
-  public void describeTo(Description description) {
-    if(!indicesMatch) {
-      description.appendText("Bad search request indices: ");
-      description.appendText(" expected=");
-      description.appendValue(expectedIndices);
-      description.appendText(", got=");
-      description.appendValue(actualIndices);
-      description.appendText("  ");
-    }
-
-    if(!sourcesMatch) {
-      description.appendText("Bad search request sources: ");
-      description.appendText(" expected=");
-      description.appendValue(expectedSource);
-      description.appendText(", got=");
-      description.appendValue(actualSource);
-      description.appendText("  ");
-    }
-  }
-}
diff --git a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
index 4f47a65..c16401e 100644
--- a/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
+++ b/metron-platform/metron-indexing/src/main/java/org/apache/metron/indexing/dao/AccessConfig.java
@@ -31,8 +31,7 @@
   private TableProvider tableProvider = null;
 
   /**
-   * A supplier which will return the current global config.
-   * @return
+   * @return A supplier which will return the current global config.
    */
   public Supplier<Map<String, Object>> getGlobalConfigSupplier() {
     return globalConfigSupplier;
@@ -43,8 +42,7 @@
   }
 
   /**
-   * The maximum search result.
-   * @return
+   * @return The maximum number of search results.
    */
   public Integer getMaxSearchResults() {
     return maxSearchResults;
@@ -55,8 +53,7 @@
   }
 
   /**
-   * The maximum search groups.
-   * @return
+   * @return The maximum number of search groups.
    */
   public Integer getMaxSearchGroups() {
     return maxSearchGroups;
@@ -67,8 +64,7 @@
   }
 
   /**
-   * Get optional settings for initializing indices.
-   * @return
+   * @return Optional settings for initializing indices.
    */
   public Map<String, String> getOptionalSettings() {
     return optionalSettings;
@@ -79,8 +75,7 @@
   }
 
   /**
-   * Return the table provider to use for NoSql DAOs
-   * @return
+   * @return The table provider to use for NoSql DAOs
    */
   public TableProvider getTableProvider() {
     return tableProvider;
@@ -89,5 +84,4 @@
   public void setTableProvider(TableProvider tableProvider) {
     this.tableProvider = tableProvider;
   }
-
 }
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
index f2108de..002ec28 100644
--- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
+++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/InMemoryDao.java
@@ -21,14 +21,34 @@
 import com.google.common.base.Splitter;
 import com.google.common.collect.ComparisonChain;
 import com.google.common.collect.Iterables;
-import java.util.Map.Entry;
+import com.google.common.collect.Ordering;
 import org.apache.metron.common.Constants;
 import org.apache.metron.common.utils.JSONUtils;
-import org.apache.metron.indexing.dao.search.*;
+import org.apache.metron.indexing.dao.search.FieldType;
+import org.apache.metron.indexing.dao.search.GetRequest;
+import org.apache.metron.indexing.dao.search.Group;
+import org.apache.metron.indexing.dao.search.GroupRequest;
+import org.apache.metron.indexing.dao.search.GroupResponse;
+import org.apache.metron.indexing.dao.search.GroupResult;
+import org.apache.metron.indexing.dao.search.InvalidSearchException;
+import org.apache.metron.indexing.dao.search.SearchRequest;
+import org.apache.metron.indexing.dao.search.SearchResponse;
+import org.apache.metron.indexing.dao.search.SearchResult;
+import org.apache.metron.indexing.dao.search.SortField;
+import org.apache.metron.indexing.dao.search.SortOrder;
 import org.apache.metron.indexing.dao.update.Document;
 
 import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.UUID;
 
 public class InMemoryDao implements IndexDao {
   // Map from index to list of documents as JSON strings
@@ -102,13 +122,14 @@
 
   private static class ComparableComparator implements Comparator<Comparable>  {
     SortOrder order = null;
+
     public ComparableComparator(SortOrder order) {
       this.order = order;
     }
     @Override
     public int compare(Comparable o1, Comparable o2) {
-      int result = ComparisonChain.start().compare(o1, o2).result();
-      return order == SortOrder.ASC?result:-1*result;
+      int result = ComparisonChain.start().compare(o1, o2, Ordering.natural().nullsLast()).result();
+      return order == SortOrder.ASC ? result : -1*result;
     }
   }
 
diff --git a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java
index 8f32946..4d3ff9b 100644
--- a/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java
+++ b/metron-platform/metron-indexing/src/test/java/org/apache/metron/indexing/dao/SearchIntegrationTest.java
@@ -18,31 +18,31 @@
 package org.apache.metron.indexing.dao;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import java.util.Iterator;
-import java.util.Optional;
 import org.adrianwalker.multilinestring.Multiline;
 import org.apache.metron.common.utils.JSONUtils;
 import org.apache.metron.indexing.dao.search.FieldType;
 import org.apache.metron.indexing.dao.search.GetRequest;
 import org.apache.metron.indexing.dao.search.GroupRequest;
 import org.apache.metron.indexing.dao.search.GroupResponse;
+import org.apache.metron.indexing.dao.search.GroupResult;
 import org.apache.metron.indexing.dao.search.InvalidSearchException;
 import org.apache.metron.indexing.dao.search.SearchRequest;
 import org.apache.metron.indexing.dao.search.SearchResponse;
 import org.apache.metron.indexing.dao.search.SearchResult;
-import org.apache.metron.indexing.dao.search.GroupResult;
 import org.apache.metron.indexing.dao.update.Document;
 import org.apache.metron.integration.InMemoryComponent;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
-import org.junit.*;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 public abstract class SearchIntegrationTest {
   /**
@@ -59,8 +59,8 @@
 
   /**
    * [
-   * {"source:type": "snort", "ip_src_addr":"192.168.1.6", "ip_src_port": 8005, "long_field": 10000, "timestamp":6, "latitude": 48.5839, "score": 50.0, "is_alert":false, "location_point": "50.0,7.7455", "snort_field": 10, "duplicate_name_field": 1, "guid":"snort_1"},
-   * {"source:type": "snort", "ip_src_addr":"192.168.1.1", "ip_src_port": 8004, "long_field": 10000, "timestamp":7, "latitude": 48.5839, "score": 10.0, "is_alert":true, "location_point": "48.5839,7.7455", "snort_field": 20, "duplicate_name_field": 2, "guid":"snort_2"},
+   * {"source:type": "snort", "ip_src_addr":"192.168.1.6", "ip_src_port": 8005, "long_field": 10000, "timestamp":6, "latitude": 48.5839, "score": 50.0, "is_alert":false, "location_point": "50.0,7.7455", "snort_field": 10, "duplicate_name_field": 1, "guid":"snort_1", "threat:triage:score":"10"},
+   * {"source:type": "snort", "ip_src_addr":"192.168.1.1", "ip_src_port": 8004, "long_field": 10000, "timestamp":7, "latitude": 48.5839, "score": 10.0, "is_alert":true, "location_point": "48.5839,7.7455", "snort_field": 20, "duplicate_name_field": 2, "guid":"snort_2", "threat:triage:score":"20"},
    * {"source:type": "snort", "ip_src_addr":"192.168.1.7", "ip_src_port": 8003, "long_field": 10000, "timestamp":8, "latitude": 48.5839, "score": 20.0, "is_alert":false, "location_point": "48.5839,7.7455", "snort_field": 30, "duplicate_name_field": 3, "guid":"snort_3"},
    * {"source:type": "snort", "ip_src_addr":"192.168.1.1", "ip_src_port": 8002, "long_field": 20000, "timestamp":9, "latitude": 48.0001, "score": 50.0, "is_alert":true, "location_point": "48.5839,7.7455", "snort_field": 40, "duplicate_name_field": 4, "guid":"snort_4"},
    * {"source:type": "snort", "ip_src_addr":"192.168.1.8", "ip_src_port": 8001, "long_field": 10000, "timestamp":10, "latitude": 48.5839, "score": 10.0, "is_alert":false, "location_point": "48.5839,7.7455", "snort_field": 50, "duplicate_name_field": 5, "guid":"snort_5"}
@@ -155,6 +155,46 @@
 
   /**
    * {
+   *  "indices": [
+   *    "snort",
+   *    "bro"
+   *  ],
+   * "query": "*",
+   * "from": 0,
+   * "size": 25,
+   * "sort": [
+   *    {
+   *      "field": "threat:triage:score",
+   *      "sortOrder": "asc"
+   *    }
+   *  ]
+   * }
+   */
+  @Multiline
+  public static String sortAscendingWithMissingFields;
+
+  /**
+   * {
+   *  "indices": [
+   *    "snort",
+   *    "bro"
+   *  ],
+   * "query": "*",
+   * "from": 0,
+   * "size": 25,
+   * "sort": [
+   *    {
+   *      "field": "threat:triage:score",
+   *      "sortOrder": "desc"
+   *    }
+   *  ]
+   * }
+   */
+  @Multiline
+  public static String sortDescendingWithMissingFields;
+
+  /**
+   * {
    * "indices": ["bro", "snort"],
    * "query": "*",
    * "from": 4,
@@ -407,6 +447,7 @@
       SearchResponse response = dao.search(request);
       Assert.assertEquals(10, response.getTotal());
       List<SearchResult> results = response.getResults();
+      Assert.assertEquals(10, results.size());
       for(int i = 0;i < 5;++i) {
         Assert.assertEquals("snort", results.get(i).getSource().get("source:type"));
         Assert.assertEquals(10-i, results.get(i).getSource().get("timestamp"));
@@ -461,6 +502,50 @@
         Assert.assertEquals(i, results.get(i-8001).getSource().get("ip_src_port"));
       }
     }
+    //Sort descending with missing fields
+    {
+      SearchRequest request = JSONUtils.INSTANCE.load(sortDescendingWithMissingFields, SearchRequest.class);
+      SearchResponse response = dao.search(request);
+      Assert.assertEquals(10, response.getTotal());
+      List<SearchResult> results = response.getResults();
+      Assert.assertEquals(10, results.size());
+
+      // validate sorted order - there are only 2 with a 'threat:triage:score'
+      Assert.assertEquals("20", results.get(0).getSource().get("threat:triage:score"));
+      Assert.assertEquals("10", results.get(1).getSource().get("threat:triage:score"));
+
+      // the remaining are missing the 'threat:triage:score' and should be sorted last
+      Assert.assertFalse(results.get(2).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(3).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(4).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(5).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(6).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(7).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(8).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(9).getSource().containsKey("threat:triage:score"));
+    }
+    //Sort ascending with missing fields
+    {
+      SearchRequest request = JSONUtils.INSTANCE.load(sortAscendingWithMissingFields, SearchRequest.class);
+      SearchResponse response = dao.search(request);
+      Assert.assertEquals(10, response.getTotal());
+      List<SearchResult> results = response.getResults();
+      Assert.assertEquals(10, results.size());
+
+      // the remaining are missing the 'threat:triage:score' and should be sorted last
+      Assert.assertFalse(results.get(0).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(1).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(2).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(3).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(4).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(5).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(6).getSource().containsKey("threat:triage:score"));
+      Assert.assertFalse(results.get(7).getSource().containsKey("threat:triage:score"));
+
+      // validate sorted order - there are only 2 with a 'threat:triage:score'
+      Assert.assertEquals("10", results.get(8).getSource().get("threat:triage:score"));
+      Assert.assertEquals("20", results.get(9).getSource().get("threat:triage:score"));
+    }
     //pagination test case
     {
       SearchRequest request = JSONUtils.INSTANCE.load(paginationQuery, SearchRequest.class);
@@ -490,13 +575,18 @@
     {
       SearchRequest request = JSONUtils.INSTANCE.load(facetQuery, SearchRequest.class);
       SearchResponse response = dao.search(request);
-      Assert.assertEquals(10, response.getTotal());
+      Assert.assertEquals(12, response.getTotal());
+
       Map<String, Map<String, Long>> facetCounts = response.getFacetCounts();
       Assert.assertEquals(8, facetCounts.size());
+
+      // source:type
       Map<String, Long> sourceTypeCounts = facetCounts.get("source:type");
       Assert.assertEquals(2, sourceTypeCounts.size());
       Assert.assertEquals(new Long(5), sourceTypeCounts.get("bro"));
       Assert.assertEquals(new Long(5), sourceTypeCounts.get("snort"));
+
+      // ip_src_addr
       Map<String, Long> ipSrcAddrCounts = facetCounts.get("ip_src_addr");
       Assert.assertEquals(8, ipSrcAddrCounts.size());
       Assert.assertEquals(new Long(3), ipSrcAddrCounts.get("192.168.1.1"));
@@ -507,6 +597,8 @@
       Assert.assertEquals(new Long(1), ipSrcAddrCounts.get("192.168.1.6"));
       Assert.assertEquals(new Long(1), ipSrcAddrCounts.get("192.168.1.7"));
       Assert.assertEquals(new Long(1), ipSrcAddrCounts.get("192.168.1.8"));
+
+      // ip_src_port
       Map<String, Long> ipSrcPortCounts = facetCounts.get("ip_src_port");
       Assert.assertEquals(10, ipSrcPortCounts.size());
       Assert.assertEquals(new Long(1), ipSrcPortCounts.get("8001"));
@@ -519,10 +611,14 @@
       Assert.assertEquals(new Long(1), ipSrcPortCounts.get("8008"));
       Assert.assertEquals(new Long(1), ipSrcPortCounts.get("8009"));
       Assert.assertEquals(new Long(1), ipSrcPortCounts.get("8010"));
+
+      // long_field
       Map<String, Long> longFieldCounts = facetCounts.get("long_field");
       Assert.assertEquals(2, longFieldCounts.size());
       Assert.assertEquals(new Long(8), longFieldCounts.get("10000"));
       Assert.assertEquals(new Long(2), longFieldCounts.get("20000"));
+
+      // timestamp
       Map<String, Long> timestampCounts = facetCounts.get("timestamp");
       Assert.assertEquals(10, timestampCounts.size());
       Assert.assertEquals(new Long(1), timestampCounts.get("1"));
@@ -535,6 +631,8 @@
       Assert.assertEquals(new Long(1), timestampCounts.get("8"));
       Assert.assertEquals(new Long(1), timestampCounts.get("9"));
       Assert.assertEquals(new Long(1), timestampCounts.get("10"));
+
+      // latitude
       Map<String, Long> latitudeCounts = facetCounts.get("latitude");
       Assert.assertEquals(2, latitudeCounts.size());
       List<String> latitudeKeys = new ArrayList<>(latitudeCounts.keySet());
@@ -543,6 +641,8 @@
       Assert.assertEquals(48.5839, Double.parseDouble(latitudeKeys.get(1)), 0.00001);
       Assert.assertEquals(new Long(2), latitudeCounts.get(latitudeKeys.get(0)));
       Assert.assertEquals(new Long(8), latitudeCounts.get(latitudeKeys.get(1)));
+
+      // score
       Map<String, Long> scoreFieldCounts = facetCounts.get("score");
       Assert.assertEquals(4, scoreFieldCounts.size());
       List<String> scoreFieldKeys = new ArrayList<>(scoreFieldCounts.keySet());
@@ -555,6 +655,8 @@
       Assert.assertEquals(new Long(2), scoreFieldCounts.get(scoreFieldKeys.get(1)));
       Assert.assertEquals(new Long(3), scoreFieldCounts.get(scoreFieldKeys.get(2)));
       Assert.assertEquals(new Long(1), scoreFieldCounts.get(scoreFieldKeys.get(3)));
+
+      // is_alert
       Map<String, Long> isAlertCounts = facetCounts.get("is_alert");
       Assert.assertEquals(2, isAlertCounts.size());
       Assert.assertEquals(new Long(6), isAlertCounts.get("true"));
@@ -568,7 +670,7 @@
         Assert.fail("Exception expected, but did not come.");
       }
       catch(InvalidSearchException ise) {
-        Assert.assertEquals("Could not execute search", ise.getMessage());
+        // success
       }
     }
     //Disabled facet query
@@ -591,7 +693,7 @@
     // getColumnMetadata with multiple indices
     {
       Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Arrays.asList("bro", "snort"));
-      Assert.assertEquals(13, fieldTypes.size());
+      Assert.assertEquals(15, fieldTypes.size());
       Assert.assertEquals(FieldType.STRING, fieldTypes.get("guid"));
       Assert.assertEquals(FieldType.STRING, fieldTypes.get("source:type"));
       Assert.assertEquals(FieldType.IP, fieldTypes.get("ip_src_addr"));
@@ -605,25 +707,51 @@
       Assert.assertEquals(FieldType.STRING, fieldTypes.get("bro_field"));
       Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("snort_field"));
       Assert.assertEquals(FieldType.OTHER, fieldTypes.get("duplicate_name_field"));
+      Assert.assertEquals(FieldType.FLOAT, fieldTypes.get("threat:triage:score"));
+      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("alert"));
     }
     // getColumnMetadata with only bro
     {
       Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Collections.singletonList("bro"));
-      Assert.assertEquals(12, fieldTypes.size());
+      Assert.assertEquals(13, fieldTypes.size());
+      Assert.assertEquals(FieldType.STRING, fieldTypes.get("guid"));
+      Assert.assertEquals(FieldType.STRING, fieldTypes.get("source:type"));
+      Assert.assertEquals(FieldType.IP, fieldTypes.get("ip_src_addr"));
+      Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("ip_src_port"));
+      Assert.assertEquals(FieldType.LONG, fieldTypes.get("long_field"));
+      Assert.assertEquals(FieldType.DATE, fieldTypes.get("timestamp"));
+      Assert.assertEquals(FieldType.FLOAT, fieldTypes.get("latitude"));
+      Assert.assertEquals(FieldType.DOUBLE, fieldTypes.get("score"));
+      Assert.assertEquals(FieldType.BOOLEAN, fieldTypes.get("is_alert"));
+      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("location_point"));
       Assert.assertEquals(FieldType.STRING, fieldTypes.get("bro_field"));
+      Assert.assertEquals(FieldType.STRING, fieldTypes.get("duplicate_name_field"));
+      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("alert"));
     }
     // getColumnMetadata with only snort
     {
       Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Collections.singletonList("snort"));
-      Assert.assertEquals(12, fieldTypes.size());
+      Assert.assertEquals(14, fieldTypes.size());
       Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("snort_field"));
+      Assert.assertEquals(FieldType.STRING, fieldTypes.get("guid"));
+      Assert.assertEquals(FieldType.STRING, fieldTypes.get("source:type"));
+      Assert.assertEquals(FieldType.IP, fieldTypes.get("ip_src_addr"));
+      Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("ip_src_port"));
+      Assert.assertEquals(FieldType.LONG, fieldTypes.get("long_field"));
+      Assert.assertEquals(FieldType.DATE, fieldTypes.get("timestamp"));
+      Assert.assertEquals(FieldType.FLOAT, fieldTypes.get("latitude"));
+      Assert.assertEquals(FieldType.DOUBLE, fieldTypes.get("score"));
+      Assert.assertEquals(FieldType.BOOLEAN, fieldTypes.get("is_alert"));
+      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("location_point"));
+      Assert.assertEquals(FieldType.INTEGER, fieldTypes.get("duplicate_name_field"));
+      Assert.assertEquals(FieldType.OTHER, fieldTypes.get("alert"));
     }
     // getColumnMetadata with an index that doesn't exist
     {
       Map<String, FieldType> fieldTypes = dao.getColumnMetadata(Collections.singletonList("someindex"));
       Assert.assertEquals(0, fieldTypes.size());
     }
-    //Fields query
+     //Fields query
     {
       SearchRequest request = JSONUtils.INSTANCE.load(fieldsQuery, SearchRequest.class);
       SearchResponse response = dao.search(request);
@@ -793,7 +921,7 @@
         Assert.fail("Exception expected, but did not come.");
       }
       catch(InvalidSearchException ise) {
-        Assert.assertEquals("Could not execute search", ise.getMessage());
+        // success
       }
     }
     //Group by IP query