SSTable lazy retrieval, 15% speedup

tomfran · Oct 2, 2023 · 9d223ee · 9d223ee
1 parent 281f87d
commit 9d223ee
Show file tree

Hide file tree

Showing 18 changed files with 225 additions and 226 deletions.
diff --git a/src/jmh/java/com/tomfran/lsm/memtable/SkipListBenchmark.java b/src/jmh/java/com/tomfran/lsm/memtable/SkipListBenchmark.java
@@ -1,20 +1,20 @@
 package com.tomfran.lsm.memtable;
 
-import com.tomfran.lsm.types.Item;
+import com.tomfran.lsm.types.ByteArrayPair;
 import it.unimi.dsi.fastutil.objects.ObjectArrayList;
 import org.openjdk.jmh.annotations.*;
 import org.openjdk.jmh.infra.Blackhole;
 
 import java.util.concurrent.TimeUnit;
 
-import static com.tomfran.lsm.TestUtils.getRandomItem;
+import static com.tomfran.lsm.TestUtils.getRandomPair;
 
 @OutputTimeUnit(TimeUnit.SECONDS)
 @State(Scope.Benchmark)
 public class SkipListBenchmark {
 
     SkipList l;
-    Item[] items;
+    ByteArrayPair[] items;
 
     int NUM_ITEMS = 200000;
     int index = 0;
@@ -27,16 +27,16 @@ public void setup() {
         l = new SkipList(NUM_ITEMS / 2);
 
         // generate random items and insert half
-        ObjectArrayList<Item> tmp = new ObjectArrayList<>();
+        ObjectArrayList<ByteArrayPair> tmp = new ObjectArrayList<>();
         for (int i = 0; i < NUM_ITEMS; i++) {
-            var it = getRandomItem();
+            var it = getRandomPair();
             if (i < NUM_ITEMS / 2)
                 l.add(it);
 
             tmp.add(it);
         }
 
-        items = tmp.toArray(new Item[0]);
+        items = tmp.toArray(new ByteArrayPair[0]);
 
         // generate sequence of add/remove operations
         addRemove = new boolean[NUM_ITEMS];

diff --git a/src/jmh/java/com/tomfran/lsm/sstable/SSTableBenchmark.java b/src/jmh/java/com/tomfran/lsm/sstable/SSTableBenchmark.java
@@ -1,7 +1,7 @@
 package com.tomfran.lsm.sstable;
 
 import com.tomfran.lsm.comparator.ByteArrayComparator;
-import com.tomfran.lsm.types.Item;
+import com.tomfran.lsm.types.ByteArrayPair;
 import it.unimi.dsi.fastutil.objects.ObjectArrayList;
 import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
 import org.openjdk.jmh.annotations.*;
@@ -13,7 +13,7 @@
 import java.util.Collections;
 import java.util.concurrent.TimeUnit;
 
-import static com.tomfran.lsm.TestUtils.getRandomItem;
+import static com.tomfran.lsm.TestUtils.getRandomPair;
 
 @OutputTimeUnit(TimeUnit.SECONDS)
 @State(Scope.Benchmark)
@@ -23,8 +23,8 @@ public class SSTableBenchmark {
     static final int NUM_ITEMS = 100000;
     static final int SAMPLE_SIZE = NUM_ITEMS / 1000;
 
-    static Item[] insertedArray;
-    static Item[] skippedArray;
+    static ByteArrayPair[] insertedArray;
+    static ByteArrayPair[] skippedArray;
     static SSTable sstable;
 
     static int index = 0;
@@ -38,18 +38,18 @@ public void setup() throws IOException {
         Files.createDirectory(DIR);
 
         // generate random items
-        var l = new ObjectOpenHashSet<Item>();
+        var l = new ObjectOpenHashSet<ByteArrayPair>();
         for (int i = 0; i < NUM_ITEMS * 2; i++) {
-            l.add(getRandomItem());
+            l.add(getRandomPair());
         }
 
         // sort and divide into inserted and skipped
         var items = l.stream()
                 .sorted((a, b) -> ByteArrayComparator.compare(a.key(), b.key()))
                 .toList();
 
-        var inserted = new ObjectArrayList<Item>();
-        var skipped = new ObjectArrayList<Item>();
+        var inserted = new ObjectArrayList<ByteArrayPair>();
+        var skipped = new ObjectArrayList<ByteArrayPair>();
 
         for (int i = 0; i < items.size(); i++) {
             var e = items.get(i);
@@ -64,8 +64,8 @@ public void setup() throws IOException {
         // shuffle to avoid sequential access
         Collections.shuffle(inserted);
         Collections.shuffle(skipped);
-        insertedArray = inserted.toArray(Item[]::new);
-        skippedArray = skipped.toArray(Item[]::new);
+        insertedArray = inserted.toArray(ByteArrayPair[]::new);
+        skippedArray = skipped.toArray(ByteArrayPair[]::new);
     }
 
     @TearDown

diff --git a/src/main/java/com/tomfran/lsm/io/BaseInputStream.java b/src/main/java/com/tomfran/lsm/io/BaseInputStream.java
@@ -1,5 +1,6 @@
 package com.tomfran.lsm.io;
 
+import com.tomfran.lsm.types.ByteArrayPair;
 import it.unimi.dsi.fastutil.io.FastBufferedInputStream;
 
 import java.io.FileInputStream;
@@ -67,6 +68,37 @@ public byte[] readNBytes(int n) {
         }
     }
 
+    public ByteArrayPair readBytePair() {
+        try {
+            int keyLength = readVByteInt();
+            int valueLength = readVByteInt();
+
+            return new ByteArrayPair(
+                    readNBytes(keyLength),
+                    readNBytes(valueLength)
+            );
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
+    public byte[] readByteArray() {
+        try {
+            int len = readVByteInt();
+            return fis.readNBytes(len);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public long skip(int n) {
+        try {
+            return fis.skip(n);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     public void seek(long offset) {
         try {
             fis.position(offset);

diff --git a/src/main/java/com/tomfran/lsm/io/BaseOutputStream.java b/src/main/java/com/tomfran/lsm/io/BaseOutputStream.java
@@ -1,5 +1,6 @@
 package com.tomfran.lsm.io;
 
+import com.tomfran.lsm.types.ByteArrayPair;
 import it.unimi.dsi.fastutil.io.FastBufferedOutputStream;
 
 import java.io.FileOutputStream;
@@ -39,6 +40,20 @@ public int writeLong(long n) {
         return write(longToBytes(n));
     }
 
+    public int writeBytePair(ByteArrayPair pair) {
+        byte[] key = pair.key(), value = pair.value();
+        byte[] keyBytes = intToVByte(key.length), valueBytes = intToVByte(value.length);
+
+        byte[] result = new byte[keyBytes.length + valueBytes.length + key.length + value.length];
+
+        System.arraycopy(keyBytes, 0, result, 0, keyBytes.length);
+        System.arraycopy(valueBytes, 0, result, keyBytes.length, valueBytes.length);
+
+        System.arraycopy(key, 0, result, keyBytes.length + valueBytes.length, key.length);
+        System.arraycopy(value, 0, result, keyBytes.length + valueBytes.length + key.length, value.length);
+        return write(result);
+    }
+
     byte[] intToVByte(int n) {
         return longToVByte(n);
     }

diff --git a/src/main/java/com/tomfran/lsm/io/ItemsInputStream.java b/src/main/java/com/tomfran/lsm/io/ItemsInputStream.java
diff --git a/src/main/java/com/tomfran/lsm/io/ItemsOutputStream.java b/src/main/java/com/tomfran/lsm/io/ItemsOutputStream.java
diff --git a/src/main/java/com/tomfran/lsm/memtable/Memtable.java b/src/main/java/com/tomfran/lsm/memtable/Memtable.java
@@ -1,7 +1,7 @@
 package com.tomfran.lsm.memtable;
 
 import com.tomfran.lsm.sstable.SSTable;
-import com.tomfran.lsm.types.Item;
+import com.tomfran.lsm.types.ByteArrayPair;
 
 public class Memtable {
 
@@ -17,16 +17,16 @@ public Memtable(int numElements) {
         list = new SkipList(numElements);
     }
 
-    public void add(Item item) {
+    public void add(ByteArrayPair item) {
         list.add(item);
     }
 
-    public Item get(byte[] key) {
+    public ByteArrayPair get(byte[] key) {
         return list.get(key);
     }
 
     public void remove(byte[] key) {
-        list.add(new Item(key, null));
+        list.add(new ByteArrayPair(key, null));
     }
 
     public int size() {

diff --git a/src/main/java/com/tomfran/lsm/memtable/SkipList.java b/src/main/java/com/tomfran/lsm/memtable/SkipList.java
@@ -1,6 +1,6 @@
 package com.tomfran.lsm.memtable;
 
-import com.tomfran.lsm.types.Item;
+import com.tomfran.lsm.types.ByteArrayPair;
 import it.unimi.dsi.util.XoRoShiRo128PlusRandom;
 
 import java.util.Iterator;
@@ -12,7 +12,7 @@
 /**
  * A skip list implementation of items.
  */
-public class SkipList implements Iterable<Item> {
+public class SkipList implements Iterable<ByteArrayPair> {
 
     static final int DEFAULT_ELEMENTS = 1 << 16;
 
@@ -49,7 +49,7 @@ public SkipList(int numElements) {
      *
      * @param item The item to add.
      */
-    public void add(Item item) {
+    public void add(ByteArrayPair item) {
         Node current = sentinel;
         for (int i = levels - 1; i >= 0; i--) {
             while (current.next[i] != null && current.next[i].value.compareTo(item) < 0)
@@ -83,7 +83,7 @@ private int randomLevel() {
      * @param key The key of the item to retrieve.
      * @return The item if found, null otherwise.
      */
-    public Item get(byte[] key) {
+    public ByteArrayPair get(byte[] key) {
         Node current = sentinel;
         for (int i = levels - 1; i >= 0; i--) {
             while (current.next[i] != null && compare(current.next[i].value.key(), key) < 0)
@@ -135,7 +135,7 @@ public int size() {
      * @return An iterator over the items in the skip list.
      */
     @Override
-    public Iterator<Item> iterator() {
+    public Iterator<ByteArrayPair> iterator() {
         return new SkipListIterator(sentinel);
     }
 
@@ -155,24 +155,24 @@ public String toString() {
     }
 
     private static final class Node {
-        Item value;
+        ByteArrayPair value;
         Node[] next;
 
-        Node(Item value, int numLevels) {
+        Node(ByteArrayPair value, int numLevels) {
             this.value = value;
             this.next = new Node[numLevels];
         }
     }
 
-    private record SkipListIterator(Node node) implements Iterator<Item> {
+    private record SkipListIterator(Node node) implements Iterator<ByteArrayPair> {
 
         @Override
         public boolean hasNext() {
             return node.next[0] != null;
         }
 
         @Override
-        public Item next() {
+        public ByteArrayPair next() {
             return node.next[0].value;
         }
     }