Refrain from using ByteArray as key in tests

src/main/kotlin/org/example/index/CheckpointableIndex.kt

@@ -1,13 +1,15 @@
 package org.example.index
 
 import kotlinx.coroutines.CoroutineDispatcher
+import mu.KotlinLogging
 import org.example.encoder.Encoder
 import org.example.log.Log
 import org.example.log.LogEncoderFactory
 import org.example.log.LogFactory
 import org.example.recurrent.OpsBasedRecurrentJob
 import org.example.recurrent.RecurrentJob
 import java.nio.file.Files.createFile
+import java.nio.file.Files.notExists
 import java.nio.file.Path
 
 interface CheckpointableIndex<K>: Index<K> {
@@ -49,12 +51,29 @@ class CheckpointableIndexFactory<K>(private val innerFactory: IndexFactory<K>,
                                     private val checkpointCycle: Long,
                                     private val coroutineDispatcher: CoroutineDispatcher): IndexFactory<K> {
 
-    override fun create(indexName: String): Index<K> = "index-$indexName.log"
-        .let { indexDir.resolve(it) }
-        .let { createFile(it) }
-        .let { entryLogFactory.create(it) }
-        .let { LogCheckpointableIndex(innerFactory.create(indexName), it) }
-        .let { RecurrentCheckpointableIndex(it, createRecurringCheckpoint(it)) }
+    private val logger = KotlinLogging.logger {}
+
+    override fun create(indexName: String): CheckpointableIndex<K> {
+
+        val indexPath = "index-$indexName.log"
+            .let { indexDir.resolve(it) }
+
+        val innerIndex = innerFactory.create(indexName)
+
+        val indexLog: Log<IndexEntry<K>>
+        if (notExists(indexPath)) {
+            logger.debug { "Creating index file." }
+            createFile(indexPath)
+            indexLog = entryLogFactory.create(indexPath)
+        } else {
+            logger.debug { "Index file already exists. Recovering index data." }
+            indexLog = entryLogFactory.create(indexPath)
+            indexLog.useEntries { innerIndex.putAllOffsets(it.asIterable()) }
+        }
+
+        return LogCheckpointableIndex(innerIndex, indexLog)
+            .let { RecurrentCheckpointableIndex(it, createRecurringCheckpoint(it)) }
+    }
 
     private fun createRecurringCheckpoint(index: CheckpointableIndex<K>): RecurrentJob {
 

src/main/kotlin/org/example/index/HashIndex.kt

@@ -6,7 +6,9 @@ private class HashIndex<K>(private val index: MutableMap<K, Long> = HashMap()):
         index[key] = offset
     }
 
-    override fun getOffset(key: K) = index[key]
+    override fun getOffset(key: K): Long? {
+        return index[key]
+    }
 
     override fun putAllOffsets(pairs: Iterable<IndexEntry<K>>) {
         pairs.forEach { index[it.key] = it.offset }

src/main/kotlin/org/example/size/ByteArraySizeCalculator.kt

@@ -8,6 +8,12 @@ object ByteArraySizeCalculator: SizeCalculator<ByteArray> {
 
 }
 
+object LongSizeCalculator: SizeCalculator<Long> {
+
+    override fun sizeOf(value: Long) = Long.SIZE_BYTES
+
+}
+
 class StringSizeCalculator(private val charset: Charset,
                            private val byteArraySizeCalculator: SizeCalculator<ByteArray>): SizeCalculator<String> {
 

src/test/kotlin/org/example/index/CheckpointableIndexTest.kt

@@ -7,24 +7,47 @@ import org.example.TestInstance
 import org.example.TestResources
 import org.example.encoder.Encoders
 import org.example.log.LogFactories
+import org.example.test
 import org.junit.jupiter.api.AfterAll
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.TestFactory
 import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicLong
 
-internal abstract class AbstractLogCheckpointableIndexTest<K>: IndexTest<K> {
+internal abstract class CheckpointableIndexTest<K>: IndexTest<K> {
 
     internal val uniqueGenerator = AtomicLong()
 
+    abstract override fun instances(): Sequence<TestInstance<CheckpointableIndex<K>>>
+
+    abstract fun factories(): Sequence<TestInstance<CheckpointableIndexFactory<K>>>
+
+    @TestFactory fun `checkpointable index should be recoverable`() = factories().test { factory ->
+
+        val expected = (1..100).map { IndexEntry(nextKey(), it.toLong()) }
+
+        val indexName = "2bRecovered-${nextKey()}"
+        val index = factory.create(indexName)
+
+        index.putAllOffsets(expected)
+        index.checkpoint()
+
+        // should recover the index from file
+        val recoveredIndexed = factory.create(indexName)
+
+        expected.forEach { assertEquals(it.offset, recoveredIndexed.getOffset(it.key)) }
+    }
+
     companion object {
 
-        @JvmStatic
         internal val resources = TestResources()
 
         @JvmStatic
-        internal val dispatcher= Executors.newSingleThreadExecutor().asCoroutineDispatcher()
+        internal val dispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
 
         @JvmStatic
-        internal val indexes = Indexes(LogFactories(Encoders()), resources, dispatcher)
+        internal val indexes = CheckpointableIndexes(resources, LogFactories(Encoders()), dispatcher)
 
         @JvmStatic
         @AfterAll
@@ -36,22 +59,29 @@ internal abstract class AbstractLogCheckpointableIndexTest<K>: IndexTest<K> {
 
 }
 
-internal class CheckpointableStringIndexTest: AbstractLogCheckpointableIndexTest<String>() {
+internal class CheckpointableStringIndexTest: CheckpointableIndexTest<String>() {
 
     @ExperimentalSerializationApi
-    override fun instances(): Sequence<TestInstance<Index<String>>> = indexes.instances(serializer())
+    override fun instances(): Sequence<TestInstance<CheckpointableIndex<String>>> = indexes.indexes(serializer())
 
     override fun nextKey() = uniqueGenerator.getAndIncrement().toString()
 
+    @ExperimentalSerializationApi
+    override fun factories(): Sequence<TestInstance<CheckpointableIndexFactory<String>>> = indexes
+        .comparableIndexFactories(serializer())
+
 }
 
-internal class CheckpointableBinaryIndexTest: AbstractLogCheckpointableIndexTest<ByteArray>() {
+internal class CheckpointableBinaryLongTest: CheckpointableIndexTest<Long>() {
 
     @ExperimentalSerializationApi
-    override fun instances(): Sequence<TestInstance<Index<ByteArray>>> = indexes.nonComparableInstances(serializer())
+    override fun instances(): Sequence<TestInstance<CheckpointableIndex<Long>>> = indexes
+        .nonComparableIndexes(serializer())
 
     override fun nextKey() = uniqueGenerator.getAndIncrement()
-            .toString()
-            .toByteArray()
+
+    @ExperimentalSerializationApi
+    override fun factories(): Sequence<TestInstance<CheckpointableIndexFactory<Long>>> = indexes
+        .nonComparableIndexFactories(serializer())
 
 }

src/test/kotlin/org/example/index/CheckpointableIndexes.kt

@@ -0,0 +1,85 @@
+package org.example.index
+
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.KSerializer
+import org.example.TestInstance
+import org.example.TestResources
+import org.example.encoder.JsonStringEncoder
+import org.example.encoder.ProtobufBinaryEncoder
+import org.example.log.LogFactories
+import java.util.concurrent.atomic.AtomicLong
+
+class CheckpointableIndexes(private val resources: TestResources,
+                            private val logs: LogFactories,
+                            private val dispatcher: CoroutineDispatcher) {
+
+    private val generator = AtomicLong()
+
+    @ExperimentalSerializationApi
+    fun <K : Comparable<K>> indexes(serializer: KSerializer<IndexEntry<K>>)
+            : Sequence<TestInstance<CheckpointableIndex<K>>>
+            = comparableIndexes(serializer) + nonComparableIndexes(serializer)
+
+    @ExperimentalSerializationApi
+    fun <K: Comparable<K>> comparableIndexes(serializer: KSerializer<IndexEntry<K>>):
+            Sequence<TestInstance<CheckpointableIndex<K>>> = indexes(TreeIndexFactory(), serializer)
+
+    @ExperimentalSerializationApi
+    fun <K> nonComparableIndexes(serializer: KSerializer<IndexEntry<K>>)
+            : Sequence<TestInstance<CheckpointableIndex<K>>> = indexes(HashIndexFactory(), serializer)
+
+    @ExperimentalSerializationApi
+    private fun <K> indexes(innerIndexFactory: IndexFactory<K>, serializer: KSerializer<IndexEntry<K>>):
+            Sequence<TestInstance<CheckpointableIndex<K>>> = indexFactories(innerIndexFactory, serializer)
+        .map { case ->
+            TestInstance(case.name) {
+                case.instance().create("CheckpointableIndex${generator.getAndIncrement()}")
+            }
+        }
+
+    @ExperimentalSerializationApi
+    fun <K> nonComparableIndexFactories(serializer: KSerializer<IndexEntry<K>>) = indexFactories(
+        HashIndexFactory(),
+        serializer
+    )
+
+    @ExperimentalSerializationApi
+    fun <K: Comparable<K>> comparableIndexFactories(serializer: KSerializer<IndexEntry<K>>) = indexFactories(
+        TreeIndexFactory(),
+        serializer
+    )
+
+    @ExperimentalSerializationApi
+    private fun <K> indexFactories(innerIndexFactory: IndexFactory<K>, serializer: KSerializer<IndexEntry<K>>)
+    : Sequence<TestInstance<CheckpointableIndexFactory<K>>> = sequence {
+
+        for (lineLogInstance in logs.lineLogInstances()) {
+
+            yield(TestInstance("Checkpointable String Index ~ ${lineLogInstance.name}") {
+                val indexDir = resources.allocateTempDir("index-string-")
+                CheckpointableIndexFactory(
+                    innerIndexFactory,
+                    indexDir,
+                    IndexEntryLogFactory(lineLogInstance.instance(), JsonStringEncoder(serializer)),
+                    10_000,
+                    dispatcher
+                )
+            })
+        }
+
+        for (binaryInstance in logs.binaryInstances()) {
+
+            yield(TestInstance("Checkpointable Binary Index ~ ${binaryInstance.name}") {
+                val indexDir = resources.allocateTempDir("index-binary-")
+                CheckpointableIndexFactory(
+                    innerIndexFactory,
+                    indexDir,
+                    IndexEntryLogFactory(binaryInstance.instance(), ProtobufBinaryEncoder(serializer)),
+                    10_000,
+                    dispatcher
+                )
+            })
+        }
+    }
+}

src/test/kotlin/org/example/index/HashIndexTest.kt

@@ -0,0 +1,30 @@
+package org.example.index
+
+import org.example.TestInstance
+import java.util.concurrent.atomic.AtomicLong
+
+internal abstract class HashIndexTest<K>: IndexTest<K> {
+
+    internal val uniqueGenerator = AtomicLong()
+
+    override fun instances(): Sequence<TestInstance<Index<K>>> = indexes.hashIndexes()
+
+    companion object {
+
+        @JvmStatic
+        internal val indexes = Indexes()
+    }
+
+}
+
+internal class LongHashIndexTest: HashIndexTest<Long>() {
+
+    override fun nextKey() = uniqueGenerator.getAndIncrement()
+
+}
+
+internal class StringHashIndexTest: HashIndexTest<String>() {
+
+    override fun nextKey() = uniqueGenerator.getAndIncrement().toString()
+
+}

src/test/kotlin/org/example/index/Indexes.kt

@@ -1,93 +1,31 @@
 package org.example.index
 
-import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.serialization.ExperimentalSerializationApi
-import kotlinx.serialization.KSerializer
 import org.example.TestInstance
-import org.example.TestResources
-import org.example.encoder.JsonStringEncoder
-import org.example.encoder.ProtobufBinaryEncoder
-import org.example.log.LogFactories
 import java.util.concurrent.atomic.AtomicLong
 
-class Indexes(private val logs: LogFactories,
-              private val resources: TestResources,
-              private val dispatcher: CoroutineDispatcher) {
+class Indexes {
 
     private val generator = AtomicLong()
 
-    @ExperimentalSerializationApi
-    fun <K : Comparable<K>> instances(serializer: KSerializer<IndexEntry<K>>): Sequence<TestInstance<Index<K>>>
-            = comparableInstances(serializer) + nonComparableInstances(serializer)
-
-    @ExperimentalSerializationApi
-    fun <K> nonComparableInstances(serializer: KSerializer<IndexEntry<K>>): Sequence<TestInstance<Index<K>>> {
+    fun <K> hashIndexes(): Sequence<TestInstance<Index<K>>> {
 
         val factory = HashIndexFactory<K>()
 
         val volatileHash = TestInstance("Hash Index") {
             factory.create("HashIndex${generator.getAndIncrement()}")
         }
 
-        return sequenceOf(volatileHash) + checkpointableNonComparableIndexes(serializer)
+        return sequenceOf(volatileHash)
     }
 
-    @ExperimentalSerializationApi
-    private fun <K : Comparable<K>> comparableInstances(serializer: KSerializer<IndexEntry<K>>):
-            Sequence<TestInstance<Index<K>>> {
+    fun <K: Comparable<K>> treeIndexes(): Sequence<TestInstance<Index<K>>> {
 
         val factory = TreeIndexFactory<K>()
-        val volatileTree = TestInstance("Tree Index") {
+        val transientTree = TestInstance("Tree Index") {
             factory.create("TreeIndex${generator.getAndIncrement()}")
         }
 
-        return sequenceOf(volatileTree) + checkpointableComparableIndexes(serializer)
+        return sequenceOf(transientTree)
     }
 
-    @ExperimentalSerializationApi
-    private fun <K: Comparable<K>> checkpointableComparableIndexes(serializer: KSerializer<IndexEntry<K>>):
-            Sequence<TestInstance<Index<K>>> = checkpointableIndexes(TreeIndexFactory(), serializer)
-
-    @ExperimentalSerializationApi
-    fun <K> checkpointableNonComparableIndexes(serializer: KSerializer<IndexEntry<K>>): Sequence<TestInstance<Index<K>>>
-            = checkpointableIndexes(HashIndexFactory(), serializer)
-
-    @ExperimentalSerializationApi
-    private fun <K> checkpointableIndexes(innerIndexFactory: IndexFactory<K>, serializer: KSerializer<IndexEntry<K>>):
-            Sequence<TestInstance<Index<K>>> =
-        sequence {
-
-            val indexDir = resources.allocateTempDir("index-")
-
-            for (lineLogInstance in logs.lineLogInstances()) {
-                val stringFactory = CheckpointableIndexFactory(
-                    innerIndexFactory,
-                    indexDir,
-                    IndexEntryLogFactory(lineLogInstance.instance(), JsonStringEncoder(serializer)),
-                    10_000,
-                    dispatcher
-                )
-
-                yield(TestInstance("Checkpointable String Index") {
-                    stringFactory.create("CheckpointableIndex${generator.getAndIncrement()} ~ ${lineLogInstance
-                        .name}")
-                })
-            }
-
-            for (binaryInstance in logs.binaryInstances()) {
-                val binaryFactory = CheckpointableIndexFactory(
-                    innerIndexFactory,
-                    indexDir,
-                    IndexEntryLogFactory(binaryInstance.instance(), ProtobufBinaryEncoder(serializer)),
-                    10_000,
-                    dispatcher
-                )
-
-                yield(TestInstance("Checkpointable Binary Index") {
-                    binaryFactory.create("CheckpointableIndex${generator.getAndIncrement()} ~ ${binaryInstance
-                        .name}")
-                })
-            }
-        }
-
 }