Index: oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentWriter.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentWriter.java	(date 1468909672000)
+++ oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentWriter.java	(date 1468933663000)
@@ -987,27 +987,12 @@
             }
             assert compactionStats != null;
             compactionStats.nodeCount++;
-            if (state instanceof SegmentNodeState) {
-                SegmentNodeState sns = ((SegmentNodeState) state);
-                if (sameStore(sns)) {
-                    // This is a segment node state from an old generation. Check whether
-                    // an equivalent one of the current generation is in the cache
-                    if (isOldGeneration(sns.getRecordId())) {
-                        RecordId cachedId = nodeCache.get(sns.getStableId());
-                        if (cachedId != null) {
-                            compactionStats.cacheHits++;
-                            return cachedId;
-                        } else {
-                            compactionStats.cacheMiss++;
+
+            RecordId compactedId = deduplicateNode(state);
+
+            if (compactedId != null) {
+                return compactedId;
-                        }
+            }
-                    } else {
-                        // This segment node state is already in this store,
-                        // no need to write it again,
-                        compactionStats.deDupNodes++;
-                        return sns.getRecordId();
-                    }
-                }
-            }
 
             compactionStats.writesOps++;
             RecordId recordId = writeNodeUncached(state, depth);
@@ -1022,22 +1007,25 @@
         }
 
         private RecordId writeNodeUncached(@Nonnull NodeState state, int depth) throws IOException {
-            SegmentNodeState before = null;
-            Template beforeTemplate = null;
             ModifiedNodeState after = null;
+
             if (state instanceof ModifiedNodeState) {
                 after = (ModifiedNodeState) state;
-                NodeState base = after.getBaseState();
-                if (base instanceof SegmentNodeState) {
-                    SegmentNodeState sns = ((SegmentNodeState) base);
-                    if (sameStore(sns)) {
-                        if (!isOldGeneration(sns.getRecordId())) {
-                            before = sns;
-                            beforeTemplate = before.getTemplate();
-                        }
+            }
+
+            RecordId beforeId = null;
+
+            if (after != null) {
+                beforeId = deduplicateNode(after.getBaseState());
-                    }
+            }
+
+            SegmentNodeState before = null;
+            Template beforeTemplate = null;
+
+            if (beforeId != null) {
+                before = reader.readNode(beforeId);
+                beforeTemplate = before.getTemplate();
-                }
+            }
-            }
 
             List<RecordId> ids = newArrayList();
             Template template = new Template(reader, state);
@@ -1076,6 +1064,17 @@
                 PropertyState property = state.getProperty(name);
                 assert property != null;
 
+                if (before != null) {
+                    // If this property is already present in before (the base state)
+                    // and it hasn't been modified use that one. This will result
+                    // in an already compacted property to be reused given before
+                    // has been already compacted.
+                    PropertyState beforeProperty = before.getProperty(name);
+                    if (property.equals(beforeProperty)) {
+                        property = beforeProperty;
+                    }
+                }
+
                 if (sameStore(property)) {
                     RecordId pid = ((Record) property).getRecordId();
                     if (isOldGeneration(pid)) {
@@ -1114,6 +1113,46 @@
                 stableId = writeBlock(id, 0, id.length);
             }
             return newNodeStateWriter(stableId, ids).write(writer);
+        }
+
+        /**
+         * Try to deduplicate the passed {@code node}. This succeeds if
+         * the passed node state has already been persisted to this store and
+         * either it has the same generation or it has been already compacted
+         * and is still in the deduplication cache for nodes.
+         *
+         * @param node
+         * @return  the id of the de-duplicated node or {@code null} if none.
+         */
+        private RecordId deduplicateNode(NodeState node) {
+            assert compactionStats != null;
+            if (!(node instanceof SegmentNodeState)) {
+                // Deduplication only for persisted node states
+                return null;
+            }
+
+            SegmentNodeState sns = (SegmentNodeState) node;
+            if (!sameStore(sns)) {
+                // Deduplication only within same store
+                return null;
+            }
+
+            if (!isOldGeneration(sns.getRecordId())) {
+                // This segment node state is already in this store, no need to write it again
+                compactionStats.deDupNodes++;
+                return sns.getRecordId();
+            }
+
+            // This is a segment node state from an old generation. Check whether
+            // an equivalent one of the current generation is in the cache
+            RecordId compacted = nodeCache.get(sns.getStableId());
+            if (compacted == null) {
+                compactionStats.cacheMiss++;
+                return null;
+            }
+
+            compactionStats.cacheHits++;
+            return compacted;
         }
 
         /**
Index: oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java	(date 1468909672000)
+++ oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java	(date 1468933663000)
@@ -20,6 +20,7 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 import javax.annotation.Nonnull;
@@ -28,7 +29,6 @@
 import org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState;
 import org.apache.jackrabbit.oak.segment.file.FileStore;
 import org.apache.jackrabbit.oak.segment.file.FileStoreBuilder;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -90,7 +90,6 @@
     }
 
     @Test
-    @Ignore("OAK-4570")
     public void baseNodeStateShouldBeReusedAcrossGenerations() throws Exception {
         try (FileStore store = newFileStore()) {
             Generation generation = new Generation();
@@ -148,6 +147,15 @@
             // to a new generation.
 
             assertEquals(modified.getTemplateId(), compacted.getTemplateId());
+
+            // Similarly the node state should have reused the property from
+            // the compacted node state, since this property didn't change.
+
+            Record modifiedProperty = (Record) modified.getProperty("a");
+            Record compactedProperty = (Record) compacted.getProperty("a");
+            assertNotNull(modifiedProperty);
+            assertNotNull(compactedProperty);
+            assertEquals(modifiedProperty.getRecordId(), compactedProperty.getRecordId());
         }
     }