diff --git a/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryKeyIndex.java b/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryKeyIndex.java
index 24cdf68365..1f163308ab 100644
--- a/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryKeyIndex.java
+++ b/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryKeyIndex.java
@@ -18,20 +18,10 @@
 
 import com.netflix.hollow.core.HollowDataset;
 import com.netflix.hollow.core.index.key.PrimaryKey;
-import com.netflix.hollow.core.read.HollowBlobInput;
-import com.netflix.hollow.core.read.engine.HollowBlobReader;
 import com.netflix.hollow.core.read.engine.HollowReadStateEngine;
 import com.netflix.hollow.core.read.engine.object.HollowObjectTypeReadState;
 import com.netflix.hollow.core.util.SimultaneousExecutor;
-import com.netflix.hollow.core.write.HollowBlobWriter;
-import com.netflix.hollow.core.write.HollowWriteStateEngine;
 import com.netflix.hollow.tools.history.HollowHistory;
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -46,18 +36,15 @@ public class HollowHistoryKeyIndex {
 
     private final HollowHistory history;
     private final Map<String, HollowHistoryTypeKeyIndex> typeKeyIndexes;
-    private final HollowWriteStateEngine indexWriteStateEngine;
-    private HollowReadStateEngine indexReadStateEngine;
+    private boolean isInitialized;
 
     public HollowHistoryKeyIndex(HollowHistory history) {
         this.history = history;
-        this.typeKeyIndexes = new HashMap<String, HollowHistoryTypeKeyIndex>();
-        this.indexWriteStateEngine = new HollowWriteStateEngine();
-        this.indexReadStateEngine = new HollowReadStateEngine();
+        this.typeKeyIndexes = new HashMap<>();
     }
 
     public int numUniqueKeys(String type) {
-        return indexReadStateEngine.getTypeState(type).maxOrdinal() + 1;
+        return typeKeyIndexes.get(type).getMaxIndexedOrdinal();
     }
 
     public String getKeyDisplayString(String type, int keyOrdinal) {
@@ -77,7 +64,7 @@ public void addTypeIndex(PrimaryKey primaryKey) {
     }
 
     public HollowHistoryTypeKeyIndex addTypeIndex(PrimaryKey primaryKey, HollowDataset dataModel) {
-        HollowHistoryTypeKeyIndex keyIdx = new HollowHistoryTypeKeyIndex(primaryKey, dataModel, indexWriteStateEngine, indexReadStateEngine);
+        HollowHistoryTypeKeyIndex keyIdx = new HollowHistoryTypeKeyIndex(primaryKey, dataModel);
         typeKeyIndexes.put(primaryKey.getType(), keyIdx);
         return keyIdx;
     }
@@ -117,24 +104,11 @@ public void update(HollowReadStateEngine latestStateEngine, boolean isDelta) {
 
         // This call updates the type key indexes of all types in this history key index.
         updateTypeIndexes(latestStateEngine, isDelta && !isInitialUpdate);
-
-        HollowReadStateEngine newIndexReadState = roundTripStateEngine(isInitialUpdate, !isDelta);
-
-        // if snapshot update then a new read state was generated, update the types in the history index to point to this
-        // new read state
-        if (newIndexReadState != indexReadStateEngine) {
-            // New ReadState was created so let's update references to old one
-            indexReadStateEngine = newIndexReadState;
-            for(final Map.Entry<String, HollowHistoryTypeKeyIndex> entry : typeKeyIndexes.entrySet()) {
-                entry.getValue().updateReadStateEngine(indexReadStateEngine);
-            }
-        }
-
-        rehashKeys();
+        isInitialized = true;
     }
 
     public boolean isInitialized() {
-        return !indexReadStateEngine.getTypeStates().isEmpty();
+        return isInitialized;
     }
 
     private void initializeTypeIndexes(HollowReadStateEngine latestStateEngine) {
@@ -146,7 +120,7 @@ private void initializeTypeIndexes(HollowReadStateEngine latestStateEngine) {
             HollowObjectTypeReadState typeState = (HollowObjectTypeReadState) latestStateEngine.getTypeState(type);
             if (typeState == null) continue;
 
-            index.initialize(typeState);
+            index.initializeKeySchema(typeState);
         }
     }
 
@@ -166,81 +140,4 @@ private void updateTypeIndexes(final HollowReadStateEngine latestStateEngine, fi
             throw new RuntimeException(e);
         }
     }
-
-    /**
-     * Updates the tracked readStateEngine based on populated writeStateEngine. Depending on whether this is an initial
-     * load or double snapshot etc. it invokes a snapshot or delta transition.
-     */
-    private HollowReadStateEngine roundTripStateEngine(boolean isInitialUpdate, boolean isSnapshot) {
-        HollowBlobWriter writer = new HollowBlobWriter(indexWriteStateEngine);
-        // Use existing readStateEngine on initial update or delta;
-        // otherwise, create new one to properly handle double snapshot
-        HollowReadStateEngine newReadStateEngine = (isInitialUpdate || !isSnapshot)
-                ? indexReadStateEngine : new HollowReadStateEngine();
-        HollowBlobReader reader = new HollowBlobReader(newReadStateEngine);
-
-        // Use a pipe to write and read concurrently to avoid writing
-        // to temporary files or allocating memory
-        // @@@ for small states it's more efficient to sequentially write to
-        // and read from a byte array but it is tricky to estimate the size
-        SimultaneousExecutor executor = new SimultaneousExecutor(1, HollowHistoryKeyIndex.class, "round-trip");
-        Exception pipeException = null;
-        // Ensure read-side is closed after completion of read
-        try (PipedInputStream is = new PipedInputStream(1 << 15)) {
-            BufferedOutputStream out = new BufferedOutputStream(new PipedOutputStream(is));
-            executor.execute(() -> {
-                // Ensure write-side is closed after completion of write
-                try (Closeable ac = out) {
-                    if (isInitialUpdate || isSnapshot) {
-                        writer.writeSnapshot(out);
-                    } else {
-                        writer.writeDelta(out);
-                    }
-                } catch (IOException e) {
-                    throw new RuntimeException(e);
-                }
-            });
-
-            HollowBlobInput in = HollowBlobInput.serial(new BufferedInputStream(is));
-            if (isInitialUpdate || isSnapshot) {
-                reader.readSnapshot(in);
-            } else {
-                reader.applyDelta(in);
-            }
-
-        } catch (Exception e) {
-            pipeException = e;
-        }
-
-        // Ensure no underlying writer exception is lost due to broken pipe
-        try {
-            executor.awaitSuccessfulCompletion();
-        } catch (InterruptedException | ExecutionException e) {
-            if (pipeException == null) {
-                throw new RuntimeException(e);
-            }
-
-            pipeException.addSuppressed(e);
-        }
-        if (pipeException != null)
-            throw new RuntimeException(pipeException);
-
-        indexWriteStateEngine.prepareForNextCycle();
-        return newReadStateEngine;
-    }
-
-    private void rehashKeys() {
-        SimultaneousExecutor executor = new SimultaneousExecutor(getClass(), "rehash-keys");
-
-        for(final Map.Entry<String, HollowHistoryTypeKeyIndex> entry : typeKeyIndexes.entrySet()) {
-            executor.execute(() -> entry.getValue().hashRecordKeys());
-        }
-
-        try {
-            executor.awaitSuccessfulCompletion();
-        } catch (InterruptedException | ExecutionException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
 }
diff --git a/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryTypeKeyIndex.java b/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryTypeKeyIndex.java
index 5d260439ae..3a1ba595a4 100644
--- a/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryTypeKeyIndex.java
+++ b/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowHistoryTypeKeyIndex.java
@@ -18,485 +18,245 @@
 
 import static com.netflix.hollow.core.HollowConstants.ORDINAL_NONE;
 import static com.netflix.hollow.tools.util.SearchUtils.MULTI_FIELD_KEY_DELIMITER;
-import static com.netflix.hollow.tools.util.SearchUtils.getFieldPathIndexes;
-import static com.netflix.hollow.tools.util.SearchUtils.getOrdinalToDisplay;
-import static com.netflix.hollow.tools.util.SearchUtils.parseKey;
 
+import com.netflix.hollow.Hollow;
 import com.netflix.hollow.core.HollowDataset;
 import com.netflix.hollow.core.index.key.PrimaryKey;
 import com.netflix.hollow.core.memory.encoding.HashCodes;
 import com.netflix.hollow.core.read.HollowReadFieldUtils;
-import com.netflix.hollow.core.read.engine.HollowReadStateEngine;
 import com.netflix.hollow.core.read.engine.PopulatedOrdinalListener;
 import com.netflix.hollow.core.read.engine.object.HollowObjectTypeReadState;
 import com.netflix.hollow.core.schema.HollowObjectSchema;
+import com.netflix.hollow.core.schema.HollowObjectSchema.FieldType;
 import com.netflix.hollow.core.util.IntList;
-import com.netflix.hollow.core.util.LongList;
 import com.netflix.hollow.core.util.RemovedOrdinalIterator;
-import com.netflix.hollow.core.write.HollowObjectTypeWriteState;
-import com.netflix.hollow.core.write.HollowObjectWriteRecord;
-import com.netflix.hollow.core.write.HollowTypeWriteState;
-import com.netflix.hollow.core.write.HollowWriteStateEngine;
 import java.util.Arrays;
 import java.util.BitSet;
+import java.util.HashMap;
 
 public class HollowHistoryTypeKeyIndex {
-
     private final PrimaryKey primaryKey;
-    private final String[][] keyFieldParts;
+    private final FieldType[] fieldTypes;
+
+    private final String[][] keyFieldNames;
+    private final int[][] keyFieldIndices;
     private final boolean[] keyFieldIsIndexed;
-    private HollowObjectSchema keySchema;
 
-    private int[] hashedRecordKeys;
-    private int[][] hashedFieldKeys;
-    private LongList hashedFieldKeyChains;
+    private boolean isInitialized = false;
+    private int maxIndexedOrdinal = 0;
 
-    private int maxIndexedKeyOrdinal = 0;
+    private final HollowOrdinalMapper ordinalMapping;
+    private final HashMap<Integer, IntList> ordinalFieldHashMapping;
 
-    private final HollowWriteStateEngine writeStateEngine;
-    private HollowReadStateEngine readStateEngine;
-    private boolean isInitialized = false;
 
-    public HollowHistoryTypeKeyIndex(PrimaryKey primaryKey, HollowDataset dataModel, HollowWriteStateEngine writeEngine, HollowReadStateEngine readEngine) {
+    public HollowHistoryTypeKeyIndex(PrimaryKey primaryKey, HollowDataset dataModel) {
         this.primaryKey = primaryKey;
-        this.writeStateEngine = writeEngine;
-        this.readStateEngine = readEngine;
-        this.keyFieldParts = getKeyFieldParts(dataModel);
+        this.fieldTypes = new FieldType[primaryKey.numFields()];
+
+        this.keyFieldNames = new String[primaryKey.numFields()][];
+        this.keyFieldIndices = new int[primaryKey.numFields()][];
         this.keyFieldIsIndexed = new boolean[primaryKey.numFields()];
+        initializeKeyParts(dataModel);
+
+        this.ordinalMapping = new HollowOrdinalMapper(primaryKey, keyFieldIsIndexed, keyFieldIndices);
+        this.ordinalFieldHashMapping = new HashMap<>();
+    }
+
+    public boolean isInitialized() {
+        return isInitialized;
+    }
+
+    public int findKeyIndexOrdinal(HollowObjectTypeReadState typeState, int ordinal) {
+        return ordinalMapping.findAssignedOrdinal(typeState, ordinal);
+    }
+
+    public int getMaxIndexedOrdinal() {
+        return maxIndexedOrdinal;
+    }
+
+    public String[] getKeyFields() {
+        return primaryKey.getFieldPaths();
     }
 
     public void addFieldIndex(String fieldName, HollowDataset dataModel) {
         String[] fieldPathParts = PrimaryKey.getCompleteFieldPathParts(dataModel, primaryKey.getType(), fieldName);
-        
-        for(int i=0;i<primaryKey.numFields();i++) {
+        for (int i = 0; i < primaryKey.numFields(); i++) {
             String[] pkFieldPathParts = PrimaryKey.getCompleteFieldPathParts(dataModel, primaryKey.getType(), primaryKey.getFieldPath(i));
-            if(Arrays.equals(pkFieldPathParts, fieldPathParts)) {
+            if (Arrays.equals(pkFieldPathParts, fieldPathParts)) {
                 keyFieldIsIndexed[i] = true;
                 break;
             }
         }
     }
 
-    public boolean isInitialized() {
-        return isInitialized;
-    }
-
-    public void initialize(HollowObjectTypeReadState initialTypeState) {
+    public void initializeKeySchema(HollowObjectTypeReadState initialTypeState) {
         if (isInitialized) return;
+        HollowObjectSchema schema = initialTypeState.getSchema();
 
-        initKeySchema(initialTypeState.getSchema());
-        initializeTypeWriteState();
+        for (int i= 0; i < keyFieldNames.length; i ++) {
+            String[] keyFieldPart = keyFieldNames[i];
+            fieldTypes[i] = addSchemaField(schema, keyFieldPart, 0);
+        }
         isInitialized = true;
     }
 
-    public void updateReadStateEngine(HollowReadStateEngine readEngine) {
-        readStateEngine=readEngine;
+    private FieldType addSchemaField(HollowObjectSchema schema, String[] keyFieldNames, int keyFieldPartPosition) {
+        int schemaPosition = schema.getPosition(keyFieldNames[keyFieldPartPosition]);
+        if (keyFieldPartPosition < keyFieldNames.length - 1) {
+            HollowObjectSchema nextPartSchema = (HollowObjectSchema) schema.getReferencedTypeState(schemaPosition).getSchema();
+            return addSchemaField(nextPartSchema, keyFieldNames, keyFieldPartPosition + 1);
+        }
+        return schema.getFieldType(schemaPosition);
+    }
+
+    private void initializeKeyParts(HollowDataset dataModel) {
+        for (int i = 0; i < primaryKey.numFields(); i++) {
+            keyFieldNames[i] = PrimaryKey.getCompleteFieldPathParts(dataModel, primaryKey.getType(), primaryKey.getFieldPath(i));
+            keyFieldIndices[i] = PrimaryKey.getFieldPathIndex(dataModel, primaryKey.getType(), primaryKey.getFieldPath(i));
+        }
     }
 
     public void update(HollowObjectTypeReadState latestTypeState, boolean isDeltaAndIndexInitialized) {
-        // copies over keys corresponding to previously populated ordinals AND currently populated ordinals
-        copyExistingKeys(); // only populates the ordianl map (to be the OR of previous and current ordinals)
         if (latestTypeState == null) return;
 
         if (isDeltaAndIndexInitialized) {
-            // record all newly seen keys when going from v1->v2 or v2->v1 to the history type key index
             populateNewCurrentRecordKeysIntoIndex(latestTypeState);
         } else {
-            // usually on the first fwd delta transition, or on  double snapshot
+            maxIndexedOrdinal = 0;
             populateAllCurrentRecordKeysIntoIndex(latestTypeState);
         }
     }
 
-    public void hashRecordKeys() {
-        HollowObjectTypeReadState keyTypeState = (HollowObjectTypeReadState) readStateEngine.getTypeState(primaryKey.getType());
-        if (keyTypeState == null) return;
-
-        int hashTableSize = HashCodes.hashTableSize(keyTypeState.maxOrdinal() + 1);
+    private void populateNewCurrentRecordKeysIntoIndex(HollowObjectTypeReadState typeState) {
+        PopulatedOrdinalListener listener = typeState.getListener(PopulatedOrdinalListener.class);
+        BitSet populatedOrdinals = listener.getPopulatedOrdinals();
+        BitSet previousOrdinals = listener.getPreviousOrdinals();
 
-        if(hashedRecordKeys == null || hashedRecordKeys.length < hashTableSize) {
-            rehashAllRecordKeys(keyTypeState, hashTableSize);
-        } else {
-            hashNewRecordKeys(keyTypeState);
+        RemovedOrdinalIterator iter = new RemovedOrdinalIterator(populatedOrdinals, previousOrdinals);
+        int ordinal = iter.next();
+        while (ordinal != ORDINAL_NONE) {
+            writeKeyObject(typeState, ordinal, true);
+            ordinal = iter.next();
         }
     }
 
-    private void hashNewRecordKeys(HollowObjectTypeReadState keyTypeState) {
-        for(int i=maxIndexedKeyOrdinal+1;i<=keyTypeState.maxOrdinal();i++)
-            indexOrdinal(keyTypeState, i, hashedRecordKeys, hashedFieldKeys, hashedFieldKeyChains);
-        maxIndexedKeyOrdinal = keyTypeState.maxOrdinal();
-    }
-
-    private void rehashAllRecordKeys(HollowObjectTypeReadState keyTypeState, int hashTableSize) {
-        int[] hashedRecordKeys = initializeHashedKeyArray(hashTableSize);
-        int[][] hashedFieldKeys = new int[primaryKey.numFields()][];
-        LongList hashedFieldKeyChains = new LongList();
-
-        for(int i=0;i<primaryKey.numFields();i++)
-            if(keyFieldIsIndexed[i])
-                hashedFieldKeys[i] = initializeHashedKeyArray(hashTableSize);
-
-        for(int i=0;i<=keyTypeState.maxOrdinal();i++)
-            indexOrdinal(keyTypeState, i, hashedRecordKeys, hashedFieldKeys, hashedFieldKeyChains);
-
-        this.hashedRecordKeys = hashedRecordKeys;
-        this.hashedFieldKeys = hashedFieldKeys;
-        this.hashedFieldKeyChains = hashedFieldKeyChains;
-        this.maxIndexedKeyOrdinal = keyTypeState.maxOrdinal();
-    }
-
-    private void indexOrdinal(HollowObjectTypeReadState keyTypeState, int ordinal, int[] hashedRecordKeys, int[][] hashedFieldKeys, LongList hashedFieldKeyChains) {
-        int bucketMask = hashedRecordKeys.length - 1;
-
-        int bucket = hashKeyRecord(keyTypeState, ordinal) & bucketMask;
-        while(hashedRecordKeys[bucket] != ORDINAL_NONE)
-            bucket = (bucket + 1) & bucketMask;
-        hashedRecordKeys[bucket] = ordinal;
-
-        for(int j=0;j<primaryKey.numFields();j++) {
-            if(keyFieldIsIndexed[j]) {
-                int fieldBucket = HashCodes.hashInt(HollowReadFieldUtils.fieldHashCode(keyTypeState, ordinal, j)) & bucketMask;
-                int chainStartIndex = hashedFieldKeys[j][fieldBucket];
-                while(chainStartIndex != ORDINAL_NONE) {
-                    int representativeOrdinal = (int)hashedFieldKeyChains.get(chainStartIndex);
-                    if(HollowReadFieldUtils.fieldsAreEqual(keyTypeState, ordinal, j, keyTypeState, representativeOrdinal, j)) {
-                        hashedFieldKeyChains.add(((long)chainStartIndex << 32) | ordinal);
-                        hashedFieldKeys[j][fieldBucket] = hashedFieldKeyChains.size() - 1;
-                        break;
-                    }
-                    fieldBucket = (fieldBucket + 1) & bucketMask;
-                    chainStartIndex = hashedFieldKeys[j][fieldBucket];
-                }
-                if (chainStartIndex == ORDINAL_NONE) {
-                    hashedFieldKeyChains.add(((long) Integer.MAX_VALUE << 32) | ordinal);
-                    hashedFieldKeys[j][fieldBucket] = hashedFieldKeyChains.size() - 1;
-                }
-            }
-        }
-    }
+    private void populateAllCurrentRecordKeysIntoIndex(HollowObjectTypeReadState typeState) {
+        PopulatedOrdinalListener listener = typeState.getListener(PopulatedOrdinalListener.class);
+        BitSet previousOrdinals = listener.getPreviousOrdinals();
+        BitSet populatedOrdinals = listener.getPopulatedOrdinals();
 
-    private int[] initializeHashedKeyArray(int hashTableSize) {
-        int[] hashedRecordKeys = new int[hashTableSize];
-        Arrays.fill(hashedRecordKeys, ORDINAL_NONE);
-        return hashedRecordKeys;
-    }
+        final int maxLength = Math.max(previousOrdinals.length(), populatedOrdinals.length());
 
-    private int hashKeyRecord(HollowObjectTypeReadState typeState, int ordinal) {
-        int hashCode = 0;
-        for(int i=0;i<primaryKey.numFields();i++) {
-            int fieldHashCode = HollowReadFieldUtils.fieldHashCode(typeState, ordinal, i);
-            hashCode = (hashCode * 31) ^ fieldHashCode;
+        for (int i = 0; i < maxLength; i++) {
+            if (populatedOrdinals.get(i) || previousOrdinals.get(i))
+                writeKeyObject(typeState, i, false);
         }
-        return HashCodes.hashInt(hashCode);
     }
 
-    public int findKeyIndexOrdinal(HollowObjectTypeReadState typeState, int ordinal) {
-        HollowObjectTypeReadState keyTypeState = (HollowObjectTypeReadState) readStateEngine.getTypeState(primaryKey.getType());
-
-        int bucketMask = hashedRecordKeys.length - 1;
-
-        int bucket = findKeyHashCode(typeState, ordinal) & bucketMask;
-
-        while(hashedRecordKeys[bucket] != ORDINAL_NONE) {
-            if(recordMatchesKey(typeState, ordinal, keyTypeState, hashedRecordKeys[bucket]))
-                return hashedRecordKeys[bucket];
-
-            bucket++;
-            bucket &= bucketMask;
+    private void writeKeyObject(HollowObjectTypeReadState typeState, int ordinal, boolean isDelta) {
+        int assignedOrdinal = isDelta ? maxIndexedOrdinal : ordinal;
+        maxIndexedOrdinal+=1;
+        int assignedIndex = ordinalMapping.storeNewRecord(typeState, ordinal, assignedOrdinal);
+
+        // Identical record already in memory, no need to store fields
+        if(assignedIndex==ORDINAL_NONE)
+            return;
+
+        for (int i = 0; i < primaryKey.numFields(); i++)
+            writeKeyField(assignedOrdinal, i);
+    }
+
+    private void writeKeyField(int assignedOrdinal, int fieldIdx) {
+        if (!keyFieldIsIndexed[fieldIdx])
+            return;
+
+        Object fieldObject = ordinalMapping.getFieldObject(assignedOrdinal, fieldIdx);
+        int fieldHash = HashCodes.hashInt(hashObject(fieldObject));
+        if(!ordinalFieldHashMapping.containsKey(fieldHash))
+            ordinalFieldHashMapping.put(fieldHash, new IntList());
+
+        IntList matchingFieldList = ordinalFieldHashMapping.get(fieldHash);
+        matchingFieldList.add(assignedOrdinal);
+    }
+
+    // Retain consistency with rest of Hollow Hashing when hashing boxed primitives
+    private int hashObject(Object value) {
+        if(value instanceof Integer) {
+            return HollowReadFieldUtils.intHashCode((Integer)value);
+        } else if(value instanceof String) {
+            return HollowReadFieldUtils.stringHashCode((String)value);
+        } else if(value instanceof Float) {
+            return HollowReadFieldUtils.floatHashCode((Float)value);
+        } else if(value instanceof Double) {
+            return HollowReadFieldUtils.doubleHashCode((Double)value);
+        } else if(value instanceof Boolean) {
+            return HollowReadFieldUtils.booleanHashCode((Boolean) value);
+        } else if(value instanceof Long) {
+            return HollowReadFieldUtils.longHashCode((Long) value);
+        } else if(value instanceof byte[]) {
+            return HollowReadFieldUtils.byteArrayHashCode((byte[]) value);
+        } else {
+            throw new RuntimeException("Unable to hash field of type " + value.getClass().getName());
         }
-
-        return ORDINAL_NONE;
     }
 
-    private int findKeyHashCode(HollowObjectTypeReadState typeState, int ordinal) {
-        int hashCode = 0;
-        for (String[] keyFieldPart : keyFieldParts) {
-            int fieldHashCode = findKeyFieldHashCode(typeState, ordinal, keyFieldPart, 0);
-            hashCode = (hashCode * 31) ^ fieldHashCode;
+    public String getKeyDisplayString(int keyOrdinal) {
+        StringBuilder builder = new StringBuilder();
+        for (int i = 0; i < primaryKey.numFields(); i++) {
+            Object valueAtField = ordinalMapping.getFieldObject(keyOrdinal, i);
+            builder.append(valueAtField);
+            if (i < primaryKey.numFields() - 1)
+                builder.append(MULTI_FIELD_KEY_DELIMITER);
         }
-        return HashCodes.hashInt(hashCode);
+        return builder.toString();
     }
 
     public IntList queryIndexedFields(final String query) {
-        final HollowObjectTypeReadState keyTypeState = (HollowObjectTypeReadState) readStateEngine.getTypeState(primaryKey.getType());
         IntList matchingKeys = new IntList();
-        if (keyTypeState == null){
-            return matchingKeys;
-        }
 
-        String[] parts;
-        if (query.contains(MULTI_FIELD_KEY_DELIMITER)) {  // composite field query, uses ':' as separator
-            parts = query.split(MULTI_FIELD_KEY_DELIMITER);
-            if (parts.length != primaryKey.numFields()) {
-                return matchingKeys;
-            }
-
-            Object[] parsedKey;
-            try {
-                parsedKey = parseKey(readStateEngine, primaryKey, query);
-            } catch(Exception e) {
-                return matchingKeys;
-            }
-
-            BitSet selectedOrdinals = keyTypeState.getPopulatedOrdinals();
-            int fieldPathIndexes[][] = getFieldPathIndexes(readStateEngine, primaryKey);
-            int ordinal = getOrdinalToDisplay(readStateEngine, query, parsedKey, ORDINAL_NONE, selectedOrdinals, fieldPathIndexes, keyTypeState);
-            matchingKeys.add(ordinal);
+        if (!isInitialized) {
             return matchingKeys;
         }
 
-        // match query against each indexed field
-        for(int i=0;i<primaryKey.numFields();i++) {
-            final int fieldIndex = i;
+        for (int i = 0; i < primaryKey.numFields(); i++) {
+            int hashCode = 0;
             try {
-                int hashCode;
-                if(keyFieldIsIndexed[i]) {
-                    switch(keySchema.getFieldType(i)) {
+                switch (fieldTypes[i]) {
                     case INT:
                         final int queryInt = Integer.parseInt(query);
                         hashCode = HollowReadFieldUtils.intHashCode(queryInt);
-                        addMatches(new Matcher() {
-                            public boolean foundMatch(int ordinal) {
-                                return keyTypeState.readInt(ordinal, fieldIndex) == queryInt;
-                            }
-                        }, i, hashCode, matchingKeys);
                         break;
                     case LONG:
                         final long queryLong = Long.parseLong(query);
                         hashCode = HollowReadFieldUtils.longHashCode(queryLong);
-                        addMatches(new Matcher() {
-                            public boolean foundMatch(int ordinal) {
-                                return keyTypeState.readLong(ordinal, fieldIndex) == queryLong;
-                            }
-                        }, i, hashCode, matchingKeys);
                         break;
                     case STRING:
                         hashCode = HashCodes.hashCode(query);
-                        addMatches(new Matcher() {
-                            public boolean foundMatch(int ordinal) {
-                                return keyTypeState.isStringFieldEqual(ordinal, fieldIndex, query);
-                            }
-                        }, i, hashCode, matchingKeys);
                         break;
                     case DOUBLE:
                         final double queryDouble = Double.parseDouble(query);
                         hashCode = HollowReadFieldUtils.doubleHashCode(queryDouble);
-                        addMatches(new Matcher() {
-                            public boolean foundMatch(int ordinal) {
-                                return keyTypeState.readDouble(ordinal, fieldIndex) == queryDouble;
-                            }
-                        }, i, hashCode, matchingKeys);
                         break;
                     case FLOAT:
                         final float queryFloat = Float.parseFloat(query);
                         hashCode = HollowReadFieldUtils.floatHashCode(queryFloat);
-                        addMatches(new Matcher() {
-                            public boolean foundMatch(int ordinal) {
-                                return keyTypeState.readFloat(ordinal, fieldIndex) == queryFloat;
-                            }
-                        }, i, hashCode, matchingKeys);
                         break;
                     default:
-                    }
                 }
-            } catch(NumberFormatException ignore) { }
+                addMatches(HashCodes.hashInt(hashCode), matchingKeys);
+            } catch(NumberFormatException ignore) {}
         }
-
         return matchingKeys;
     }
 
-    private void addMatches(Matcher matcher, int fieldIndex, int hashCode, IntList results) {
-        int hashIntCode = HashCodes.hashInt(hashCode);
-        int bucket = hashIntCode & (hashedFieldKeys[fieldIndex].length - 1);
-
-        while(hashedFieldKeys[fieldIndex][bucket] != ORDINAL_NONE) {
-            int chainIndex = hashedFieldKeys[fieldIndex][bucket];
-            int representativeOrdinal = (int)hashedFieldKeyChains.get(chainIndex);
-            if(matcher.foundMatch(representativeOrdinal)) {
-                while(representativeOrdinal != ORDINAL_NONE) {
-                    results.add(representativeOrdinal);
-                    chainIndex = (int)(hashedFieldKeyChains.get(chainIndex) >> 32);
-                    representativeOrdinal = (chainIndex == Integer.MAX_VALUE) ? ORDINAL_NONE : (int)hashedFieldKeyChains.get(chainIndex);
-                }
-                return;
-            }
-
-            bucket++;
-            bucket &= hashedFieldKeys[fieldIndex].length - 1;
-        }
-    }
-
-    private interface Matcher {
-        boolean foundMatch(int ordinal);
-    }
-
-    private int findKeyFieldHashCode(HollowObjectTypeReadState typeState, int ordinal, String[] keyFieldParts, int keyFieldPartPosition) {
-        int schemaPosition = typeState.getSchema().getPosition(keyFieldParts[keyFieldPartPosition]);
-        if(keyFieldPartPosition < keyFieldParts.length - 1) {
-            HollowObjectTypeReadState nextPartTypeState = (HollowObjectTypeReadState) typeState.getSchema().getReferencedTypeState(schemaPosition);
-            int nextOrdinal = typeState.readOrdinal(ordinal, schemaPosition);
-            return findKeyFieldHashCode(nextPartTypeState, nextOrdinal, keyFieldParts, keyFieldPartPosition + 1);
-        } else {
-            return HollowReadFieldUtils.fieldHashCode(typeState, ordinal, schemaPosition);
-        }
-    }
-
-    private boolean recordMatchesKey(HollowObjectTypeReadState typeState, int ordinal, HollowObjectTypeReadState keyTypeState, int keyOrdinal) {
-        for(int i=0;i<keyFieldParts.length;i++) {
-            if(!recordFieldMatchesKey(typeState, ordinal, keyTypeState, keyOrdinal, i, keyFieldParts[i], 0))
-                return false;
-        }
-        return true;
-    }
-
-    private boolean recordFieldMatchesKey(HollowObjectTypeReadState typeState, int ordinal, HollowObjectTypeReadState keyTypeState, int keyOrdinal, int keyFieldPosition, String[] keyFieldParts, int keyFieldPartPosition) {
-        int schemaPosition = typeState.getSchema().getPosition(keyFieldParts[keyFieldPartPosition]);
-        if(keyFieldPartPosition < keyFieldParts.length - 1) {
-            HollowObjectTypeReadState nextPartTypeState = (HollowObjectTypeReadState) typeState.getSchema().getReferencedTypeState(schemaPosition);
-            int nextOrdinal = typeState.readOrdinal(ordinal, schemaPosition);
-            return recordFieldMatchesKey(nextPartTypeState, nextOrdinal, keyTypeState, keyOrdinal, keyFieldPosition, keyFieldParts, keyFieldPartPosition + 1);
-        } else {
-            return HollowReadFieldUtils.fieldsAreEqual(typeState, ordinal, schemaPosition, keyTypeState, keyOrdinal, keyFieldPosition);
-        }
-    }
-
-    private void copyExistingKeys() {
-        HollowTypeWriteState typeState = writeStateEngine.getTypeState(primaryKey.getType());
-        if (typeState == null) return;
-
-        typeState.addAllObjectsFromPreviousCycle();
-    }
-
-    private void populateAllCurrentRecordKeysIntoIndex(HollowObjectTypeReadState typeState) {
-        HollowObjectWriteRecord rec = new HollowObjectWriteRecord(keySchema);
-        PopulatedOrdinalListener listener = typeState.getListener(PopulatedOrdinalListener.class);
-        BitSet previousOrdinals = listener.getPreviousOrdinals();
-        BitSet populatedOrdinals = listener.getPopulatedOrdinals();
-
-        int maxLength = Math.max(previousOrdinals.length(), populatedOrdinals.length());
-
-        for(int i=0;i<maxLength;i++) {
-            if(populatedOrdinals.get(i) || previousOrdinals.get(i))
-                writeKeyObject(typeState, i, rec);
-        }
-    }
-
-    private void populateNewCurrentRecordKeysIntoIndex(HollowObjectTypeReadState typeState) {
-        HollowObjectWriteRecord rec = new HollowObjectWriteRecord(keySchema);
-        PopulatedOrdinalListener listener = typeState.getListener(PopulatedOrdinalListener.class);
-        BitSet populatedOrdinals = listener.getPopulatedOrdinals();
-        BitSet previousOrdinals = listener.getPreviousOrdinals();
-
-        RemovedOrdinalIterator iter = new RemovedOrdinalIterator(populatedOrdinals, previousOrdinals);
-        int ordinal = iter.next();
-        while(ordinal != ORDINAL_NONE) {
-            writeKeyObject(typeState, ordinal, rec);
-            ordinal = iter.next();
-        }
-    }
-
-    public String[] getKeyFields() {
-        return primaryKey.getFieldPaths();
+    public void addMatches(int hashCode, IntList results) {
+        if (!ordinalFieldHashMapping.containsKey(hashCode))
+            return;
+        IntList res2 = ordinalFieldHashMapping.get(hashCode);
+        results.addAll(res2);
     }
 
     public Object getKeyFieldValue(int keyFieldIdx, int keyOrdinal) {
-        return HollowReadFieldUtils.fieldValueObject((HollowObjectTypeReadState)readStateEngine.getTypeState(primaryKey.getType()), keyOrdinal, keyFieldIdx);
-    }
-
-    public String getKeyDisplayString(int keyOrdinal) {
-        HollowObjectTypeReadState typeState = (HollowObjectTypeReadState) readStateEngine.getTypeState(primaryKey.getType());
-
-        StringBuilder builder = new StringBuilder();
-        for(int i=0;i<primaryKey.numFields();i++) {
-            builder.append(HollowReadFieldUtils.displayString(typeState, keyOrdinal, i));
-            if(i < primaryKey.numFields() - 1)
-                builder.append(MULTI_FIELD_KEY_DELIMITER);
-        }
-        return builder.toString();
-    }
-
-    private void writeKeyObject(HollowObjectTypeReadState typeState, int ordinal, HollowObjectWriteRecord rec) {
-        rec.reset();
-        for(int i=0;i<keyFieldParts.length;i++) {
-            writeKeyField(typeState, ordinal, rec, primaryKey.getFieldPath(i), keyFieldParts[i], 0);
-        }
-        writeStateEngine.add(primaryKey.getType(), rec);
-    }
-
-    private void writeKeyField(HollowObjectTypeReadState typeState, int ordinal, HollowObjectWriteRecord rec, String keyField, String[] keyFieldParts, int keyFieldPartPosition) {
-        int schemaPosition = typeState.getSchema().getPosition(keyFieldParts[keyFieldPartPosition]);
-        if(keyFieldPartPosition < keyFieldParts.length - 1) {
-            HollowObjectTypeReadState nextPartTypeState = (HollowObjectTypeReadState) typeState.getSchema().getReferencedTypeState(schemaPosition);
-            int nextOrdinal = typeState.readOrdinal(ordinal, schemaPosition);
-            writeKeyField(nextPartTypeState, nextOrdinal, rec, keyField, keyFieldParts, keyFieldPartPosition + 1);
-        } else {
-            switch(typeState.getSchema().getFieldType(schemaPosition)) {
-            case BOOLEAN:
-                Boolean bool = typeState.readBoolean(ordinal, schemaPosition);
-                if(bool != null)
-                    rec.setBoolean(keyField, bool);
-                break;
-            case BYTES:
-                byte[] b = typeState.readBytes(ordinal, schemaPosition);
-                if(b != null)
-                    rec.setBytes(keyField, b);
-                break;
-            case DOUBLE:
-                double d = typeState.readDouble(ordinal, schemaPosition);
-                if(!Double.isNaN(d))
-                    rec.setDouble(keyField, d);
-                break;
-            case FLOAT:
-                float f = typeState.readFloat(ordinal, schemaPosition);
-                if(!Float.isNaN(f))
-                    rec.setFloat(keyField, f);
-                break;
-            case INT:
-                int i = typeState.readInt(ordinal, schemaPosition);
-                rec.setInt(keyField, i);
-                break;
-            case LONG:
-                long l = typeState.readLong(ordinal, schemaPosition);
-                rec.setLong(keyField, l);
-                break;
-            case STRING:
-                String s = typeState.readString(ordinal, schemaPosition);
-                if(s != null)
-                    rec.setString(keyField, s);
-                break;
-            default:
-                throw new IllegalArgumentException("Primary key components must be a value leaf node");
-            }
-        }
+        return ordinalMapping.getFieldObject(keyOrdinal, keyFieldIdx);
     }
-
-    private void initKeySchema(HollowObjectSchema entireObjectSchema) {
-        if(keySchema == null) {
-            keySchema = new HollowObjectSchema(primaryKey.getType(), primaryKey.numFields());
-            for(int i=0;i<keyFieldParts.length;i++)
-                addSchemaField(entireObjectSchema, keySchema, primaryKey.getFieldPath(i), keyFieldParts[i], 0);
-        }
-    }
-
-    private void addSchemaField(HollowObjectSchema schema, HollowObjectSchema keySchema, String keyField, String[] keyFieldParts, int keyFieldPartPosition) {
-        int schemaPosition = schema.getPosition(keyFieldParts[keyFieldPartPosition]);
-        if(keyFieldPartPosition < keyFieldParts.length - 1) {
-            HollowObjectSchema nextPartSchema = (HollowObjectSchema) schema.getReferencedTypeState(schemaPosition).getSchema();
-            addSchemaField(nextPartSchema, keySchema, keyField, keyFieldParts, keyFieldPartPosition + 1);
-        } else {
-            keySchema.addField(keyField, schema.getFieldType(schemaPosition), schema.getReferencedType(schemaPosition));
-        }
-    }
-
-    private void initializeTypeWriteState() {
-        HollowObjectTypeWriteState writeState = new HollowObjectTypeWriteState(keySchema);
-        writeStateEngine.addTypeState(writeState);
-    }
-
-    private String[][] getKeyFieldParts(HollowDataset dataModel) {
-        String[][] keyFieldParts = new String[primaryKey.numFields()][];
-        for(int i=0;i<primaryKey.numFields();i++)
-            keyFieldParts[i] = PrimaryKey.getCompleteFieldPathParts(dataModel, primaryKey.getType(), primaryKey.getFieldPath(i));
-        return keyFieldParts;
-    }
-
-}
+}
\ No newline at end of file
diff --git a/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowOrdinalMapper.java b/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowOrdinalMapper.java
new file mode 100644
index 0000000000..9b3e410cbe
--- /dev/null
+++ b/hollow/src/main/java/com/netflix/hollow/tools/history/keyindex/HollowOrdinalMapper.java
@@ -0,0 +1,196 @@
+/*
+ *  Copyright 2016-2019 Netflix, Inc.
+ *
+ *     Licensed 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 com.netflix.hollow.tools.history.keyindex;
+
+import com.netflix.hollow.core.index.key.PrimaryKey;
+import com.netflix.hollow.core.memory.encoding.HashCodes;
+import com.netflix.hollow.core.read.HollowReadFieldUtils;
+import com.netflix.hollow.core.read.engine.object.HollowObjectTypeReadState;
+import com.netflix.hollow.core.schema.HollowObjectSchema;
+import com.netflix.hollow.tools.util.ObjectInternPool;
+
+import static com.netflix.hollow.core.HollowConstants.ORDINAL_NONE;
+import java.util.Arrays;
+import java.util.HashMap;
+
+public class HollowOrdinalMapper {
+    private int size = 0;
+    private static final double LOAD_FACTOR = 0.7;
+    
+    private Integer[] ordinalMappings;
+    private Integer[] originalHash;
+
+    private HashMap<Integer, Object[]> indexFieldObjectMapping;
+    private final HashMap<Integer, Integer> assignedOrdinalToIndex;
+
+    private final PrimaryKey primaryKey;
+    private final int[][] keyFieldIndices;
+    private final boolean[] keyFieldIsIndexed;
+
+    private final ObjectInternPool memoizedPool;
+
+    public HollowOrdinalMapper(PrimaryKey primaryKey, boolean[] keyFieldIsIndexed, int[][] keyFieldIndices) {
+        // Start with prime number to assist OA
+        this.ordinalMappings = new Integer[2069];
+        this.originalHash = new Integer[2069];
+        Arrays.fill(this.ordinalMappings, ORDINAL_NONE);
+
+        this.primaryKey = primaryKey;
+        this.keyFieldIndices = keyFieldIndices;
+        this.keyFieldIsIndexed = keyFieldIsIndexed;
+
+        this.indexFieldObjectMapping = new HashMap<>();
+        this.assignedOrdinalToIndex = new HashMap<>();
+
+        this.memoizedPool = new ObjectInternPool();
+    }
+
+    public int findAssignedOrdinal(HollowObjectTypeReadState typeState, int keyOrdinal) {
+        int hashedRecord = hashKeyRecord(typeState, keyOrdinal);
+        int index = indexFromHash(hashedRecord, ordinalMappings.length);
+
+        while (ordinalMappings[index]!=ORDINAL_NONE) {
+            if(recordsAreEqual(typeState, keyOrdinal, index))
+                return ordinalMappings[index];
+            index = (index + 1) % ordinalMappings.length;
+        }
+
+        return ORDINAL_NONE;
+    }
+
+    private boolean recordsAreEqual(HollowObjectTypeReadState typeState, int keyOrdinal, int index) {
+        for(int fieldIdx=0;fieldIdx<primaryKey.numFields();fieldIdx++) {
+            if(!keyFieldIsIndexed[fieldIdx])
+                continue;
+
+            Object newFieldValue = readValueInState(typeState, keyOrdinal, fieldIdx);
+            Object existingFieldValue = indexFieldObjectMapping.get(index)[fieldIdx];
+            if(!newFieldValue.equals(existingFieldValue)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public int storeNewRecord(HollowObjectTypeReadState typeState, int ordinal, int assignedOrdinal) {
+        int hashedRecord = hashKeyRecord(typeState, ordinal);
+
+        if ((double) size / ordinalMappings.length > LOAD_FACTOR) {
+            expandAndRehashTable();
+        }
+
+        int index = indexFromHash(hashedRecord, ordinalMappings.length);
+
+        // Linear probing
+        while (ordinalMappings[index] != ORDINAL_NONE) {
+            if(recordsAreEqual(typeState, ordinal, index)) {
+                this.assignedOrdinalToIndex.put(assignedOrdinal, index);
+                return ORDINAL_NONE;
+            }
+            index = (index + 1) % ordinalMappings.length;
+        }
+
+        ordinalMappings[index] = assignedOrdinal;
+        originalHash[index] = hashedRecord;
+        size++;
+
+        storeFields(typeState, ordinal, index);
+
+        this.assignedOrdinalToIndex.put(assignedOrdinal, index);
+        return index;
+    }
+
+    private void storeFields(HollowObjectTypeReadState typeState, int ordinal, int index) {
+        if(!indexFieldObjectMapping.containsKey(index))
+            indexFieldObjectMapping.put(index, new Object[primaryKey.numFields()]);
+
+        for(int i=0;i<primaryKey.numFields();i++) {
+            if(!keyFieldIsIndexed[i])
+                continue;
+
+            Object objectToStore = readValueInState(typeState, ordinal, i);
+            indexFieldObjectMapping.get(index)[i] = memoizedPool.intern(objectToStore);
+        }
+    }
+
+    private void expandAndRehashTable() {
+        Integer[] newTable = new Integer[ordinalMappings.length*2];
+        Arrays.fill(newTable, ORDINAL_NONE);
+
+        Integer[] newOriginalHash = new Integer[originalHash.length*2];
+        HashMap<Integer, Object[]> newIndexFieldObjectMapping = new HashMap<>();
+
+        for(int i=0;i<ordinalMappings.length;i++) {
+            if(ordinalMappings[i]==ORDINAL_NONE)
+                continue;
+            int newIndex = rehashExistingRecord(newTable, originalHash[i], ordinalMappings[i]);
+            newOriginalHash[newIndex] = originalHash[i];
+
+            Object[] fieldObjects = indexFieldObjectMapping.get(i);
+            newIndexFieldObjectMapping.put(newIndex, fieldObjects);
+        }
+
+        this.ordinalMappings = newTable;
+        this.originalHash = newOriginalHash;
+        this.indexFieldObjectMapping = newIndexFieldObjectMapping;
+    }
+
+    private int rehashExistingRecord(Integer[] newTable, int originalHash, int assignedOrdinal) {
+        int newIndex = indexFromHash(originalHash, newTable.length);
+        while (newTable[newIndex]!=ORDINAL_NONE)
+            newIndex = (newIndex + 1) % newTable.length;
+
+        assignedOrdinalToIndex.put(assignedOrdinal, newIndex);
+        newTable[newIndex] = assignedOrdinal;
+        return newIndex;
+    }
+
+    public Object getFieldObject(int keyOrdinal, int fieldIndex) {
+        int index = assignedOrdinalToIndex.get(keyOrdinal);
+        return indexFieldObjectMapping.get(index)[fieldIndex];
+    }
+
+    private int hashKeyRecord(HollowObjectTypeReadState typeState, int ordinal) {
+        int hashCode = 0;
+        for (int i = 0; i < primaryKey.numFields(); i++) {
+            int fieldHashCode = HollowReadFieldUtils.fieldHashCode(typeState, ordinal, i);
+            hashCode = (hashCode * 31) ^ fieldHashCode;
+        }
+        return HashCodes.hashInt(hashCode);
+    }
+
+    //taken and modified from HollowPrimaryKeyValueDeriver
+    private Object readValueInState(HollowObjectTypeReadState typeState, int ordinal, int fieldIdx) {
+        HollowObjectSchema schema = typeState.getSchema();
+
+        int lastFieldPath = keyFieldIndices[fieldIdx].length - 1;
+        for (int i = 0; i < lastFieldPath; i++) {
+            int fieldPosition = keyFieldIndices[fieldIdx][i];
+            ordinal = typeState.readOrdinal(ordinal, fieldPosition);
+            typeState = (HollowObjectTypeReadState) schema.getReferencedTypeState(fieldPosition);
+            schema = typeState.getSchema();
+        }
+
+        return HollowReadFieldUtils.fieldValueObject(typeState, ordinal, keyFieldIndices[fieldIdx][lastFieldPath]);
+    }
+
+    // Java modulo is more like a remainder, indices can't be negative
+    private static int indexFromHash(int hashedValue, int length) {
+        int modulus = hashedValue % length;
+        return modulus < 0 ? modulus + length : modulus;
+    }
+}
\ No newline at end of file
diff --git a/hollow/src/main/java/com/netflix/hollow/tools/util/ObjectInternPool.java b/hollow/src/main/java/com/netflix/hollow/tools/util/ObjectInternPool.java
new file mode 100644
index 0000000000..fa862a2ff4
--- /dev/null
+++ b/hollow/src/main/java/com/netflix/hollow/tools/util/ObjectInternPool.java
@@ -0,0 +1,68 @@
+package com.netflix.hollow.tools.util;
+
+import java.util.Map;
+import java.util.HashMap;
+
+// This class memoizes types by returning references to existing objects, or storing
+// Objects if they are not currently in the pool
+public class ObjectInternPool {
+    final private GenericInternPool<Integer> integerInternPool = new GenericInternPool<>();
+    final private GenericInternPool<Float> floatInternPool = new GenericInternPool<>();
+    final private GenericInternPool<Double> doubleInternPool = new GenericInternPool<>();
+    final private GenericInternPool<Long> longInternPool = new GenericInternPool<>();
+
+    public Object intern(Object objectToIntern) {
+        if(objectToIntern==null) {
+            throw new IllegalArgumentException("Cannot intern null objects");
+        }
+
+        // Automatically handles booleans and integers within cached range
+        if(objectAutomaticallyCached(objectToIntern)) {
+            return objectToIntern;
+        }
+
+        if(objectToIntern instanceof Float) {
+            return floatInternPool.intern((Float) objectToIntern);
+        } else if(objectToIntern instanceof Double) {
+            return doubleInternPool.intern((Double) objectToIntern);
+        } else if(objectToIntern instanceof Integer) {
+            return integerInternPool.intern((Integer) objectToIntern);
+        } else if(objectToIntern instanceof Long) {
+            return longInternPool.intern((Long) objectToIntern);
+        } else if(objectToIntern instanceof String) {
+            // Use Java's builtin intern function
+            return ((String) objectToIntern).intern();
+        } else {
+            String className = objectToIntern.getClass().getName();
+            throw new IllegalArgumentException("Cannot intern object of type " + className);
+        }
+    }
+
+    private boolean objectAutomaticallyCached(Object objectToIntern) {
+        if(objectToIntern instanceof Boolean) {
+            return true;
+        } else if(objectToIntern instanceof Integer) {
+            return -128 <= (Integer) objectToIntern && (Integer) objectToIntern <= 127;
+        }
+        return false;
+    }
+}
+
+class GenericInternPool<T> {
+    private final Map<T, T> pool = new HashMap<>();
+
+    public T intern(T object) {
+        if (object == null) {
+            throw new IllegalArgumentException("Cannot intern null objects");
+        }
+
+        synchronized (pool) {
+            T interned = pool.get(object);
+            if (interned == null) {
+                interned = object;
+                pool.put(object, object);
+            }
+            return interned;
+        }
+    }
+}
\ No newline at end of file