Do not create new write actions for RocksDB if values exist in both t…

…he new as old version.
marykdb · Sep 28, 2024 · 370395d · 370395d
1 parent 306f875
commit 370395d
Show file tree

Hide file tree

Showing 4 changed files with 177 additions and 74 deletions.
diff --git a/...ksdb/src/commonMain/kotlin/maryk/datastore/rocksdb/processors/helpers/getCurrentValues.kt b/...ksdb/src/commonMain/kotlin/maryk/datastore/rocksdb/processors/helpers/getCurrentValues.kt
@@ -0,0 +1,27 @@
+package maryk.datastore.rocksdb.processors.helpers
+
+import maryk.core.properties.types.Key
+import maryk.datastore.rocksdb.TableColumnFamilies
+import maryk.datastore.rocksdb.Transaction
+import maryk.lib.extensions.compare.compareDefinedTo
+import org.rocksdb.ReadOptions
+
+internal fun getCurrentValues(
+    transaction: Transaction,
+    columnFamilies: TableColumnFamilies,
+    key: Key<*>,
+    referenceAsBytes: ByteArray
+): List<Pair<ByteArray, ByteArray>> {
+    val prefix = key.bytes + referenceAsBytes
+    val currentValues = mutableListOf<Pair<ByteArray, ByteArray>>()
+
+    val iterator = transaction.getIterator(ReadOptions(), columnFamilies.table)
+    iterator.seek(prefix)
+
+    while (iterator.isValid() && prefix.compareDefinedTo(iterator.key()) == 0) {
+        currentValues.add(iterator.key() to iterator.value())
+        iterator.next()
+    }
+
+    return currentValues
+}
diff --git a/...rocksdb/src/commonMain/kotlin/maryk/datastore/rocksdb/processors/helpers/setTypedValue.kt b/...rocksdb/src/commonMain/kotlin/maryk/datastore/rocksdb/processors/helpers/setTypedValue.kt
@@ -20,7 +20,9 @@ internal fun setTypedValue(
     columnFamilies: TableColumnFamilies,
     key: Key<*>,
     reference: ByteArray,
-    versionBytes: ByteArray
+    versionBytes: ByteArray,
+    qualifiersToKeep: MutableList<ByteArray>? = null,
+    shouldWrite: ((referenceBytes: ByteArray, valueBytes: ByteArray) -> Boolean)? = null,
 ) {
     val properValue = if (value is MultiTypeEnum<*>) {
         TypedValue(value, Unit)
@@ -31,11 +33,14 @@ internal fun setTypedValue(
     val typeDefinition = TypeValue.castDefinition(definition)
 
     var index = 0
+    qualifiersToKeep?.add(reference)
     if (typedValue.value == Unit) {
         val valueBytes = ByteArray(typedValue.type.index.calculateVarIntWithExtraInfoByteSize())
         typedValue.type.index.writeVarIntWithExtraInfo(TypeIndicator.ComplexTypeIndicator.byte) { valueBytes[index++] = it }
 
-        setValue(transaction, columnFamilies, key, reference, versionBytes, valueBytes)
+        if (shouldWrite?.invoke(reference, valueBytes) != false) {
+            setValue(transaction, columnFamilies, key, reference, versionBytes, valueBytes)
+        }
     } else {
         val typeValueDefinition = typeDefinition.definition(typedValue.type) as IsSimpleValueDefinition<Any, *>
         val valueBytes = ByteArray(
@@ -47,13 +52,15 @@ internal fun setTypedValue(
         typedValue.type.index.writeVarIntWithExtraInfo(TypeIndicator.SimpleTypeIndicator.byte, writer)
         typeValueDefinition.writeStorageBytes(typedValue.value, writer)
 
-        setValue(
-            transaction,
-            columnFamilies,
-            key,
-            reference,
-            versionBytes,
-            valueBytes
-        )
+        if (shouldWrite?.invoke(reference, valueBytes) != false) {
+            setValue(
+                transaction,
+                columnFamilies,
+                key,
+                reference,
+                versionBytes,
+                valueBytes
+            )
+        }
     }
 }
diff --git a/...src/commonMain/kotlin/maryk/datastore/rocksdb/processors/helpers/unsetNonChangedValues.kt b/...src/commonMain/kotlin/maryk/datastore/rocksdb/processors/helpers/unsetNonChangedValues.kt
@@ -0,0 +1,33 @@
+package maryk.datastore.rocksdb.processors.helpers
+
+import maryk.core.properties.types.Key
+import maryk.datastore.rocksdb.TableColumnFamilies
+import maryk.datastore.rocksdb.Transaction
+import maryk.datastore.shared.TypeIndicator
+import maryk.lib.extensions.compare.compareTo
+import maryk.lib.extensions.compare.compareToWithOffsetLength
+
+internal fun unsetNonChangedValues(
+    transaction: Transaction,
+    columnFamilies: TableColumnFamilies,
+    key: Key<*>,
+    currentValues: List<Pair<ByteArray, ByteArray>>,
+    qualifiersToKeep: List<ByteArray>,
+    versionBytes: ByteArray
+) {
+    val sortedQualifiersToKeep = qualifiersToKeep.sortedWith { o1, o2 -> o1.compareTo(o2) }
+    var minIndex = 0
+
+    for ((qualifier, _) in currentValues) {
+        val index = sortedQualifiersToKeep.binarySearch(fromIndex = minIndex) {
+            it.compareToWithOffsetLength(qualifier, key.bytes.size, qualifier.size - key.bytes.size)
+        }
+        if (index < 0) {
+            // Delete the value by setting it to the DeletedIndicator
+            setValue(transaction, columnFamilies, qualifier, versionBytes, TypeIndicator.DeletedIndicator.byteArray)
+        } else {
+            // Start next time comparing with next value in qualifiersToKeep as they are ordered
+            minIndex = index + 1
+        }
+    }
+}