Use offset on single kv to avoid scanning the full file

src/main/kotlin/org/example/kv/IndexedKeyValueStore.kt

@@ -26,9 +26,7 @@ private class IndexedKeyValueStore<K, V>(
         }
     }
 
-    private fun getRaw(key: K, nullifyTombstone: Boolean): V? {
-
-        val offset = index.getOffset(key)
+    private fun getRaw(key: K, offset: Long?, nullifyTombstone: Boolean): V? {
 
         if (offset == tombstoneIndex) {
             return if (nullifyTombstone) null else tombstone
@@ -43,9 +41,17 @@ private class IndexedKeyValueStore<K, V>(
                 ?.value
     }
 
-    override fun getWithTombstone(key: K) = getRaw(key, false)
+    override fun getWithTombstone(key: K, offset: Long?) = getRaw(
+        key,
+        offset ?: index.getOffset(key),
+        false
+    )
 
-    override fun get(key: K) = getRaw(key, true)
+    override fun get(key: K, offset: Long?) = getRaw(
+        key,
+        offset ?: index.getOffset(key),
+        true
+    )
 
     override fun delete(key: K) {
         logKV.delete(key)

src/main/kotlin/org/example/kv/KeyValueStore.kt

@@ -23,8 +23,6 @@ interface LogBasedKeyValueStore<K, V>: TombstoneKeyValueStore<K, V> {
 
     fun appendAll(entries: Map<K, V>): Sequence<Long>
 
-    fun get(key: K, offset: Long): V?
-
     fun getWithOffset(key: K): ValueWithOffset<V>?
 
     fun <R> useEntries(block: (Sequence<Map.Entry<K, V>>) -> R): R = log.useEntries(0) { block(it) }
@@ -38,8 +36,11 @@ interface LogBasedKeyValueStore<K, V>: TombstoneKeyValueStore<K, V> {
 
 interface TombstoneKeyValueStore<K, V>: KeyValueStore<K, V> {
 
-    fun getWithTombstone(key: K): V?
+    fun getWithTombstone(key: K, offset: Long? = null): V?
+
+    fun get(key: K, offset: Long?): V?
 
+    override fun get(key: K) = get(key, null)
 }
 
 data class ValueWithOffset<V>(val offset: Long, val value: V)

src/main/kotlin/org/example/kv/SingleLogKeyValueStore.kt

@@ -32,13 +32,13 @@ private class SingleLogKeyValueStore<K, V>(override val log: Log<Map.Entry<K, V>
         }
     }
 
-    override fun get(key: K, offset: Long) = getWithTombstone(key)
+    override fun get(key: K, offset: Long?) = getWithTombstone(key, offset)
             ?.takeUnless { possiblyArrayEquals(it, tombstone) }
 
     override fun getWithOffset(key: K) = log.useEntriesWithOffset { it.findLastKey(key) }
             ?.takeUnless { possiblyArrayEquals(it.value, tombstone) }
 
-    override fun getWithTombstone(key: K): V? = log.useEntries(0) { it.findLastKey(key) }
+    override fun getWithTombstone(key: K, offset: Long?): V? = log.useEntries(offset ?: 0L) { it.findLastKey(key) }
 
     private fun Sequence<Map.Entry<K, V>>.findLastKey(key: K): V? = this
             .findLast { possiblyArrayEquals(key, it.key) }?.value

src/main/kotlin/org/example/kv/lsm/sequential/SequentialLogMergeStrategy.kt

@@ -1,6 +1,9 @@
-package org.example.kv.lsm
+package org.example.kv.lsm.sequential
 
 import mu.KotlinLogging
+import org.example.kv.lsm.Segment
+import org.example.kv.lsm.SegmentFactory
+import org.example.kv.lsm.SegmentMergeStrategy
 import org.example.size.SizeCalculator
 import java.util.*
 

src/main/kotlin/org/example/kv/lsm/sequential/SequentialOpenSegment.kt

@@ -1,22 +1,14 @@
-package org.example.kv.lsm
+package org.example.kv.lsm.sequential
 
 import org.example.kv.LogBasedKeyValueStore
 import org.example.kv.LogBasedKeyValueStoreFactory
+import org.example.kv.TombstoneKeyValueStore
+import org.example.kv.lsm.*
 import org.example.log.LogFactory
 
 private class SequentialOpenSegment<K, V>(private val keyValueStore: LogBasedKeyValueStore<K, V>,
                                           private val segmentThreshold: Long = 1024 * 1024):
-    OpenSegment<K, V> {
-
-    override fun put(key: K, value: V) = keyValueStore.put(key, value)
-
-    override fun getWithTombstone(key: K): V? = keyValueStore.getWithTombstone(key)
-
-    override fun get(key: K): V? = keyValueStore.get(key)
-
-    override fun delete(key: K) = keyValueStore.delete(key)
-
-    override fun clear() = keyValueStore.clear()
+    OpenSegment<K, V>, TombstoneKeyValueStore<K, V> by keyValueStore {
 
     override fun closeSegment(): Segment<K, V> = Segment(keyValueStore, segmentThreshold)
 
@@ -27,7 +19,8 @@ class SequentialSegmentManager<K, V>(private val segmentDirectory: SegmentDirect
                                      private val logFactory: LogFactory<Map.Entry<K, V>>,
                                      private val keyValueStoreFactory: LogBasedKeyValueStoreFactory<K, V>,
                                      segmentThreshold: Long,
-                                     mergeStrategy: SequentialLogMergeStrategy<K, V>):
+                                     mergeStrategy: SequentialLogMergeStrategy<K, V>
+):
     SegmentManager<K, V>(segmentDirectory, logFactory, keyValueStoreFactory, mergeStrategy, segmentThreshold) {
 
     override fun createOpenSegment(): OpenSegment<K, V> = segmentDirectory

src/main/kotlin/org/example/kv/lsm/sstable/SortedStringTable.kt

@@ -32,8 +32,9 @@ private class SortedMapStringTable<K: Comparable<K>, V>(private val memTable: So
         }
     }
 
-    override fun get(key: K): V? = when(dumped) {
-        true -> dump.get(key)
+    override fun get(key: K, offset: Long?): V? = when(dumped) {
+        true -> dump.get(key, offset)
+        // offset is ignored here. Is this problematic?
         false -> memTable[key]
     }
 
@@ -69,7 +70,7 @@ private class SortedMapStringTable<K: Comparable<K>, V>(private val memTable: So
         }
     }
 
-    override fun getWithTombstone(key: K): V? = get(key)
+    override fun getWithTombstone(key: K, offset: Long?): V? = get(key)
 
     override fun isFull(): Boolean = size >= segmentThreshold
 

src/test/kotlin/org/example/kv/KeyValueStores.kt

@@ -9,6 +9,8 @@ import org.example.TestInstance
 import org.example.TestResources
 import org.example.encoder.Encoders
 import org.example.kv.lsm.*
+import org.example.kv.lsm.sequential.SequentialLogMergeStrategy
+import org.example.kv.lsm.sequential.SequentialSegmentManager
 import org.example.kv.lsm.sstable.SSTableMergeStrategy
 import org.example.kv.lsm.sstable.SSTableSegmentManager
 import org.example.log.LogEncoderFactory