Support source include/exclude for realtime GET

Currently realtime GET does not take source includes/excludes into account.
This patch adds support for the source field mapper includes/excludes
when getting an entry from the transaction log. Even though it introduces
a slight performance penalty, it now adheres to the defined configuration
instead of returning all source data when a realtime get is done.

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

@@ -23,11 +23,16 @@ import com.google.common.collect.Sets;
 import org.apache.lucene.index.Term;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.lucene.uid.UidField;
 import org.elasticsearch.common.metrics.CounterMetric;
 import org.elasticsearch.common.metrics.MeanMetric;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.fielddata.IndexFieldDataService;
 import org.elasticsearch.index.fieldvisitor.CustomFieldsVisitor;
@@ -270,7 +275,27 @@ public class ShardGetService extends AbstractIndexShardComponent {
                     sourceRequested = false;
                 }
 
-                return new GetResult(shardId.index().name(), type, id, get.version(), get.exists(), sourceRequested ? source.source : null, fields);
+                // Cater for source excludes/includes at the cost of performance
+                BytesReference sourceToBeReturned = null;
+                if (sourceRequested) {
+                    sourceToBeReturned = source.source;
+
+                    SourceFieldMapper sourceFieldMapper = docMapper.sourceMapper();
+                    if (sourceFieldMapper.enabled()) {
+                        boolean filtered = sourceFieldMapper.includes().length > 0 || sourceFieldMapper.excludes().length > 0;
+                        if (filtered) {
+                            Tuple<XContentType, Map<String, Object>> mapTuple = XContentHelper.convertToMap(source.source, true);
+                            Map<String, Object> filteredSource = XContentMapValues.filter(mapTuple.v2(), sourceFieldMapper.includes(), sourceFieldMapper.excludes());
+                            try {
+                                sourceToBeReturned = XContentFactory.contentBuilder(mapTuple.v1()).map(filteredSource).bytes();
+                            } catch (IOException e) {
+                                throw new ElasticSearchException("Failed to get type [" + type + "] and id [" + id + "] with includes/excludes set", e);
+                            }
+                        }
+                    }
+                }
+
+                return new GetResult(shardId.index().name(), type, id, get.version(), get.exists(), sourceToBeReturned, fields);
             }
         } finally {
             get.release();

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

@@ -214,6 +214,14 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
         return this.enabled;
     }
 
+    public String[] excludes() {
+        return this.excludes;
+
+    }
+    public String[] includes() {
+        return this.includes;
+    }
+
     @Override
     public FieldType defaultFieldType() {
         return Defaults.FIELD_TYPE;

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

@@ -381,4 +381,118 @@ public class GetActionTests extends AbstractNodesTests {
         assertThat(((List) response.getFields().get("field").getValues().get(0)).get(1).toString(), equalTo("2"));
     }
 
+    @Test
+    public void testThatGetFromTranslogShouldWorkWithExclude() throws Exception {
+        client.admin().indices().prepareDelete().execute().actionGet();
+        String index = "test";
+        String type = "type1";
+
+        String mapping = jsonBuilder()
+                .startObject()
+                    .startObject("source_excludes")
+                        .startObject("_source")
+                            .array("excludes", "excluded")
+                        .endObject()
+                    .endObject()
+                .endObject()
+            .string();
+
+        client.admin().indices().prepareCreate(index)
+                .addMapping(type, mapping)
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .execute().actionGet();
+
+        client.prepareIndex(index, type, "1")
+                .setSource(jsonBuilder().startObject().field("field", "1", "2").field("excluded", "should not be seen").endObject())
+                .execute().actionGet();
+
+        GetResponse responseBeforeFlush = client.prepareGet(index, type, "1").execute().actionGet();
+        client.admin().indices().prepareFlush(index).execute().actionGet();
+        GetResponse responseAfterFlush = client.prepareGet(index, type, "1").execute().actionGet();
+
+        assertThat(responseBeforeFlush.isExists(), is(true));
+        assertThat(responseAfterFlush.isExists(), is(true));
+        assertThat(responseBeforeFlush.getSourceAsMap(), hasKey("field"));
+        assertThat(responseBeforeFlush.getSourceAsMap(), not(hasKey("excluded")));
+        assertThat(responseBeforeFlush.getSourceAsString(), is(responseAfterFlush.getSourceAsString()));
+    }
+
+    @Test
+    public void testThatGetFromTranslogShouldWorkWithInclude() throws Exception {
+        client.admin().indices().prepareDelete().execute().actionGet();
+        String index = "test";
+        String type = "type1";
+
+        String mapping = jsonBuilder()
+            .startObject()
+                .startObject("source_excludes")
+                    .startObject("_source")
+                        .array("includes", "included")
+                    .endObject()
+                .endObject()
+            .endObject()
+            .string();
+
+        client.admin().indices().prepareCreate(index)
+                .addMapping(type, mapping)
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .execute().actionGet();
+
+        client.prepareIndex(index, type, "1")
+                .setSource(jsonBuilder().startObject().field("field", "1", "2").field("included", "should be seen").endObject())
+                .execute().actionGet();
+
+        GetResponse responseBeforeFlush = client.prepareGet(index, type, "1").execute().actionGet();
+        client.admin().indices().prepareFlush(index).execute().actionGet();
+        GetResponse responseAfterFlush = client.prepareGet(index, type, "1").execute().actionGet();
+
+        assertThat(responseBeforeFlush.isExists(), is(true));
+        assertThat(responseAfterFlush.isExists(), is(true));
+        assertThat(responseBeforeFlush.getSourceAsMap(), not(hasKey("field")));
+        assertThat(responseBeforeFlush.getSourceAsMap(), hasKey("included"));
+        assertThat(responseBeforeFlush.getSourceAsString(), is(responseAfterFlush.getSourceAsString()));
+    }
+
+    @Test
+    public void testThatGetFromTranslogShouldWorkWithIncludeExcludeAndFields() throws Exception {
+        client.admin().indices().prepareDelete().execute().actionGet();
+        String index = "test";
+        String type = "type1";
+
+        String mapping = jsonBuilder()
+            .startObject()
+                .startObject("source_excludes")
+                    .startObject("_source")
+                        .array("includes", "included")
+                        .array("exlcudes", "excluded")
+                    .endObject()
+                .endObject()
+            .endObject()
+            .string();
+
+        client.admin().indices().prepareCreate(index)
+                .addMapping(type, mapping)
+                .setSettings(ImmutableSettings.settingsBuilder().put("index.refresh_interval", -1))
+                .execute().actionGet();
+
+        client.prepareIndex(index, type, "1")
+                .setSource(jsonBuilder().startObject()
+                        .field("field", "1", "2")
+                        .field("included", "should be seen")
+                        .field("excluded", "should not be seen")
+                    .endObject())
+                .execute().actionGet();
+
+        GetResponse responseBeforeFlush = client.prepareGet(index, type, "1").setFields("_source", "included", "excluded").execute().actionGet();
+        client.admin().indices().prepareFlush(index).execute().actionGet();
+        GetResponse responseAfterFlush = client.prepareGet(index, type, "1").setFields("_source", "included", "excluded").execute().actionGet();
+
+        assertThat(responseBeforeFlush.isExists(), is(true));
+        assertThat(responseAfterFlush.isExists(), is(true));
+        assertThat(responseBeforeFlush.getSourceAsMap(), not(hasKey("excluded")));
+        assertThat(responseBeforeFlush.getSourceAsMap(), not(hasKey("field")));
+        assertThat(responseBeforeFlush.getSourceAsMap(), hasKey("included"));
+        assertThat(responseBeforeFlush.getSourceAsString(), is(responseAfterFlush.getSourceAsString()));
+    }
+
 }