map type num shards

Author: Sunjeet

hollow/src/main/java/com/netflix/hollow/core/write/HollowMapTypeWriteState.java

@@ -16,6 +16,8 @@
  */
 package com.netflix.hollow.core.write;
 
+import static com.netflix.hollow.core.index.FieldPaths.FieldPathException.ErrorKind.NOT_BINDABLE;
+
 import com.netflix.hollow.core.index.FieldPaths;
 import com.netflix.hollow.core.memory.ByteData;
 import com.netflix.hollow.core.memory.ByteDataArray;
@@ -30,8 +32,6 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-import static com.netflix.hollow.core.index.FieldPaths.FieldPathException.ErrorKind.NOT_BINDABLE;
-
 public class HollowMapTypeWriteState extends HollowTypeWriteState {
     private static final Logger LOG = Logger.getLogger(HollowMapTypeWriteState.class.getName());
 
@@ -77,20 +77,32 @@ public void prepareForWrite() {
     }
 
     private void gatherStatistics() {
-        if(numShards == -1)
-            calculateNumShards();
-        revNumShards = numShards;
+        int maxOrdinal = ordinalMap.maxOrdinal();
+        if(numShards == -1) {
+            numShards = calculateNumShards(maxOrdinal);
+            revNumShards = numShards;
+        } else {
+            revNumShards = numShards;
+            if (allowTypeResharding()) {
+                numShards = calculateNumShards(maxOrdinal);
+                if (numShards != revNumShards) {    // re-sharding
+                    // limit numShards to 2x or .5x of prevShards per producer cycle
+                    numShards = numShards > revNumShards ? revNumShards * 2 : revNumShards / 2;
+
+                    LOG.info(String.format("Num shards for type %s changing from %s to %s", schema.getName(), revNumShards, numShards));
+                    addReshardingHeader(revNumShards, numShards);   // SNAP: TODO: Here,
+                }
+            }
+        }
+
+        maxShardOrdinal = calcMaxShardOrdinal(maxOrdinal, numShards);
+        if (revNumShards > 0) {
+            revMaxShardOrdinal = calcMaxShardOrdinal(maxOrdinal, revNumShards);
+        }
 
         int maxKeyOrdinal = 0;
         int maxValueOrdinal = 0;
 
-        int maxOrdinal = ordinalMap.maxOrdinal();
-
-        maxShardOrdinal = new int[numShards];
-        int minRecordLocationsPerShard = (maxOrdinal + 1) / numShards; 
-        for(int i=0;i<numShards;i++)
-            maxShardOrdinal[i] = (i < ((maxOrdinal + 1) & (numShards - 1))) ? minRecordLocationsPerShard : minRecordLocationsPerShard - 1;
-        
         int maxMapSize = 0;
         ByteData data = ordinalMap.getByteData().getUnderlyingArray();
 
@@ -143,10 +155,9 @@ private void gatherStatistics() {
         bitsPerMapPointer = 64 - Long.numberOfLeadingZeros(maxShardTotalOfMapBuckets);
     }
 
-    private void calculateNumShards() {
+    int calculateNumShards(int maxOrdinal) {
         int maxKeyOrdinal = 0;
         int maxValueOrdinal = 0;
-        int maxOrdinal = ordinalMap.maxOrdinal();
 
         int maxMapSize = 0;
         ByteData data = ordinalMap.getByteData().getUnderlyingArray();
@@ -194,9 +205,10 @@ private void calculateNumShards() {
         long projectedSizeOfType = (bitsPerMapSizeValue + bitsPerMapPointer) * (maxOrdinal + 1) / 8;
         projectedSizeOfType += ((bitsPerKeyElement + bitsPerValueElement) * totalOfMapBuckets) / 8;
 
-        numShards = 1;
-        while(stateEngine.getTargetMaxTypeShardSize() * numShards < projectedSizeOfType) 
-            numShards *= 2;
+        int targetNumShards = 1;
+        while(stateEngine.getTargetMaxTypeShardSize() * targetNumShards < projectedSizeOfType)
+            targetNumShards *= 2;
+        return targetNumShards;
     }
 
     @Override