Allow to update the _source mapping exclude/include dynamically when we merge mappings.

Closes #3491

src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java

@@ -372,8 +372,8 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
                 }
 
                 // update the current cluster state
-                logger.debug("Updating cluster state version {}", newClusterState.version());
                 clusterState = newClusterState;
+                logger.debug("Set cluster state to version {}. Broadcasting to listeners.", newClusterState.version());
 
                 for (ClusterStateListener listener : priorityClusterStateListeners) {
                     listener.clusterChanged(clusterChangedEvent);
@@ -401,7 +401,7 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
                     ((ProcessedClusterStateUpdateTask) updateTask).clusterStateProcessed(source, previousClusterState, newClusterState);
                 }
 
-                logger.debug("processing [{}]: done applying updated cluster_state", source);
+                logger.debug("processing [{}]: done applying updated cluster_state (version: {})", source, newClusterState.version());
             } catch (Exception e) {
                 StringBuilder sb = new StringBuilder("failed to apply updated cluster state:\nversion [").append(newClusterState.version()).append("], source [").append(source).append("]\n");
                 sb.append(newClusterState.nodes().prettyPrint());

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

@@ -182,11 +182,9 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
     private final boolean enabled;
 
     private Boolean compress;
-
     private long compressThreshold;
 
     private String[] includes;
-
     private String[] excludes;
 
     private String format;
@@ -406,6 +404,12 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
             if (sourceMergeWith.compressThreshold != -1) {
                 this.compressThreshold = sourceMergeWith.compressThreshold;
             }
+            if (sourceMergeWith.includes != null) {
+                this.includes = sourceMergeWith.includes;
+            }
+            if (sourceMergeWith.excludes != null) {
+                this.excludes = sourceMergeWith.excludes;
+            }
         }
     }
 }

src/test/java/org/elasticsearch/test/integration/AbstractSharedClusterTest.java

@@ -325,6 +325,10 @@ public abstract class AbstractSharedClusterTest extends ElasticsearchTestCase {
         return client().prepareGet(index, type, id).execute().actionGet();
     }
 
+    protected IndexResponse index(String index, String type, String id, XContentBuilder source) {
+        return client().prepareIndex(index, type, id).setSource(source).execute().actionGet();
+    }
+
     protected IndexResponse index(String index, String type, String id, String field, Object value) {
         return client().prepareIndex(index, type, id).setSource(field, value).execute().actionGet();
     }
@@ -393,7 +397,7 @@ public abstract class AbstractSharedClusterTest extends ElasticsearchTestCase {
     }
 
     // TODO move this into a base class for integration tests
-    public void indexRandom(String index, boolean forceRefresh, IndexRequestBuilder...builders) throws InterruptedException, ExecutionException {
+    public void indexRandom(String index, boolean forceRefresh, IndexRequestBuilder... builders) throws InterruptedException, ExecutionException {
         Random random = getRandom();
         List<IndexRequestBuilder> list = Arrays.asList(builders);
         Collections.shuffle(list, random);

src/test/java/org/elasticsearch/test/integration/indices/mapping/UpdateMappingTests.java

@@ -4,6 +4,8 @@ import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
 import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.count.CountResponse;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.cluster.metadata.MappingMetaData;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
@@ -109,6 +111,68 @@ public class UpdateMappingTests extends AbstractSharedClusterTest {
     }
 
 
+    @SuppressWarnings("unchecked")
+    @Test
+    public void updateIncludeExclude() throws Exception {
+        createIndexMapped("test", "type", "normal", "long", "exclude", "long", "include", "long");
+
+        logger.info("Index doc 1");
+        index("test", "type", "1", JsonXContent.contentBuilder().startObject()
+                .field("normal", 1).field("exclude", 1).field("include", 1)
+                .endObject()
+        );
+        refresh(); // commit it for later testing.
+
+
+        logger.info("Adding exclude settings");
+        PutMappingResponse putResponse = client().admin().indices().preparePutMapping("test").setType("type").setSource(
+                JsonXContent.contentBuilder().startObject().startObject("type")
+                        .startObject("_source")
+                        .startArray("excludes").value("exclude").endArray()
+                        .endObject().endObject()
+        ).get();
+
+        assertTrue(putResponse.isAcknowledged());
+
+        logger.info("Index doc 2");
+        index("test", "type", "2", JsonXContent.contentBuilder().startObject()
+                .field("normal", 2).field("exclude", 1).field("include", 2)
+                .endObject()
+        );
+
+        GetResponse getResponse = get("test", "type", "2");
+        assertThat(getResponse.getSource(), hasKey("normal"));
+        assertThat(getResponse.getSource(), not(hasKey("exclude")));
+        assertThat(getResponse.getSource(), hasKey("include"));
+
+
+        putResponse = client().admin().indices().preparePutMapping("test").setType("type").setSource(
+                JsonXContent.contentBuilder().startObject().startObject("type")
+                        .startObject("_source")
+                        .startArray("excludes").endArray()
+                        .startArray("includes").value("include").endArray()
+                        .endObject().endObject()
+        ).get();
+        assertTrue(putResponse.isAcknowledged());
+
+        GetMappingsResponse getMappingsResponse = client().admin().indices().prepareGetMappings("test").get();
+        MappingMetaData typeMapping = getMappingsResponse.getMappings().get("test").get("type");
+        assertThat((Map<String, Object>) typeMapping.getSourceAsMap().get("_source"), hasKey("includes"));
+        assertThat((Map<String, Object>) typeMapping.getSourceAsMap().get("_source"), not(hasKey("excludes")));
+
+
+        index("test", "type", "3", JsonXContent.contentBuilder().startObject()
+                .field("normal", 3).field("exclude", 3).field("include", 3)
+                .endObject()
+        );
+
+        getResponse = get("test", "type", "3");
+        assertThat(getResponse.getSource(), not(hasKey("normal")));
+        assertThat(getResponse.getSource(), not(hasKey("exclude")));
+        assertThat(getResponse.getSource(), hasKey("include"));
+
+    }
+
     @SuppressWarnings("unchecked")
     @Test
     public void updateDefaultMappingSettings() throws Exception {