From 652afc377eef57523e260a200b3e40ddbd6f2a3f Mon Sep 17 00:00:00 2001
From: Matthieu Monsch <monsch@alum.mit.edu>
Date: Tue, 17 Nov 2015 07:50:02 -0800
Subject: [PATCH] Share slot read buffers.

Rather than keep a separate `slotBuffer` per key length, we can track
the size of the largest slot and create a common one.  If there are many
different sizes, this could save a reasonable amount of in-heap memory.
---
 .../com/linkedin/paldb/impl/StorageReader.java  | 17 +++++++++--------
 1 file changed, 9 insertions(+), 8 deletions(-)

diff --git a/paldb/src/main/java/com/linkedin/paldb/impl/StorageReader.java b/paldb/src/main/java/com/linkedin/paldb/impl/StorageReader.java
index 4111dac..5fb26fe 100644
--- a/paldb/src/main/java/com/linkedin/paldb/impl/StorageReader.java
+++ b/paldb/src/main/java/com/linkedin/paldb/impl/StorageReader.java
@@ -85,7 +85,7 @@ public class StorageReader implements Iterable<Map.Entry<byte[], byte[]>> {
   private final boolean mMapData;
   // Buffers
   private final DataInputOutput sizeBuffer = new DataInputOutput(new byte[5]);
-  private final byte[][] slotBuffers;
+  private final byte[] slotBuffer;
 
   StorageReader(Configuration configuration, File file)
       throws IOException {
@@ -148,20 +148,22 @@ public class StorageReader implements Iterable<Map.Entry<byte[], byte[]>> {
     keyCounts = new int[maxKeyLength + 1];
     slots = new int[maxKeyLength + 1];
     slotSizes = new int[maxKeyLength + 1];
-    slotBuffers = new byte[maxKeyLength + 1][];
+
+    int maxSlotSize = 0;
     for (int i = 0; i < keyLengthCount; i++) {
       int keyLength = dataInputStream.readInt();
 
       keyCounts[keyLength] = dataInputStream.readInt();
       slots[keyLength] = dataInputStream.readInt();
       slotSizes[keyLength] = dataInputStream.readInt();
-      slotBuffers[keyLength] = new byte[slotSizes[keyLength]];
-
       indexOffsets[keyLength] = dataInputStream.readInt();
-
       dataOffsets[keyLength] = dataInputStream.readLong();
+
+      maxSlotSize = Math.max(maxSlotSize, slotSizes[keyLength]);
     }
 
+    slotBuffer = new byte[maxSlotSize];
+
     //Read serializers
     try {
       Serializers.deserialize(dataInputStream, config.getSerializers());
@@ -238,15 +240,14 @@ public byte[] get(byte[] key)
     }
     long hash = (long) HashUtils.hash(key);
     int numSlots = slots[keyLength];
-    byte[] slotBuffer = slotBuffers[keyLength];
-    int slotSize = slotBuffer.length;
+    int slotSize = slotSizes[keyLength];
     int indexOffset = indexOffsets[keyLength];
     long dataOffset = dataOffsets[keyLength];
 
     for (int probe = 0; probe < numSlots; probe++) {
       int slot = (int) ((hash + probe) % numSlots);
       indexBuffer.position(indexOffset + slot * slotSize);
-      indexBuffer.get(slotBuffer);
+      indexBuffer.get(slotBuffer, 0, slotSize);
 
       long offset = LongPacker.unpackLong(slotBuffer, keyLength);
       if (offset == 0) {