diff --git a/oak-core/pom.xml b/oak-core/pom.xml
index 007f49b..b3562cb 100644
--- a/oak-core/pom.xml
+++ b/oak-core/pom.xml
@@ -270,6 +270,12 @@
 
     <!-- Test Dependencies -->
     <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-all</artifactId>
+      <version>1.3</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
diff --git a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSet.java b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSet.java
index 5f15255..b2b9703 100644
--- a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSet.java
+++ b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSet.java
@@ -23,6 +23,11 @@
 import javax.annotation.CheckForNull;
 
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
+import org.apache.jackrabbit.oak.commons.json.JsopBuilder;
+import org.apache.jackrabbit.oak.commons.json.JsopReader;
+import org.apache.jackrabbit.oak.commons.json.JsopTokenizer;
+import org.apache.jackrabbit.oak.commons.json.JsopWriter;
 
 /**
  * A ChangeSet is a collection of items that have been changed as part of a
@@ -53,7 +58,7 @@
  * Naming: note that path, node name and node types all refer to the *parent* of
  * a change. While properties naturally are leafs.
  */
-public class ChangeSet {
+public final class ChangeSet {
 
     private final int maxPathDepth;
     private final Set<String> parentPaths;
@@ -116,4 +121,105 @@ public boolean anyOverflow() {
                 getPropertyNames() == null;
     }
 
+    //~---------------------------------------------------< equals/hashcode >
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        ChangeSet changeSet = (ChangeSet) o;
+
+        if (maxPathDepth != changeSet.maxPathDepth) return false;
+        if (parentPaths != null ? !parentPaths.equals(changeSet.parentPaths) : changeSet.parentPaths != null)
+            return false;
+        if (parentNodeNames != null ? !parentNodeNames.equals(changeSet.parentNodeNames) : changeSet.parentNodeNames != null)
+            return false;
+        if (parentNodeTypes != null ? !parentNodeTypes.equals(changeSet.parentNodeTypes) : changeSet.parentNodeTypes != null)
+            return false;
+        if (propertyNames != null ? !propertyNames.equals(changeSet.propertyNames) : changeSet.propertyNames != null)
+            return false;
+        return allNodeTypes != null ? allNodeTypes.equals(changeSet.allNodeTypes) : changeSet.allNodeTypes == null;
+    }
+
+    @Override
+    public int hashCode() {
+        return 0;
+    }
+
+    //~----------------------------------------------------< json support >
+
+    public String asString(){
+        JsopWriter json = new JsopBuilder();
+        json.object();
+        json.key("maxPathDepth").value(maxPathDepth);
+        addToJson(json, "parentPaths", parentPaths);
+        addToJson(json, "parentNodeNames", parentNodeNames);
+        addToJson(json, "parentNodeTypes", parentNodeTypes);
+        addToJson(json, "propertyNames", propertyNames);
+        addToJson(json, "allNodeTypes", allNodeTypes);
+        json.endObject();
+        return json.toString();
+    }
+
+    public static ChangeSet fromString(String json) {
+        JsopReader reader = new JsopTokenizer(json);
+        int maxPathDepth = 0;
+        Set<String> parentPaths = null;
+        Set<String> parentNodeNames = null;
+        Set<String> parentNodeTypes = null;
+        Set<String> propertyNames = null;
+        Set<String> allNodeTypes = null;
+
+        reader.read('{');
+        if (!reader.matches('}')) {
+            do {
+                String name = reader.readString();
+                reader.read(':');
+                if ("maxPathDepth".equals(name)){
+                    maxPathDepth = Integer.parseInt(reader.read(JsopReader.NUMBER));
+                } else {
+                    Set<String> data = readArrayAsSet(reader);
+                    if ("parentPaths".equals(name)){
+                        parentPaths = data;
+                    } else if ("parentNodeNames".equals(name)){
+                        parentNodeNames = data;
+                    } else if ("parentNodeTypes".equals(name)){
+                        parentNodeTypes = data;
+                    } else if ("propertyNames".equals(name)){
+                        propertyNames = data;
+                    } else if ("allNodeTypes".equals(name)){
+                        allNodeTypes = data;
+                    }
+                }
+            } while (reader.matches(','));
+            reader.read('}');
+        }
+        reader.read(JsopReader.END);
+        return new ChangeSet(maxPathDepth, parentPaths, parentNodeNames, parentNodeTypes, propertyNames, allNodeTypes);
+
+    }
+
+    private static Set<String> readArrayAsSet(JsopReader reader) {
+        Set<String> values = Sets.newHashSet();
+        reader.read('[');
+        for (boolean first = true; !reader.matches(']'); first = false) {
+            if (!first) {
+                reader.read(',');
+            }
+            values.add(reader.readString());
+        }
+        return values;
+    }
+
+    private static void addToJson(JsopWriter json, String name, Set<String> values){
+        if (values == null){
+            return;
+        }
+        json.key(name).array();
+        for (String v : values){
+            json.value(v);
+        }
+        json.endArray();
+    }
 }
\ No newline at end of file
diff --git a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilder.java b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilder.java
index a590300..0c76eec 100644
--- a/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilder.java
+++ b/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilder.java
@@ -21,6 +21,7 @@
 import java.util.Set;
 
 import com.google.common.collect.Sets;
+import org.apache.jackrabbit.oak.commons.PathUtils;
 
 /**
  * Builder of a ChangeSet - only used by ChangeCollectorProvider (and tests..)
@@ -28,7 +29,7 @@
 public class ChangeSetBuilder {
 
     private final int maxItems;
-    private final int maxPathDepth;
+    private int maxPathDepth;
     private final Set<String> parentPaths = Sets.newHashSet();
     private final Set<String> parentNodeNames = Sets.newHashSet();
     private final Set<String> parentNodeTypes = Sets.newHashSet();
@@ -58,6 +59,9 @@ public boolean isParentPathOverflown() {
     }
 
     public ChangeSetBuilder addParentPath(String path){
+        if (PathUtils.getDepth(path) > maxPathDepth){
+            return this;
+        }
         parentPathOverflow = addAndCheckOverflow(parentPaths, path, maxItems, parentPathOverflow);
         return this;
     }
@@ -115,6 +119,48 @@ public int getMaxPrefilterPathDepth() {
         return maxPathDepth;
     }
 
+    public ChangeSetBuilder add(ChangeSet cs){
+        if (cs.getParentPaths() == null){
+            parentPathOverflow = true;
+        } else {
+            addPathFromChangeSet(cs);
+        }
+
+        if (cs.getParentNodeNames() == null){
+            parentNodeNameOverflow = true;
+        } else {
+            for (String parentNodeName : cs.getParentNodeNames()){
+                addParentNodeName(parentNodeName);
+            }
+        }
+
+        if (cs.getParentNodeTypes() == null){
+            parentNodeTypeOverflow = true;
+        } else {
+            for (String parentNodeType : cs.getParentNodeTypes()){
+                addParentNodeType(parentNodeType);
+            }
+        }
+
+        if (cs.getPropertyNames() == null){
+            propertyNameOverflow = true;
+        } else {
+            for (String propertyName : cs.getPropertyNames()){
+                addPropertyName(propertyName);
+            }
+        }
+
+        if (cs.getAllNodeTypes() == null){
+            allNodeTypeOverflow = true;
+        } else {
+            for (String nodeType : cs.getAllNodeTypes()){
+                addNodeType(nodeType);
+            }
+        }
+
+        return this;
+    }
+
     public ChangeSet build() {
         return new ChangeSet(maxPathDepth, parentPathOverflow ? null : parentPaths,
                 parentNodeNameOverflow ? null : parentNodeNames,
@@ -123,6 +169,34 @@ public ChangeSet build() {
                 allNodeTypeOverflow ? null : allNodeTypes);
     }
 
+    private void addPathFromChangeSet(ChangeSet cs) {
+        int maxDepthInChangeSet = cs.getMaxPrefilterPathDepth();
+
+        //If maxDepth of ChangeSet being added is less than current
+        //then truncate path in current set to that depth and change
+        //maxPathDepth to one from ChangeSet
+        if (maxDepthInChangeSet < maxPathDepth){
+            Set<String> existingPathSet = Sets.newHashSet(parentPaths);
+            parentPaths.clear();
+            for (String existingPath : existingPathSet){
+               parentPaths.add(getPathWithMaxDepth(existingPath, maxDepthInChangeSet));
+            }
+            maxPathDepth = maxDepthInChangeSet;
+        }
+
+        for (String pathFromChangeSet : cs.getParentPaths()){
+            addParentPath(getPathWithMaxDepth(pathFromChangeSet, maxPathDepth));
+        }
+    }
+
+    private static String getPathWithMaxDepth(String path, int maxDepth){
+        int depth = PathUtils.getDepth(path);
+        if (depth <= maxDepth){
+            return path;
+        }
+        return PathUtils.getAncestorPath(path, depth - maxDepth);
+    }
+
     /**
      * Add data to dataSet if dataSet size is less than maxSize.
      *
diff --git a/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilderTest.java b/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilderTest.java
new file mode 100644
index 0000000..baf076d
--- /dev/null
+++ b/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetBuilderTest.java
@@ -0,0 +1,206 @@
+/*
+ * 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.jackrabbit.oak.plugins.observation;
+
+import org.junit.Test;
+
+import static com.google.common.collect.ImmutableSet.of;
+import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+
+public class ChangeSetBuilderTest {
+
+    @Test
+    public void basicMerge() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(5, 2);
+        add(cb1, "1");
+
+        ChangeSetBuilder cb2 = new ChangeSetBuilder(5, 2);
+        add(cb2, "2");
+
+        ChangeSet cs = cb1.add(cb2.build()).build();
+        assertThat(cs.getAllNodeTypes(), containsInAnyOrder("nt-1", "nt-2"));
+        assertThat(cs.getParentPaths(), containsInAnyOrder("p-1", "p-2"));
+        assertThat(cs.getParentNodeNames(), containsInAnyOrder("nn-1", "nn-2"));
+        assertThat(cs.getParentNodeTypes(), containsInAnyOrder("pnt-1", "pnt-2"));
+        assertThat(cs.getPropertyNames(), containsInAnyOrder("pn-1", "pn-2"));
+    }
+
+    @Test
+    public void addedChangeSetAlreadyOverflown() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(5, 2);
+        add(cb1, "1");
+
+        ChangeSet cs1 = new ChangeSet(2, null, of("nn-2"), of("nt-2"), of("pn-2"), of("nt-2"));
+        ChangeSet mcs1 = cb1.add(cs1).build();
+        assertNull(mcs1.getParentPaths());
+        assertThat(mcs1.getAllNodeTypes(), containsInAnyOrder("nt-1", "nt-2"));
+    }
+
+    @Test
+    public void overflowPath() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(2, 2);
+        add(cb1, "1");
+
+        ChangeSet cs1 = new ChangeSet(2, null, of("nn-2"), of("pnt-2"), of("pn-2"), of("nt-2"));
+        ChangeSet cs = cb1.add(cs1).build();
+        assertNull(cs.getParentPaths());
+        assertThat(cs.getAllNodeTypes(), containsInAnyOrder("nt-1", "nt-2"));
+        assertThat(cs.getParentNodeNames(), containsInAnyOrder("nn-1", "nn-2"));
+        assertThat(cs.getParentNodeTypes(), containsInAnyOrder("pnt-1", "pnt-2"));
+        assertThat(cs.getPropertyNames(), containsInAnyOrder("pn-1", "pn-2"));
+
+        ChangeSet cs2 = new ChangeSet(2, of("p-2", "p-3"), of("nn-2"), of("pnt-2"), of("pn-2"), of("nt-2"));
+        cs = cb1.add(cs2).build();
+        assertNull(cs.getParentPaths());
+    }
+
+    @Test
+    public void overflowParentNodeName() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(2, 2);
+        add(cb1, "1");
+
+        ChangeSet cs1 = new ChangeSet(2, of("p-2"), null, of("pnt-2"), of("pn-2"), of("nt-2"));
+        ChangeSet cs = cb1.add(cs1).build();
+        assertNull(cs.getParentNodeNames());
+        assertThat(cs.getAllNodeTypes(), containsInAnyOrder("nt-1", "nt-2"));
+        assertThat(cs.getParentNodeTypes(), containsInAnyOrder("pnt-1", "pnt-2"));
+        assertThat(cs.getPropertyNames(), containsInAnyOrder("pn-1", "pn-2"));
+
+        ChangeSet cs2 = new ChangeSet(2, of("p-2"), of("nn-2", "nn-3"), of("pnt-2"), of("pn-2"), of("nt-2"));
+        cs = cb1.add(cs2).build();
+        assertNull(cs.getParentNodeNames());
+    }
+
+    @Test
+    public void overflowParentNodeTypes() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(2, 2);
+        add(cb1, "1");
+
+        ChangeSet cs1 = new ChangeSet(2, of("p-2"), of("nn-2"), null, of("pn-2"), of("nt-2"));
+        ChangeSet cs = cb1.add(cs1).build();
+        assertNull(cs.getParentNodeTypes());
+        assertThat(cs.getParentPaths(), containsInAnyOrder("p-1", "p-2"));
+        assertThat(cs.getAllNodeTypes(), containsInAnyOrder("nt-1", "nt-2"));
+        assertThat(cs.getParentNodeNames(), containsInAnyOrder("nn-1", "nn-2"));
+        assertThat(cs.getPropertyNames(), containsInAnyOrder("pn-1", "pn-2"));
+    }
+
+    @Test
+    public void overflowPropertyNames() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(2, 2);
+        add(cb1, "1");
+
+        ChangeSet cs1 = new ChangeSet(2, of("p-2"), of("nn-2"), of("pnt-2"), null, of("nt-2"));
+        ChangeSet cs = cb1.add(cs1).build();
+        assertNull(cs.getPropertyNames());
+        assertThat(cs.getParentPaths(), containsInAnyOrder("p-1", "p-2"));
+        assertThat(cs.getAllNodeTypes(), containsInAnyOrder("nt-1", "nt-2"));
+        assertThat(cs.getParentNodeNames(), containsInAnyOrder("nn-1", "nn-2"));
+        assertThat(cs.getParentNodeTypes(), containsInAnyOrder("pnt-1", "pnt-2"));
+    }
+
+    @Test
+    public void overflowAllNodeTypes() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(2, 2);
+        add(cb1, "1");
+
+        ChangeSet cs1 = new ChangeSet(2, of("p-2"), of("nn-2"), of("pnt-2"), of("pn-2"), null);
+        ChangeSet cs = cb1.add(cs1).build();
+        assertNull(cs.getAllNodeTypes());
+        assertThat(cs.getParentPaths(), containsInAnyOrder("p-1", "p-2"));
+        assertThat(cs.getParentNodeNames(), containsInAnyOrder("nn-1", "nn-2"));
+        assertThat(cs.getParentNodeTypes(), containsInAnyOrder("pnt-1", "pnt-2"));
+        assertThat(cs.getPropertyNames(), containsInAnyOrder("pn-1", "pn-2"));
+    }
+
+    @Test
+    public void pathDepth() throws Exception{
+        ChangeSetBuilder cb = new ChangeSetBuilder(10, 2);
+        cb.addParentPath("/a/b");
+        cb.addParentPath("/x");
+        cb.addParentPath("/p/q/r");
+
+        ChangeSet cs = cb.build();
+        assertThat(cs.getParentPaths(), containsInAnyOrder("/a/b", "/x"));
+    }
+
+    @Test
+    public void changeSetDepthMoreThanBuilder() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(10, 3);
+        cb1.addParentPath("/x");
+        cb1.addParentPath("/x/y");
+        cb1.addParentPath("/x/y/z");
+
+        ChangeSetBuilder cb2 = new ChangeSetBuilder(10, 8);
+        cb2.addParentPath("/p");
+        cb2.addParentPath("/p/q");
+        cb2.addParentPath("/p/q/r");
+        cb2.addParentPath("/a/b/c/d");
+        cb2.addParentPath("/a/b/x/y/z");
+
+        cb1.add(cb2.build());
+
+        ChangeSet cs = cb1.build();
+        assertThat(cs.getParentPaths(), containsInAnyOrder(
+                "/x", "/x/y", "/x/y/z",
+                "/p", "/p/q", "/p/q/r",
+                "/a/b/c", "/a/b/x" //Chopped paths
+        ));
+
+        assertEquals(cb1.getMaxPrefilterPathDepth(), cs.getMaxPrefilterPathDepth());
+    }
+
+    @Test
+    public void builderDepthMoreThanChangeSet() throws Exception{
+        ChangeSetBuilder cb1 = new ChangeSetBuilder(10, 8);
+        cb1.addParentPath("/p");
+        cb1.addParentPath("/p/q");
+        cb1.addParentPath("/p/q/r");
+        cb1.addParentPath("/a/b/c/d");
+        cb1.addParentPath("/a/b/x/y/z");
+
+        ChangeSetBuilder cb2 = new ChangeSetBuilder(10, 2);
+        cb2.addParentPath("/x");
+        cb2.addParentPath("/x/y");
+
+        cb1.add(cb2.build());
+
+        ChangeSet cs = cb1.build();
+        assertThat(cs.getParentPaths(), containsInAnyOrder(
+                "/x", "/x/y",
+                "/p", "/p/q",
+                "/a/b" //Chopped paths
+        ));
+
+        assertEquals(cb2.getMaxPrefilterPathDepth(), cs.getMaxPrefilterPathDepth());
+    }
+
+    private static void add(ChangeSetBuilder cb, String suffix){
+        cb.addNodeType("nt-"+suffix)
+                .addParentPath("p-"+suffix)
+                .addParentNodeName("nn-"+suffix)
+                .addParentNodeType("pnt-"+suffix)
+                .addPropertyName("pn-"+suffix);
+    }
+
+}
\ No newline at end of file
diff --git a/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetTest.java b/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetTest.java
new file mode 100644
index 0000000..157d4e4
--- /dev/null
+++ b/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/observation/ChangeSetTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.jackrabbit.oak.plugins.observation;
+
+import com.google.common.collect.ImmutableSet;
+import org.junit.Test;
+
+import static com.google.common.collect.ImmutableSet.of;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class ChangeSetTest {
+
+    @Test
+    public void asJson() throws Exception{
+        ChangeSet cs1 = new ChangeSet(2, of("p-2", "p-3"), null,
+                ImmutableSet.<String>of(), of("pn-2"), of("nt-2"));
+        String json = cs1.asString();
+
+        ChangeSet cs2 = ChangeSet.fromString(json);
+        assertEquals(cs1, cs2);
+        assertNull(cs2.getParentNodeNames());
+        assertTrue(cs2.getParentNodeTypes().isEmpty());
+    }
+
+    @Test
+    public void asJsonAll() throws Exception{
+        ChangeSet cs1 = new ChangeSet(2, of("p-2"), of("nn-2"), of("pnt-2"), of("pn-2"), of("nt-2"));
+        String json = cs1.asString();
+        ChangeSet cs2 = ChangeSet.fromString(json);
+        assertEquals(cs1, cs2);
+    }
+
+}
\ No newline at end of file