The `fields` option should always return an array for json document fields and single valued field for metadata fields.

Also the `fields` option can only be used to fetch leaf fields, trying to do fetch object fields will return in a client error.

Closes #4542

docs/reference/docs/get.asciidoc

@@ -115,6 +115,12 @@ For backward compatibility, if the requested fields are not stored, they will be
 from the `_source` (parsed and extracted). This functionality has been replaced by the
 <<get-source-filtering,source filtering>> parameter.
 
+Field values fetched from the document it self are always returned as an array. Metadata fields like `_routing` and
+`_parent` fields are never returned as an array.
+
+Also only leaf fields can be returned via the `field` option. So object fields can't be returned and such requests
+will fail.
+
 [float]
 [[_source]]
 === Getting the _source directly

docs/reference/search/request/fields.asciidoc

@@ -34,9 +34,15 @@ For backwards compatibility, if the fields parameter specifies fields which are
 `false`), it will load the `_source` and extract it from it. This functionality has been replaced by the
 <<search-request-source-filtering,source filtering>> parameter.
 
+Field values fetched from the document it self are always returned as an array. Metadata fields like `_routing` and
+`_parent` fields are never returned as an array.
+
+Also only leaf fields can be returned via the `field` option. So object fields can't be returned and such requests
+will fail.
+
 Script fields can also be automatically detected and used as fields, so
-things like `_source.obj1.obj2` can be used, though not recommended, as
-`obj1.obj2` will work as well.
+things like `_source.obj1.field1` can be used, though not recommended, as
+`obj1.field1` will work as well.
 
 [[partial]]
 ==== Partial

src/main/java/org/elasticsearch/index/get/GetField.java

@@ -22,6 +22,7 @@ package org.elasticsearch.index.get;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.index.mapper.MapperService;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -34,11 +35,9 @@ import java.util.List;
 public class GetField implements Streamable, Iterable<Object> {
 
     private String name;
-
     private List<Object> values;
 
     private GetField() {
-
     }
 
     public GetField(String name, List<Object> values) {
@@ -61,6 +60,10 @@ public class GetField implements Streamable, Iterable<Object> {
         return values;
     }
 
+    public boolean isMetadataField() {
+        return MapperService.isMetadataField(name);
+    }
+
     @Override
     public Iterator<Object> iterator() {
         return values.iterator();

src/main/java/org/elasticsearch/index/get/GetResult.java

@@ -219,11 +219,11 @@ public class GetResult implements Streamable, Iterable<GetField>, ToXContent {
                 if (field.getValues().isEmpty()) {
                     continue;
                 }
-                if (field.getValues().size() == 1) {
-                    builder.field(field.getName(), field.getValues().get(0));
+                String fieldName = field.getName();
+                if (field.isMetadataField()) {
+                    builder.field(fieldName, field.getValue());
                 } else {
-                    builder.field(field.getName());
-                    builder.startArray();
+                    builder.startArray(field.getName());
                     for (Object value : field.getValues()) {
                         builder.value(value);
                     }

src/main/java/org/elasticsearch/index/get/ShardGetService.java

@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Sets;
 import org.apache.lucene.index.Term;
 import org.elasticsearch.ElasticSearchException;
+import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.Tuple;
@@ -268,19 +269,18 @@ public class ShardGetService extends AbstractIndexShardComponent {
                                 }
 
                                 FieldMapper<?> x = docMapper.mappers().smartNameFieldMapper(field);
-                                // only if the field is stored or source is enabled we should add it..
-                                if (docMapper.sourceMapper().enabled() || x == null || x.fieldType().stored()) {
-                                    value = searchLookup.source().extractValue(field);
-                                    // normalize the data if needed (mainly for binary fields, to convert from base64 strings to bytes)
-                                    if (value != null && x != null) {
-                                        if (value instanceof List) {
-                                            List list = (List) value;
-                                            for (int i = 0; i < list.size(); i++) {
-                                                list.set(i, x.valueForSearch(list.get(i)));
-                                            }
-                                        } else {
-                                            value = x.valueForSearch(value);
+                                if (x == null) {
+                                    if (docMapper.objectMappers().get(field) != null) {
+                                        // Only fail if we know it is a object field, missing paths / fields shouldn't fail.
+                                        throw new ElasticSearchIllegalArgumentException("field [" + field + "] isn't a leaf field");
+                                    }
+                                } else if (docMapper.sourceMapper().enabled() || x.fieldType().stored()) {
+                                    List<Object> values = searchLookup.source().extractRawValues(field);
+                                    if (!values.isEmpty()) {
+                                        for (int i = 0; i < values.size(); i++) {
+                                            values.set(i, x.valueForSearch(values.get(i)));
                                         }
+                                        value = values;
                                     }
                                 }
                             }
@@ -388,24 +388,25 @@ public class ShardGetService extends AbstractIndexShardComponent {
                     }
                 } else {
                     FieldMappers x = docMapper.mappers().smartName(field);
-                    if (x == null || !x.mapper().fieldType().stored()) {
+                    if (x == null) {
+                        if (docMapper.objectMappers().get(field) != null) {
+                            // Only fail if we know it is a object field, missing paths / fields shouldn't fail.
+                            throw new ElasticSearchIllegalArgumentException("field [" + field + "] isn't a leaf field");
+                        }
+                    } else if (!x.mapper().fieldType().stored()) {
                         if (searchLookup == null) {
                             searchLookup = new SearchLookup(mapperService, fieldDataService, new String[]{type});
                             searchLookup.setNextReader(docIdAndVersion.context);
                             searchLookup.source().setNextSource(source);
                             searchLookup.setNextDocId(docIdAndVersion.docId);
                         }
-                        value = searchLookup.source().extractValue(field);
-                        // normalize the data if needed (mainly for binary fields, to convert from base64 strings to bytes)
-                        if (value != null && x != null) {
-                            if (value instanceof List) {
-                                List list = (List) value;
-                                for (int i = 0; i < list.size(); i++) {
-                                    list.set(i, x.mapper().valueForSearch(list.get(i)));
-                                }
-                            } else {
-                                value = x.mapper().valueForSearch(value);
+
+                        List<Object> values = searchLookup.source().extractRawValues(field);
+                        if (!values.isEmpty()) {
+                            for (int i = 0; i < values.size(); i++) {
+                                values.set(i, x.mapper().valueForSearch(values.get(i)));
                             }
+                            value = values;
                         }
                     }
                 }

src/main/java/org/elasticsearch/index/mapper/MapperService.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.index.mapper;
 
+import com.carrotsearch.hppc.ObjectOpenHashSet;
 import com.google.common.base.Charsets;
 import com.google.common.collect.*;
 import org.apache.lucene.analysis.Analyzer;
@@ -75,6 +76,10 @@ import static org.elasticsearch.index.mapper.DocumentMapper.MergeFlags.mergeFlag
 public class MapperService extends AbstractIndexComponent implements Iterable<DocumentMapper> {
 
     public static final String DEFAULT_MAPPING = "_default_";
+    private static ObjectOpenHashSet<String> META_FIELDS = ObjectOpenHashSet.from(
+            "_uid", "_id", "_type", "_all", "_analyzer", "_boost", "_parent", "_routing", "_index",
+            "_size", "_timestamp", "_ttl"
+    );
 
     private final AnalysisService analysisService;
     private final IndexFieldDataService fieldDataService;
@@ -841,6 +846,13 @@ public class MapperService extends AbstractIndexComponent implements Iterable<Do
         return null;
     }
 
+    /**
+     * @return Whether a field is a metadata field.
+     */
+    public static boolean isMetadataField(String fieldName) {
+        return META_FIELDS.contains(fieldName);
+    }
+
     public static class SmartNameObjectMapper {
         private final ObjectMapper mapper;
         private final DocumentMapper docMapper;

src/main/java/org/elasticsearch/search/SearchHitField.java

@@ -59,4 +59,9 @@ public interface SearchHitField extends Streamable, Iterable<Object> {
      * The field values.
      */
     List<Object> getValues();
+
+    /**
+     * @return The field is a metadata field
+     */
+    boolean isMetadataField();
 }

src/main/java/org/elasticsearch/search/fetch/FetchPhase.java

@@ -22,6 +22,7 @@ package org.elasticsearch.search.fetch;
 import com.google.common.collect.ImmutableMap;
 import org.apache.lucene.index.AtomicReaderContext;
 import org.apache.lucene.index.ReaderUtil;
+import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.text.StringAndBytesText;
 import org.elasticsearch.common.text.Text;
@@ -117,7 +118,12 @@ public class FetchPhase implements SearchPhase {
                     continue;
                 }
                 FieldMappers x = context.smartNameFieldMappers(fieldName);
-                if (x != null && x.mapper().fieldType().stored()) {
+                if (x == null) {
+                    // Only fail if we know it is a object field, missing paths / fields shouldn't fail.
+                    if (context.smartNameObjectMapper(fieldName) != null) {
+                        throw new ElasticSearchIllegalArgumentException("field [" + fieldName + "] isn't a leaf field");
+                    }
+                } else if (x.mapper().fieldType().stored()) {
                     if (fieldNames == null) {
                         fieldNames = new HashSet<String>();
                     }
@@ -180,8 +186,8 @@ public class FetchPhase implements SearchPhase {
             }
             if (extractFieldNames != null) {
                 for (String extractFieldName : extractFieldNames) {
-                    Object value = context.lookup().source().extractValue(extractFieldName);
-                    if (value != null) {
+                    List<Object> values = context.lookup().source().extractRawValues(extractFieldName);
+                    if (!values.isEmpty()) {
                         if (searchHit.fieldsOrNull() == null) {
                             searchHit.fields(new HashMap<String, SearchHitField>(2));
                         }
@@ -191,7 +197,9 @@ public class FetchPhase implements SearchPhase {
                             hitField = new InternalSearchHitField(extractFieldName, new ArrayList<Object>(2));
                             searchHit.fields().put(extractFieldName, hitField);
                         }
-                        hitField.values().add(value);
+                        for (Object value : values) {
+                            hitField.values().add(value);
+                        }
                     }
                 }
             }

src/main/java/org/elasticsearch/search/internal/InternalSearchHit.java

@@ -415,12 +415,12 @@ public class InternalSearchHit implements SearchHit {
                 if (field.values().isEmpty()) {
                     continue;
                 }
-                if (field.values().size() == 1) {
-                    builder.field(field.name(), field.values().get(0));
+                String fieldName = field.getName();
+                if (field.isMetadataField()) {
+                    builder.field(fieldName, field.value());
                 } else {
-                    builder.field(field.name());
-                    builder.startArray();
-                    for (Object value : field.values()) {
+                    builder.startArray(fieldName);
+                    for (Object value : field.getValues()) {
                         builder.value(value);
                     }
                     builder.endArray();

src/main/java/org/elasticsearch/search/internal/InternalSearchHitField.java

@@ -21,6 +21,7 @@ package org.elasticsearch.search.internal;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.search.SearchHitField;
 
 import java.io.IOException;
@@ -34,11 +35,9 @@ import java.util.List;
 public class InternalSearchHitField implements SearchHitField {
 
     private String name;
-
     private List<Object> values;
 
     private InternalSearchHitField() {
-
     }
 
     public InternalSearchHitField(String name, List<Object> values) {
@@ -77,6 +76,10 @@ public class InternalSearchHitField implements SearchHitField {
         return values();
     }
 
+    @Override
+    public boolean isMetadataField() {
+        return MapperService.isMetadataField(name);
+    }
 
     @Override
     public Iterator<Object> iterator() {

src/test/java/org/elasticsearch/get/GetActionTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.get;
 
+import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.action.delete.DeleteResponse;
@@ -495,14 +496,14 @@ public class GetActionTests extends ElasticsearchIntegrationTest {
                         .endObject())
                 .execute().actionGet();
 
-        GetResponse responseBeforeFlush = client().prepareGet(index, type, "1").setFields("_source", "included", "excluded").execute().actionGet();
+        GetResponse responseBeforeFlush = client().prepareGet(index, type, "1").setFields("_source", "included.field", "excluded.field").execute().actionGet();
         assertThat(responseBeforeFlush.isExists(), is(true));
         assertThat(responseBeforeFlush.getSourceAsMap(), not(hasKey("excluded")));
         assertThat(responseBeforeFlush.getSourceAsMap(), not(hasKey("field")));
         assertThat(responseBeforeFlush.getSourceAsMap(), hasKey("included"));
 
         // now tests that extra source filtering works as expected
-        GetResponse responseBeforeFlushWithExtraFilters = client().prepareGet(index, type, "1").setFields("included", "excluded")
+        GetResponse responseBeforeFlushWithExtraFilters = client().prepareGet(index, type, "1").setFields("included.field", "excluded.field")
                 .setFetchSource(new String[]{"field", "*.field"}, new String[]{"*.field2"}).get();
         assertThat(responseBeforeFlushWithExtraFilters.isExists(), is(true));
         assertThat(responseBeforeFlushWithExtraFilters.getSourceAsMap(), not(hasKey("excluded")));
@@ -512,8 +513,8 @@ public class GetActionTests extends ElasticsearchIntegrationTest {
         assertThat((Map<String, Object>) responseBeforeFlushWithExtraFilters.getSourceAsMap().get("included"), not(hasKey("field2")));
 
         client().admin().indices().prepareFlush(index).execute().actionGet();
-        GetResponse responseAfterFlush = client().prepareGet(index, type, "1").setFields("_source", "included", "excluded").execute().actionGet();
-        GetResponse responseAfterFlushWithExtraFilters = client().prepareGet(index, type, "1").setFields("included", "excluded")
+        GetResponse responseAfterFlush = client().prepareGet(index, type, "1").setFields("_source", "included.field", "excluded.field").execute().actionGet();
+        GetResponse responseAfterFlushWithExtraFilters = client().prepareGet(index, type, "1").setFields("included.field", "excluded.field")
                 .setFetchSource("*.field", "*.field2").get();
 
         assertThat(responseAfterFlush.isExists(), is(true));
@@ -730,4 +731,135 @@ public class GetActionTests extends ElasticsearchIntegrationTest {
         assertThat(response.getResponses()[2].getResponse().getSourceAsMap().get("field").toString(), equalTo("value2"));
     }
 
+    @Test
+    public void testGetFields_metaData() throws Exception {
+        client().admin().indices().prepareCreate("my-index")
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .get();
+
+        client().prepareIndex("my-index", "my-type1", "1")
+                .setRouting("1")
+                .setSource(jsonBuilder().startObject().field("field1", "value").endObject())
+                .get();
+
+        GetResponse getResponse = client().prepareGet("my-index", "my-type1", "1")
+                .setRouting("1")
+                .setFields("field1", "_routing")
+                .get();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getField("field1").isMetadataField(), equalTo(false));
+        assertThat(getResponse.getField("field1").getValue().toString(), equalTo("value"));
+        assertThat(getResponse.getField("_routing").isMetadataField(), equalTo(true));
+        assertThat(getResponse.getField("_routing").getValue().toString(), equalTo("1"));
+
+        client().admin().indices().prepareFlush("my-index").get();
+
+        client().prepareGet("my-index", "my-type1", "1")
+                .setFields("field1", "_routing")
+                .setRouting("1")
+                .get();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getField("field1").isMetadataField(), equalTo(false));
+        assertThat(getResponse.getField("field1").getValue().toString(), equalTo("value"));
+        assertThat(getResponse.getField("_routing").isMetadataField(), equalTo(true));
+        assertThat(getResponse.getField("_routing").getValue().toString(), equalTo("1"));
+    }
+
+    @Test
+    public void testGetFields_nonLeafField() throws Exception {
+        client().admin().indices().prepareCreate("my-index")
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .get();
+
+        client().prepareIndex("my-index", "my-type1", "1")
+                .setSource(jsonBuilder().startObject().startObject("field1").field("field2", "value1").endObject().endObject())
+                .get();
+
+        try {
+            client().prepareGet("my-index", "my-type1", "1").setFields("field1").get();
+            assert false;
+        } catch (ElasticSearchIllegalArgumentException e) {}
+
+        client().admin().indices().prepareFlush("my-index").get();
+
+        try {
+            client().prepareGet("my-index", "my-type1", "1").setFields("field1").get();
+            assert false;
+        } catch (ElasticSearchIllegalArgumentException e) {}
+    }
+
+    @Test
+    public void testGetFields_complexField() throws Exception {
+        client().admin().indices().prepareCreate("my-index")
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .addMapping("my-type2", jsonBuilder().startObject().startObject("my-type2").startObject("properties")
+                        .startObject("field1").field("type", "object")
+                            .startObject("field2").field("type", "object")
+                                .startObject("field3").field("type", "object")
+                                    .startObject("field4").field("type", "string").field("store", "yes")
+                                .endObject()
+                            .endObject()
+                        .endObject()
+                        .endObject().endObject().endObject())
+                .get();
+
+        BytesReference source = jsonBuilder().startObject()
+                    .startArray("field1")
+                        .startObject()
+                            .startObject("field2")
+                                .startArray("field3")
+                                    .startObject()
+                                        .field("field4", "value1")
+                                    .endObject()
+                                .endArray()
+                            .endObject()
+                        .endObject()
+                        .startObject()
+                            .startObject("field2")
+                                .startArray("field3")
+                                    .startObject()
+                                        .field("field4", "value2")
+                                    .endObject()
+                                .endArray()
+                            .endObject()
+                        .endObject()
+                    .endArray()
+                .endObject().bytes();
+
+        client().prepareIndex("my-index", "my-type1", "1").setSource(source).get();
+        client().prepareIndex("my-index", "my-type2", "1").setSource(source).get();
+
+
+        String field = "field1.field2.field3.field4";
+        GetResponse getResponse = client().prepareGet("my-index", "my-type1", "1").setFields(field).get();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getField(field).isMetadataField(), equalTo(false));
+        assertThat(getResponse.getField(field).getValues().size(), equalTo(2));
+        assertThat(getResponse.getField(field).getValues().get(0).toString(), equalTo("value1"));
+        assertThat(getResponse.getField(field).getValues().get(1).toString(), equalTo("value2"));
+
+        getResponse = client().prepareGet("my-index", "my-type2", "1").setFields(field).get();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getField(field).isMetadataField(), equalTo(false));
+        assertThat(getResponse.getField(field).getValues().size(), equalTo(2));
+        assertThat(getResponse.getField(field).getValues().get(0).toString(), equalTo("value1"));
+        assertThat(getResponse.getField(field).getValues().get(1).toString(), equalTo("value2"));
+
+        client().admin().indices().prepareFlush("my-index").get();
+
+        getResponse = client().prepareGet("my-index", "my-type1", "1").setFields(field).get();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getField(field).isMetadataField(), equalTo(false));
+        assertThat(getResponse.getField(field).getValues().size(), equalTo(2));
+        assertThat(getResponse.getField(field).getValues().get(0).toString(), equalTo("value1"));
+        assertThat(getResponse.getField(field).getValues().get(1).toString(), equalTo("value2"));
+
+        getResponse = client().prepareGet("my-index", "my-type2", "1").setFields(field).get();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getField(field).isMetadataField(), equalTo(false));
+        assertThat(getResponse.getField(field).getValues().size(), equalTo(2));
+        assertThat(getResponse.getField(field).getValues().get(0).toString(), equalTo("value1"));
+        assertThat(getResponse.getField(field).getValues().get(1).toString(), equalTo("value2"));
+    }
+
 }

src/test/java/org/elasticsearch/search/fields/SearchFieldsTests.java

@@ -29,6 +29,7 @@ import org.elasticsearch.common.joda.Joda;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.joda.time.DateTime;
@@ -331,4 +332,95 @@ public class SearchFieldsTests extends ElasticsearchIntegrationTest {
         assertThat(((BytesReference) searchResponse.getHits().getAt(0).fields().get("binary_field").value()).toBytesArray(), equalTo((BytesReference) new BytesArray("testing text".getBytes("UTF8"))));
 
     }
+
+    @Test
+    public void testSearchFields_metaData() throws Exception {
+        client().prepareIndex("my-index", "my-type1", "1")
+                .setRouting("1")
+                .setSource(jsonBuilder().startObject().field("field1", "value").endObject())
+                .setRefresh(true)
+                .get();
+
+        SearchResponse searchResponse = client().prepareSearch("my-index")
+                .setTypes("my-type1")
+                .addField("field1").addField("_routing")
+                .get();
+
+        assertThat(searchResponse.getHits().totalHits(), equalTo(1l));
+        assertThat(searchResponse.getHits().getAt(0).field("field1").isMetadataField(), equalTo(false));
+        assertThat(searchResponse.getHits().getAt(0).field("field1").getValue().toString(), equalTo("value"));
+        assertThat(searchResponse.getHits().getAt(0).field("_routing").isMetadataField(), equalTo(true));
+        assertThat(searchResponse.getHits().getAt(0).field("_routing").getValue().toString(), equalTo("1"));
+    }
+
+    @Test
+    public void testSearchFields_nonLeafField() throws Exception {
+        client().prepareIndex("my-index", "my-type1", "1")
+                .setSource(jsonBuilder().startObject().startObject("field1").field("field2", "value1").endObject().endObject())
+                .setRefresh(true)
+                .get();
+
+        SearchResponse searchResponse = client().prepareSearch("my-index").setTypes("my-type1").addField("field1").get();
+        assertThat(searchResponse.getShardFailures().length, equalTo(1));
+        assertThat(searchResponse.getShardFailures()[0].status(), equalTo(RestStatus.BAD_REQUEST));
+        assertThat(searchResponse.getShardFailures()[0].reason(), containsString("field [field1] isn't a leaf field"));
+    }
+
+    @Test
+    public void testGetFields_complexField() throws Exception {
+        client().admin().indices().prepareCreate("my-index")
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .addMapping("my-type2", jsonBuilder().startObject().startObject("my-type2").startObject("properties")
+                        .startObject("field1").field("type", "object")
+                        .startObject("field2").field("type", "object")
+                        .startObject("field3").field("type", "object")
+                        .startObject("field4").field("type", "string").field("store", "yes")
+                        .endObject()
+                        .endObject()
+                        .endObject()
+                        .endObject().endObject().endObject())
+                .get();
+
+        BytesReference source = jsonBuilder().startObject()
+                .startArray("field1")
+                    .startObject()
+                        .startObject("field2")
+                            .startArray("field3")
+                                .startObject()
+                                    .field("field4", "value1")
+                                .endObject()
+                            .endArray()
+                        .endObject()
+                    .endObject()
+                    .startObject()
+                        .startObject("field2")
+                            .startArray("field3")
+                                .startObject()
+                                    .field("field4", "value2")
+                                .endObject()
+                            .endArray()
+                        .endObject()
+                    .endObject()
+                .endArray()
+                .endObject().bytes();
+
+        client().prepareIndex("my-index", "my-type1", "1").setSource(source).get();
+        client().prepareIndex("my-index", "my-type2", "1").setRefresh(true).setSource(source).get();
+
+
+        String field = "field1.field2.field3.field4";
+        SearchResponse searchResponse = client().prepareSearch("my-index").setTypes("my-type1").addField(field).get();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(1l));
+        assertThat(searchResponse.getHits().getAt(0).field(field).isMetadataField(), equalTo(false));
+        assertThat(searchResponse.getHits().getAt(0).field(field).getValues().size(), equalTo(2));
+        assertThat(searchResponse.getHits().getAt(0).field(field).getValues().get(0).toString(), equalTo("value1"));
+        assertThat(searchResponse.getHits().getAt(0).field(field).getValues().get(1).toString(), equalTo("value2"));
+
+        searchResponse = client().prepareSearch("my-index").setTypes("my-type2").addField(field).get();
+        assertThat(searchResponse.getHits().totalHits(), equalTo(1l));
+        assertThat(searchResponse.getHits().getAt(0).field(field).isMetadataField(), equalTo(false));
+        assertThat(searchResponse.getHits().getAt(0).field(field).getValues().size(), equalTo(2));
+        assertThat(searchResponse.getHits().getAt(0).field(field).getValues().get(0).toString(), equalTo("value1"));
+        assertThat(searchResponse.getHits().getAt(0).field(field).getValues().get(1).toString(), equalTo("value2"));
+    }
 }