Improved filtering by _parent field

In the _parent field the type and id of the parent are stored as type#id, because of this a term filter on the _parent field with the parent id is always resolved to a terms filter with a type / id combination for each type in the mapping.

This can be improved by automatically use the most optimized filter (either term or terms) based on the number of parent types in the mapping.

Also added support to use the parent type in the term filter for the _parent field. Like this:
```json
{
   "term" : {
        "_parent" : "parent_type#1"
    }
}
```
This will then always automatically use the term filter.

Closes #3454

src/main/java/org/elasticsearch/index/mapper/internal/ParentFieldMapper.java

@@ -22,6 +22,7 @@ package org.elasticsearch.index.mapper.internal;
 import org.apache.lucene.document.Field;
 import org.apache.lucene.document.FieldType;
 import org.apache.lucene.index.FieldInfo.IndexOptions;
+import org.apache.lucene.index.Term;
 import org.apache.lucene.queries.TermsFilter;
 import org.apache.lucene.search.ConstantScoreQuery;
 import org.apache.lucene.search.Filter;
@@ -31,6 +32,8 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.lucene.BytesRefs;
 import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.common.lucene.search.Queries;
+import org.elasticsearch.common.lucene.search.TermFilter;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.codec.postingsformat.PostingsFormatProvider;
@@ -239,13 +242,29 @@ public class ParentFieldMapper extends AbstractFieldMapper<Uid> implements Inter
             return super.termFilter(value, context);
         }
         BytesRef bValue = BytesRefs.toBytesRef(value);
-        // we use all types, cause we don't know if its exact or not...
-        BytesRef[] typesValues = new BytesRef[context.mapperService().types().size()];
-        int i = 0;
-        for (String type : context.mapperService().types()) {
-            typesValues[i++] = Uid.createUidAsBytes(type, bValue);
+        if (Uid.hasDelimiter(bValue)) {
+            return new TermFilter(new Term(names.indexName(), bValue));
+        }
+
+        List<String> parentTypes = new ArrayList<String>(context.mapperService().types().size());
+        for (DocumentMapper documentMapper : context.mapperService()) {
+            if (documentMapper.parentFieldMapper() == null) {
+                parentTypes.add(documentMapper.type());
+            }
+        }
+
+        if (parentTypes.isEmpty()) {
+            return Queries.MATCH_NO_FILTER;
+        } else if (parentTypes.size() == 1) {
+            return new TermFilter(new Term(names.indexName(), Uid.createUidAsBytes(parentTypes.get(0), bValue)));
+        } else {
+            // we use all types, cause we don't know if its exact or not...
+            List<BytesRef> typesValues = new ArrayList<BytesRef>(parentTypes.size());
+            for (String type : context.mapperService().types()) {
+                typesValues.add(Uid.createUidAsBytes(type, bValue));
+            }
+            return new TermsFilter(names.indexName(), typesValues);
         }
-        return new TermsFilter(names.indexName(), typesValues);
     }
 
     @Override

src/test/java/org/elasticsearch/test/integration/search/child/SimpleChildQuerySearchTests.java

@@ -46,6 +46,7 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.index.query.FilterBuilders.*;
 import static org.elasticsearch.index.query.QueryBuilders.*;
 import static org.elasticsearch.search.facet.FacetBuilders.termsFacet;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.*;
 
@@ -1514,7 +1515,6 @@ public class SimpleChildQuerySearchTests extends AbstractSharedClusterTest {
     @Test
     // See also issue: https://github.com/elasticsearch/elasticsearch/issues/3203
     public void testHasChildQueryWithMinimumScore() throws Exception {
-
         client().admin().indices().prepareCreate("test")
                 .setSettings(
                         ImmutableSettings.settingsBuilder()
@@ -1533,7 +1533,7 @@ public class SimpleChildQuerySearchTests extends AbstractSharedClusterTest {
         client().prepareIndex("test", "child", "c3").setSource("c_field", "x").setParent("p2").execute().actionGet();
         client().prepareIndex("test", "child", "c4").setSource("c_field", "x").setParent("p2").execute().actionGet();
         client().prepareIndex("test", "child", "c5").setSource("c_field", "x").setParent("p2").execute().actionGet();
-        client().admin().indices().prepareRefresh("test").execute().actionGet();
+        refresh();
 
         SearchResponse searchResponse = client().prepareSearch("test")
                 .setQuery(hasChildQuery("child", matchAllQuery()).scoreType("sum"))
@@ -1546,4 +1546,51 @@ public class SimpleChildQuerySearchTests extends AbstractSharedClusterTest {
         assertThat(searchResponse.getHits().getAt(0).score(), equalTo(3.0f));
     }
 
+    @Test
+    public void testParentFieldFilter() throws Exception {
+        client().admin().indices().prepareCreate("test")
+                .setSettings(
+                        ImmutableSettings.settingsBuilder()
+                                .put("index.number_of_shards", 1)
+                                .put("index.number_of_replicas", 1)
+                                .put("index.refresh_interval", -1)
+                ).execute().actionGet();
+        client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().execute().actionGet();
+        client().admin().indices().preparePutMapping("test").setType("child").setSource(jsonBuilder().startObject().startObject("type")
+                .startObject("_parent").field("type", "parent").endObject()
+                .endObject().endObject()).execute().actionGet();
+
+        SearchResponse response = client().prepareSearch("test").setQuery(filteredQuery(matchAllQuery(), termFilter("_parent", "p1")))
+                .execute().actionGet();
+        assertHitCount(response, 0l);
+
+        client().prepareIndex("test", "parent", "p1").setSource("p_field", "value").execute().actionGet();
+        client().prepareIndex("test", "child", "c1").setSource("c_field", "value").setParent("p1")
+                .execute().actionGet();
+
+        response = client().prepareSearch("test").setQuery(filteredQuery(matchAllQuery(), termFilter("_parent", "p1")))
+                .execute().actionGet();
+        assertHitCount(response, 0l);
+        refresh();
+
+        response = client().prepareSearch("test").setQuery(filteredQuery(matchAllQuery(), termFilter("_parent", "p1")))
+                .execute().actionGet();
+        assertHitCount(response, 1l);
+
+        response = client().prepareSearch("test").setQuery(filteredQuery(matchAllQuery(), termFilter("_parent", "parent#p1")))
+                .execute().actionGet();
+        assertHitCount(response, 1l);
+
+        client().prepareIndex("test", "parent2", "p1").setSource("p_field", "value")
+                .setRefresh(true).execute().actionGet();
+
+        response = client().prepareSearch("test").setQuery(filteredQuery(matchAllQuery(), termFilter("_parent", "p1")))
+                .execute().actionGet();
+        assertHitCount(response, 1l);
+
+        response = client().prepareSearch("test").setQuery(filteredQuery(matchAllQuery(), termFilter("_parent", "parent#p1")))
+                .execute().actionGet();
+        assertHitCount(response, 1l);
+    }
+
 }