From fd032913e9b877ea7152f8dd05cbf0e82be4e7ea Mon Sep 17 00:00:00 2001
From: Sunjeet Singh <sstattla@yahoo.com>
Date: Thu, 23 Jan 2025 18:08:01 -0800
Subject: [PATCH] Approach 1- track rev stats at the same time as fwd stats

---
 .../hollow/core/write/HollowBlobWriter.java   |   5 +-
 .../core/write/HollowListTypeWriteState.java  |  25 +--
 .../core/write/HollowMapTypeWriteState.java   |  42 +++--
 .../write/HollowObjectTypeWriteState.java     | 143 +++++++++---------
 .../core/write/HollowSetTypeWriteState.java   |  12 +-
 .../core/write/HollowTypeWriteState.java      |  12 +-
 .../core/write/HollowWriteStateEngine.java    |  30 +++-
 7 files changed, 159 insertions(+), 110 deletions(-)

diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowBlobWriter.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowBlobWriter.java
index fe520b287..34196e099 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowBlobWriter.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowBlobWriter.java
@@ -200,8 +200,9 @@ public void writeReverseDelta(OutputStream os, ProducerOptionalBlobPartConfig.Op
         if(partStreams != null)
             partStreamsByType = partStreams.getStreamsByType();
 
-        stateEngine.prepareForWrite(true);
-        
+        stateEngine.prepareForWrite();
+        // stateEngine.prepareForWrite(true);   // SNAP: TODO: remove
+
         if(stateEngine.isRestored())
             stateEngine.ensureAllNecessaryStatesRestored();
         
diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowListTypeWriteState.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowListTypeWriteState.java
index 3271a4d4e..55ab2f297 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowListTypeWriteState.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowListTypeWriteState.java
@@ -73,7 +73,7 @@ public void prepareForWrite() {
         maxOrdinal = ordinalMap.maxOrdinal();
         gatherShardingStats(maxOrdinal);
 
-        gatherStatistics(numShards);
+        gatherStatistics(numShards != revNumShards);
     }
 
     private void gatherStatistics(boolean numShardsChanged) {
@@ -351,12 +351,11 @@ public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet
 
     @Override
     public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet toCyclePopulated, boolean isReverse) {
-        boolean numShardsChanged = this.revNumShards != this.numShards;
         int numShards = this.numShards;
-        long[] totalOfListSizes = this.totalOfListSizes;
-        if (numShardsChanged && isReverse) {
+        int bitsPerListPointer = this.bitsPerListPointer;
+        if (isReverse && this.numShards != this.revNumShards) {
             numShards = this.revNumShards;
-            totalOfListSizes = this.revTotalOfListSizes;
+            bitsPerListPointer = this.revBitsPerListPointer;
         }
 
         maxOrdinal = ordinalMap.maxOrdinal();
@@ -431,13 +430,13 @@ public void writeCalculatedDelta(DataOutputStream os, int[] maxShardOrdinal) thr
 
         /// for unsharded blobs, support pre v2.1.0 clients
         if(numShards == 1) {
-            writeCalculatedDeltaShard(os, 0, maxShardOrdinal);
+            writeCalculatedDeltaShard(os, 0, maxShardOrdinal, bitsPerListPointer, totalOfListSizes);
         } else {
             /// overall max ordinal
             VarInt.writeVInt(os, maxOrdinal);
 
             for(int i=0;i<numShards;i++) {
-                writeCalculatedDeltaShard(os, i, maxShardOrdinal);
+                writeCalculatedDeltaShard(os, i, maxShardOrdinal, bitsPerListPointer, totalOfListSizes);
             }
         }
 
@@ -449,23 +448,25 @@ public void writeCalculatedDelta(DataOutputStream os, int[] maxShardOrdinal) thr
 
     @Override
     public void writeCalculatedDelta(DataOutputStream os, boolean isReverse, int[] maxShardOrdinal) throws IOException {
-        boolean numShardsChanged = this.revNumShards != this.numShards;
         int numShards = this.numShards;
+        int bitsPerListPointer = this.bitsPerListPointer;
         long[] totalOfListSizes = this.totalOfListSizes;
-        if (numShardsChanged && isReverse) {
+        if (isReverse && this.numShards != this.revNumShards) {
             numShards = this.revNumShards;
+            bitsPerListPointer = this.revBitsPerListPointer;
             totalOfListSizes = this.revTotalOfListSizes;
         }
 
+
         /// for unsharded blobs, support pre v2.1.0 clients
         if(numShards == 1) {
-            writeCalculatedDeltaShard(os, 0, maxShardOrdinal);
+            writeCalculatedDeltaShard(os, 0, maxShardOrdinal, bitsPerListPointer, totalOfListSizes);
         } else {
             /// overall max ordinal
             VarInt.writeVInt(os, maxOrdinal);
             
             for(int i=0;i<numShards;i++) {
-                writeCalculatedDeltaShard(os, i, maxShardOrdinal);
+                writeCalculatedDeltaShard(os, i, maxShardOrdinal, bitsPerListPointer, totalOfListSizes);
             }
         }
 
@@ -476,7 +477,7 @@ public void writeCalculatedDelta(DataOutputStream os, boolean isReverse, int[] m
     }
 
 
-    private void writeCalculatedDeltaShard(DataOutputStream os, int shardNumber, int[] maxShardOrdinal) throws IOException {
+    private void writeCalculatedDeltaShard(DataOutputStream os, int shardNumber, int[] maxShardOrdinal, int bitsPerListPointer, long[] totalOfListSizes) throws IOException {
         /// 1) max shard ordinal
         VarInt.writeVInt(os, maxShardOrdinal[shardNumber]);
 
diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowMapTypeWriteState.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowMapTypeWriteState.java
index 27a1e3981..31c26efd4 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowMapTypeWriteState.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowMapTypeWriteState.java
@@ -37,10 +37,12 @@ public class HollowMapTypeWriteState extends HollowTypeWriteState {
 
     /// statistics required for writing fixed length set data
     private int bitsPerMapPointer;
+    private int revBitsPerMapPointer;
     private int bitsPerMapSizeValue;
     private int bitsPerKeyElement;
     private int bitsPerValueElement;
     private long totalOfMapBuckets[];
+    private long revTotalOfMapBuckets[];
 
     /// data required for writing snapshot or delta
     private FixedLengthElementArray mapPointersAndSizesArray[];
@@ -79,7 +81,7 @@ public void prepareForWrite() {
         maxOrdinal = ordinalMap.maxOrdinal();
         gatherShardingStats(maxOrdinal);
 
-        gatherStatistics(numShards);
+        gatherStatistics(numShards != revNumShards);
     }
 
     private void gatherStatistics(boolean numShardsChanged) {
@@ -91,7 +93,9 @@ private void gatherStatistics(boolean numShardsChanged) {
         ByteData data = ordinalMap.getByteData().getUnderlyingArray();
 
         totalOfMapBuckets = new long[numShards];
-        // SNAP: TOOD: missing revNumShards for approach 1
+        if (numShardsChanged) {
+            revTotalOfMapBuckets = new long[revNumShards];
+        }
         
         for(int i=0;i<=maxOrdinal;i++) {
             if(currentCyclePopulated.get(i) || previousCyclePopulated.get(i)) {
@@ -123,6 +127,9 @@ private void gatherStatistics(boolean numShardsChanged) {
                 }
 
                 totalOfMapBuckets[i & (numShards-1)] += numBuckets;
+                if (numShardsChanged) {
+                    revTotalOfMapBuckets[i & (revNumShards-1)] += numBuckets;
+                }
             }
         }
         
@@ -138,6 +145,15 @@ private void gatherStatistics(boolean numShardsChanged) {
         bitsPerMapSizeValue = 64 - Long.numberOfLeadingZeros(maxMapSize);
 
         bitsPerMapPointer = 64 - Long.numberOfLeadingZeros(maxShardTotalOfMapBuckets);
+
+        if (numShardsChanged) {
+            long revMaxShardTotalOfMapBuckets = 0;
+            for(int i=0;i<revNumShards;i++) {
+                if(revTotalOfMapBuckets[i] > revMaxShardTotalOfMapBuckets)
+                    revMaxShardTotalOfMapBuckets = revTotalOfMapBuckets[i];
+            }
+            revBitsPerMapPointer = 64 - Long.numberOfLeadingZeros(revMaxShardTotalOfMapBuckets);
+        }
     }
 
     @Override
@@ -509,10 +525,11 @@ public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet
 
     @Override
     public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet toCyclePopulated, boolean isReverse) {
-        boolean numShardsChanged = this.revNumShards != this.numShards;
         int numShards = this.numShards;
-        if (numShardsChanged && isReverse) {
+        int bitsPerMapPointer = this.bitsPerMapPointer;
+        if (isReverse && this.numShards != this.revNumShards) {
             numShards = this.revNumShards;
+            bitsPerMapPointer = this.revBitsPerMapPointer;
         }
 
         maxOrdinal = ordinalMap.maxOrdinal();
@@ -634,13 +651,13 @@ public void writeCalculatedDelta(DataOutputStream os, int[] maxShardOrdinal) thr
 
         /// for unsharded blobs, support pre v2.1.0 clients
         if(numShards == 1) {
-            writeCalculatedDeltaShard(os, 0, maxShardOrdinal);
+            writeCalculatedDeltaShard(os, 0, maxShardOrdinal, bitsPerMapPointer, totalOfMapBuckets);
         } else {
             /// overall max ordinal
             VarInt.writeVInt(os, maxOrdinal);
 
             for(int i=0;i<numShards;i++) {
-                writeCalculatedDeltaShard(os, i, maxShardOrdinal);
+                writeCalculatedDeltaShard(os, i, maxShardOrdinal, bitsPerMapPointer, totalOfMapBuckets);
             }
         }
 
@@ -652,21 +669,24 @@ public void writeCalculatedDelta(DataOutputStream os, int[] maxShardOrdinal) thr
 
     @Override
     public void writeCalculatedDelta(DataOutputStream os, boolean isReverse, int[] maxShardOrdinal) throws IOException {
-        boolean numShardsChanged = this.revNumShards != this.numShards;
         int numShards = this.numShards;
-        if (numShardsChanged && isReverse) {
+        int bitsPerMapPointer = this.bitsPerMapPointer;
+        long[] totalOfMapBuckets = this.totalOfMapBuckets;
+        if (isReverse && this.numShards != this.revNumShards) {
             numShards = this.revNumShards;
+            bitsPerMapPointer = this.revBitsPerMapPointer;
+            totalOfMapBuckets = this.revTotalOfMapBuckets;
         }
 
         /// for unsharded blobs, support pre v2.1.0 clients
         if(numShards == 1) {
-            writeCalculatedDeltaShard(os, 0, maxShardOrdinal);
+            writeCalculatedDeltaShard(os, 0, maxShardOrdinal, bitsPerMapPointer, totalOfMapBuckets);
         } else {
             /// overall max ordinal
             VarInt.writeVInt(os, maxOrdinal);
             
             for(int i=0;i<numShards;i++) {
-                writeCalculatedDeltaShard(os, i, maxShardOrdinal);
+                writeCalculatedDeltaShard(os, i, maxShardOrdinal, bitsPerMapPointer, totalOfMapBuckets);
             }
         }
         
@@ -676,7 +696,7 @@ public void writeCalculatedDelta(DataOutputStream os, boolean isReverse, int[] m
         deltaRemovedOrdinals = null;
     }
     
-    private void writeCalculatedDeltaShard(DataOutputStream os, int shardNumber, int[] maxShardOrdinal) throws IOException {
+    private void writeCalculatedDeltaShard(DataOutputStream os, int shardNumber, int[] maxShardOrdinal, int bitsPerMapPointer, long[] totalOfMapBuckets) throws IOException {
         
         int bitsPerMapFixedLengthPortion = bitsPerMapSizeValue + bitsPerMapPointer;
         int bitsPerMapEntry = bitsPerKeyElement + bitsPerValueElement;
diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowObjectTypeWriteState.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowObjectTypeWriteState.java
index f5c6e989e..ffab631ad 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowObjectTypeWriteState.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowObjectTypeWriteState.java
@@ -180,7 +180,7 @@ public void calculateSnapshot() {
         maxOrdinal = ordinalMap.maxOrdinal();
 
         int numBitsPerRecord = fieldStats.getNumBitsPerRecord();
-        
+
         fixedLengthLongArray = new FixedLengthElementArray[numShards];
         varLengthByteArrays = new ByteDataArray[numShards][];
         recordBitOffset = new long[numShards];
@@ -299,52 +299,55 @@ public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet
 
     @Override
     public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet toCyclePopulated, boolean isReverse) {
-        throw new UnsupportedOperationException();
-
-//        maxOrdinal = ordinalMap.maxOrdinal();
-//        int numBitsPerRecord = fieldStats.getNumBitsPerRecord();
-//
-//        ThreadSafeBitSet deltaAdditions = toCyclePopulated.andNot(fromCyclePopulated);
-//
-//        fixedLengthLongArray = new FixedLengthElementArray[numShards];
-//        deltaAddedOrdinals = new ByteDataArray[numShards];
-//        deltaRemovedOrdinals = new ByteDataArray[numShards];
-//        varLengthByteArrays = new ByteDataArray[numShards][];
-//        recordBitOffset = new long[numShards];
-//        int numAddedRecordsInShard[] = new int[numShards];
-//
-//        int shardMask = numShards - 1;
-//
-//        int addedOrdinal = deltaAdditions.nextSetBit(0);
-//        while(addedOrdinal != -1) {
-//            numAddedRecordsInShard[addedOrdinal & shardMask]++;
-//            addedOrdinal = deltaAdditions.nextSetBit(addedOrdinal + 1);
-//        }
-//
-//        for(int i=0;i<numShards;i++) {
-//            fixedLengthLongArray[i] = new FixedLengthElementArray(WastefulRecycler.DEFAULT_INSTANCE, (long)numAddedRecordsInShard[i] * numBitsPerRecord);
-//            deltaAddedOrdinals[i] = new ByteDataArray(WastefulRecycler.DEFAULT_INSTANCE);
-//            deltaRemovedOrdinals[i] = new ByteDataArray(WastefulRecycler.DEFAULT_INSTANCE);
-//            varLengthByteArrays[i] = new ByteDataArray[getSchema().numFields()];
-//        }
-//
-//        int previousRemovedOrdinal[] = new int[numShards];
-//        int previousAddedOrdinal[] = new int[numShards];
-//
-//        for(int i=0;i<=maxOrdinal;i++) {
-//            int shardNumber = i & shardMask;
-//            if(deltaAdditions.get(i)) {
-//                addRecord(i, recordBitOffset[shardNumber], fixedLengthLongArray[shardNumber], varLengthByteArrays[shardNumber]);
-//                recordBitOffset[shardNumber] += numBitsPerRecord;
-//                int shardOrdinal = i / numShards;
-//                VarInt.writeVInt(deltaAddedOrdinals[shardNumber], shardOrdinal - previousAddedOrdinal[shardNumber]);
-//                previousAddedOrdinal[shardNumber] = shardOrdinal;
-//            } else if(fromCyclePopulated.get(i) && !toCyclePopulated.get(i)) {
-//                int shardOrdinal = i / numShards;
-//                VarInt.writeVInt(deltaRemovedOrdinals[shardNumber], shardOrdinal - previousRemovedOrdinal[shardNumber]);
-//                previousRemovedOrdinal[shardNumber] = shardOrdinal;
-//            }
-//        }
+        int numShards = this.numShards;
+        if (isReverse && this.numShards != this.revNumShards) {
+            numShards = this.revNumShards;
+        }
+
+        maxOrdinal = ordinalMap.maxOrdinal();
+        int numBitsPerRecord = fieldStats.getNumBitsPerRecord();
+
+        ThreadSafeBitSet deltaAdditions = toCyclePopulated.andNot(fromCyclePopulated);
+
+        fixedLengthLongArray = new FixedLengthElementArray[numShards];
+        deltaAddedOrdinals = new ByteDataArray[numShards];
+        deltaRemovedOrdinals = new ByteDataArray[numShards];
+        varLengthByteArrays = new ByteDataArray[numShards][];
+        recordBitOffset = new long[numShards];
+        int numAddedRecordsInShard[] = new int[numShards];
+
+        int shardMask = numShards - 1;
+
+        int addedOrdinal = deltaAdditions.nextSetBit(0);
+        while(addedOrdinal != -1) {
+            numAddedRecordsInShard[addedOrdinal & shardMask]++;
+            addedOrdinal = deltaAdditions.nextSetBit(addedOrdinal + 1);
+        }
+
+        for(int i=0;i<numShards;i++) {
+            fixedLengthLongArray[i] = new FixedLengthElementArray(WastefulRecycler.DEFAULT_INSTANCE, (long)numAddedRecordsInShard[i] * numBitsPerRecord);
+            deltaAddedOrdinals[i] = new ByteDataArray(WastefulRecycler.DEFAULT_INSTANCE);
+            deltaRemovedOrdinals[i] = new ByteDataArray(WastefulRecycler.DEFAULT_INSTANCE);
+            varLengthByteArrays[i] = new ByteDataArray[getSchema().numFields()];
+        }
+
+        int previousRemovedOrdinal[] = new int[numShards];
+        int previousAddedOrdinal[] = new int[numShards];
+
+        for(int i=0;i<=maxOrdinal;i++) {
+            int shardNumber = i & shardMask;
+            if(deltaAdditions.get(i)) {
+                addRecord(i, recordBitOffset[shardNumber], fixedLengthLongArray[shardNumber], varLengthByteArrays[shardNumber]);
+                recordBitOffset[shardNumber] += numBitsPerRecord;
+                int shardOrdinal = i / numShards;
+                VarInt.writeVInt(deltaAddedOrdinals[shardNumber], shardOrdinal - previousAddedOrdinal[shardNumber]);
+                previousAddedOrdinal[shardNumber] = shardOrdinal;
+            } else if(fromCyclePopulated.get(i) && !toCyclePopulated.get(i)) {
+                int shardOrdinal = i / numShards;
+                VarInt.writeVInt(deltaRemovedOrdinals[shardNumber], shardOrdinal - previousRemovedOrdinal[shardNumber]);
+                previousRemovedOrdinal[shardNumber] = shardOrdinal;
+            }
+        }
     }
 
     @Override
@@ -372,30 +375,28 @@ public void writeCalculatedDelta(DataOutputStream os, int[] maxShardOrdinal) thr
 
     @Override
     public void writeCalculatedDelta(DataOutputStream os, boolean isReverse, int[] maxShardOrdinal) throws IOException {
-        throw new IllegalStateException();
-//        boolean numShardsChanged = this.revNumShards != this.numShards;
-//        int numShards = this.numShards;
-//        if (numShardsChanged && isReverse) {
-//            numShards = this.revNumShards;
-//        }
-//
-//        /// for unsharded blobs, support pre v2.1.0 clients
-//        if(numShards == 1) {
-//            writeCalculatedDeltaShard(os, 0, maxShardOrdinal);
-//        } else {
-//            /// overall max ordinal
-//            VarInt.writeVInt(os, maxOrdinal);
-//
-//            for(int i=0;i<numShards;i++) {
-//                writeCalculatedDeltaShard(os, i, maxShardOrdinal);
-//            }
-//        }
-//
-//        fixedLengthLongArray = null;
-//        varLengthByteArrays = null;
-//        deltaAddedOrdinals = null;
-//        deltaRemovedOrdinals = null;
-//        recordBitOffset = null;
+        int numShards = this.numShards;
+        if (isReverse && this.numShards != this.revNumShards) {
+            numShards = this.revNumShards;
+        }
+
+        /// for unsharded blobs, support pre v2.1.0 clients
+        if(numShards == 1) {
+            writeCalculatedDeltaShard(os, 0, maxShardOrdinal);
+        } else {
+            /// overall max ordinal
+            VarInt.writeVInt(os, maxOrdinal);
+
+            for(int i=0;i<numShards;i++) {
+                writeCalculatedDeltaShard(os, i, maxShardOrdinal);
+            }
+        }
+
+        fixedLengthLongArray = null;
+        varLengthByteArrays = null;
+        deltaAddedOrdinals = null;
+        deltaRemovedOrdinals = null;
+        recordBitOffset = null;
     }
 
     private void writeCalculatedDeltaShard(DataOutputStream os, int shardNumber, int[] maxShardOrdinal) throws IOException {
diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowSetTypeWriteState.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowSetTypeWriteState.java
index 050df12db..59a893032 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowSetTypeWriteState.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowSetTypeWriteState.java
@@ -78,7 +78,7 @@ public void prepareForWrite() {
         maxOrdinal = ordinalMap.maxOrdinal();
 
         gatherShardingStats(maxOrdinal);
-        gatherStatistics(numShards);
+        gatherStatistics(numShards != revNumShards);
     }
 
     // invoked once per runCycle, will gather stats for all- snapshot, delta, reverse delta
@@ -475,17 +475,17 @@ private void writeSnapshotShard(DataOutputStream os, int shardNumber) throws IOE
 
     @Override
     public void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet toCyclePopulated, boolean isReverse) {
-        boolean numShardsChanged = this.revNumShards != this.numShards;
         int numShards = this.numShards;
         int bitsPerSetPointer = this.bitsPerSetPointer;
-        if (numShardsChanged && isReverse) {
+        if (isReverse && numShards != revNumShards) {
             numShards = this.revNumShards;
             bitsPerSetPointer = this.revBitsPerSetPointer;
         }
+
         maxOrdinal = ordinalMap.maxOrdinal();
 
         int bitsPerSetFixedLengthPortion = bitsPerSetSizeValue + bitsPerSetPointer;
-        
+
         numSetsInDelta = new int[numShards];
         numBucketsInDelta = new long[numShards];
         setPointersAndSizesArray = new FixedLengthElementArray[numShards];
@@ -617,9 +617,7 @@ public void writeCalculatedDelta(DataOutputStream os, boolean isReverse, int[] m
         int numShards = this.numShards;
         int bitsPerSetPointer = this.bitsPerSetPointer;
         long[] totalOfSetBuckets = this.totalOfSetBuckets;
-
-        boolean numShardsChanged = this.revNumShards != this.numShards;
-        if (numShardsChanged && isReverse) {
+        if (isReverse && numShards != revNumShards) {
             numShards = this.revNumShards;
             bitsPerSetPointer = this.revBitsPerSetPointer;
             totalOfSetBuckets = this.revTotalOfSetBuckets;
diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowTypeWriteState.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowTypeWriteState.java
index ae0833a01..dabf04d60 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowTypeWriteState.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowTypeWriteState.java
@@ -328,21 +328,25 @@ public boolean isRestored() {
     public abstract void writeSnapshot(DataOutputStream dos) throws IOException;
 
     public void calculateDelta() {
-        calculateDelta(previousCyclePopulated, currentCyclePopulated, numShards);
+        calculateDelta(previousCyclePopulated, currentCyclePopulated, false);
+        // calculateDelta(previousCyclePopulated, currentCyclePopulated, numShards);
     }
 
     public void calculateReverseDelta() {
-        calculateDelta(currentCyclePopulated, previousCyclePopulated, revNumShards);
+        calculateDelta(currentCyclePopulated, previousCyclePopulated, true);
+        // calculateDelta(currentCyclePopulated, previousCyclePopulated, revNumShards);
     }
 
     public void writeDelta(DataOutputStream dos) throws IOException {
         LOG.log(Level.FINE, String.format("Writing delta with num shards = %s, max shard ordinals = %s", numShards, Arrays.toString(maxShardOrdinal)));
-        writeCalculatedDelta(dos, maxShardOrdinal);
+        writeCalculatedDelta(dos, false, maxShardOrdinal);
+        // writeCalculatedDelta(dos, maxShardOrdinal);
     }
 
     public void writeReverseDelta(DataOutputStream dos) throws IOException {
         LOG.log(Level.FINE, String.format("Writing reversedelta with num shards = %s, max shard ordinals = %s", revNumShards, Arrays.toString(revMaxShardOrdinal)));
-        writeCalculatedDelta(dos, revMaxShardOrdinal);
+        writeCalculatedDelta(dos, true, revMaxShardOrdinal);
+        // writeCalculatedDelta(dos, revMaxShardOrdinal);
     }
 
     public abstract void calculateDelta(ThreadSafeBitSet fromCyclePopulated, ThreadSafeBitSet toCyclePopulated, int numShards);
diff --git a/hollow/src/main/java/com/netflix/hollow/core/write/HollowWriteStateEngine.java b/hollow/src/main/java/com/netflix/hollow/core/write/HollowWriteStateEngine.java
index 3fba6d952..70f5ec3ca 100644
--- a/hollow/src/main/java/com/netflix/hollow/core/write/HollowWriteStateEngine.java
+++ b/hollow/src/main/java/com/netflix/hollow/core/write/HollowWriteStateEngine.java
@@ -189,11 +189,35 @@ public void run() {
     /**
      * Transition from the "adding records" phase of a cycle to the "writing" phase of a cycle.
      */
-    public void prepareForWrite() { // SNAP: TODO: implement the selection of true/false
-        prepareForWrite(false);
+    public void prepareForWrite() {
+
+        if(!preparedForNextCycle) {
+            return;
+        }
+
+        addTypeNamesWithDefinedHashCodesToHeader();
+
+        try {
+            SimultaneousExecutor executor = new SimultaneousExecutor(getClass(), "prepare-for-write");
+
+            for(final Map.Entry<String, HollowTypeWriteState> typeStateEntry : writeStates.entrySet()) {
+                executor.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        typeStateEntry.getValue().prepareForWrite();
+                    }
+                });
+            }
+
+            executor.awaitSuccessfulCompletion();
+        } catch(Exception ex) {
+            throw new HollowWriteStateException("Failed to prepare for write", ex);
+        }
+
+        preparedForNextCycle = false;
     }
 
-    public void prepareForWrite(boolean recomputeStats) {
+    public void prepareForWrite(boolean recomputeStats) {   // SNAP: TODO: remove
 
         if(!preparedForNextCycle) {
             if (recomputeStats) {   // already prepared for write but stats needs to be recomputed (when reverse delta has different num shards)