use TupleOrNone instead of Option[(L, R)]

core/src/main/scala/swaydb/core/segment/entry/writer/DeadlineWriter.scala

@@ -17,11 +17,12 @@
 package swaydb.core.segment.entry.writer
 
 import swaydb.core.segment.data.Memory
-import swaydb.core.segment.entry.id.BaseEntryId.DeadlineId
 import swaydb.core.segment.entry.id.{BaseEntryId, MemoryToKeyValueIdBinder}
+import swaydb.core.segment.entry.id.BaseEntryId.DeadlineId
 import swaydb.core.util.Bytes
 import swaydb.core.util.Times._
 import swaydb.utils.Options.when
+import swaydb.utils.{Options, TupleOrNone}
 
 import scala.concurrent.duration.Deadline
 
@@ -116,8 +117,11 @@ private[segment] object DeadlineWriter extends DeadlineWriter {
       previous = previousDeadline.toBytes,
       next = currentDeadline.toBytes,
       minimumCommonBytes = 1
-    ) map {
-      case (deadlineCommonBytes, deadlineCompressedBytes) =>
+    ) match {
+      case TupleOrNone.None =>
+        None
+
+      case TupleOrNone.Some(deadlineCommonBytes, deadlineCompressedBytes) =>
         val deadline = applyDeadlineId(deadlineCommonBytes, deadlineId)
 
         builder.setSegmentHasPrefixCompression()
@@ -129,6 +133,7 @@ private[segment] object DeadlineWriter extends DeadlineWriter {
         )
 
         builder.bytes addAll deadlineCompressedBytes
+        Options.unit
     }
 
   private[segment] def noDeadline[T <: Memory](current: T,

core/src/main/scala/swaydb/core/segment/entry/writer/KeyWriter.scala

@@ -20,6 +20,7 @@ import swaydb.core.segment.data.Memory
 import swaydb.core.segment.entry.id.{BaseEntryId, MemoryToKeyValueIdBinder}
 import swaydb.core.util.Bytes
 import swaydb.slice.Slice
+import swaydb.utils.{Options, TupleOrNone}
 
 trait KeyWriter {
   def write[T <: Memory](current: T,
@@ -58,8 +59,11 @@ private[segment] object KeyWriter extends KeyWriter {
                                            builder: EntryWriter.Builder,
                                            deadlineId: BaseEntryId.Deadline,
                                            previous: Memory)(implicit binder: MemoryToKeyValueIdBinder[T]): Option[Unit] =
-    Bytes.compress(previous = previous.key, next = current.mergedKey, minimumCommonBytes = 3) map {
-      case (commonBytes, remainingBytes) =>
+    Bytes.compress(previous = previous.key, next = current.mergedKey, minimumCommonBytes = 3) match {
+      case TupleOrNone.None =>
+        None
+
+      case TupleOrNone.Some(commonBytes, remainingBytes) =>
         write(
           current = current,
           builder = builder,
@@ -68,6 +72,8 @@ private[segment] object KeyWriter extends KeyWriter {
           deadlineId = deadlineId,
           isKeyCompressed = true
         )
+
+        Options.unit
     }
 
   private def writeUncompressed[T <: Memory](current: T,

core/src/main/scala/swaydb/core/segment/entry/writer/TimeWriter.scala

@@ -20,6 +20,7 @@ import swaydb.core.segment.data.{Memory, Time}
 import swaydb.core.segment.entry.id.{BaseEntryId, MemoryToKeyValueIdBinder}
 import swaydb.core.util.Bytes._
 import swaydb.utils.Options.when
+import swaydb.utils.{Options, TupleOrNone}
 
 private[segment] trait TimeWriter {
   private[segment] def write[T <: Memory](current: T,
@@ -97,9 +98,11 @@ private[segment] object TimeWriter extends TimeWriter {
       previous = previousTime.time,
       next = current.persistentTime.time,
       minimumCommonBytes = 3 //minimum 3 required because commonBytes & uncompressedByteSize requires 2 bytes.
-    ) map {
-      case (commonBytes, remainingBytes) =>
+    ) match {
+      case TupleOrNone.None =>
+        None
 
+      case TupleOrNone.Some(commonBytes, remainingBytes) =>
         builder.setSegmentHasPrefixCompression()
 
         valueWriter.write(
@@ -113,6 +116,8 @@ private[segment] object TimeWriter extends TimeWriter {
           .addUnsignedInt(commonBytes)
           .addUnsignedInt(remainingBytes.size)
           .addAll(remainingBytes)
+
+        Options.unit
     }
 
   private def writeUncompressed[T <: Memory](current: T,

core/src/main/scala/swaydb/core/segment/entry/writer/ValueWriter.scala

@@ -20,7 +20,7 @@ import swaydb.core.segment.data.Memory
 import swaydb.core.segment.entry.id.{BaseEntryId, MemoryToKeyValueIdBinder}
 import swaydb.core.util.Bytes
 import swaydb.slice.{Slice, SliceOption}
-import swaydb.utils.Options
+import swaydb.utils.{Options, TupleOrNone}
 
 private[segment] trait ValueWriter {
   def write[T <: Memory](current: T,
@@ -237,8 +237,11 @@ private[segment] object ValueWriter extends ValueWriter {
                                                                            keyWriter: KeyWriter,
                                                                            deadlineWriter: DeadlineWriter): Option[Unit] = {
     val currentValueOffset = builder.nextStartValueOffset
-    Bytes.compress(Slice.writeInt[Byte](builder.startValueOffset), Slice.writeInt[Byte](currentValueOffset), 1) map {
-      case (valueOffsetCommonBytes, valueOffsetRemainingBytes) =>
+    Bytes.compress(Slice.writeInt[Byte](builder.startValueOffset), Slice.writeInt[Byte](currentValueOffset), 1) match {
+      case TupleOrNone.None =>
+        None
+
+      case TupleOrNone.Some(valueOffsetCommonBytes, valueOffsetRemainingBytes) =>
         val valueOffsetId =
           if (valueOffsetCommonBytes == 1)
             entryId.valueUncompressed.valueOffsetOneCompressed
@@ -250,7 +253,7 @@ private[segment] object ValueWriter extends ValueWriter {
             throw new Exception(s"Fatal exception: valueOffsetCommonBytes = $valueOffsetCommonBytes")
 
         Bytes.compress(Slice.writeInt[Byte](previousValue.size), Slice.writeInt[Byte](currentValue.size), 1) match {
-          case Some((valueLengthCommonBytes, valueLengthRemainingBytes)) =>
+          case TupleOrNone.Some(valueLengthCommonBytes, valueLengthRemainingBytes) =>
             val valueLengthId =
               if (valueLengthCommonBytes == 1)
                 valueOffsetId.valueLengthOneCompressed
@@ -278,7 +281,9 @@ private[segment] object ValueWriter extends ValueWriter {
               .addAll(valueOffsetRemainingBytes)
               .addAll(valueLengthRemainingBytes)
 
-          case None =>
+            Options.unit
+
+          case TupleOrNone.None =>
             //if unable to compress valueLengthBytes then write compressed valueOffset with fully valueLength bytes.
             val currentUnsignedValueLengthBytes = Slice.writeUnsignedInt[Byte](currentValue.size)
 
@@ -296,6 +301,8 @@ private[segment] object ValueWriter extends ValueWriter {
               .bytes
               .addAll(valueOffsetRemainingBytes)
               .addAll(currentUnsignedValueLengthBytes)
+
+            Options.unit
         }
     }
   }
@@ -309,7 +316,7 @@ private[segment] object ValueWriter extends ValueWriter {
                                                                              keyWriter: KeyWriter,
                                                                              deadlineWriter: DeadlineWriter): Unit =
     Bytes.compress(previous = Slice.writeInt[Byte](previousValue.size), next = Slice.writeInt[Byte](currentValue.size), minimumCommonBytes = 1) match {
-      case Some((valueLengthCommonBytes, valueLengthRemainingBytes)) =>
+      case TupleOrNone.Some(valueLengthCommonBytes, valueLengthRemainingBytes) =>
         val valueLengthId =
           if (valueLengthCommonBytes == 1)
             entryId.valueUncompressed.valueOffsetUncompressed.valueLengthOneCompressed
@@ -339,7 +346,7 @@ private[segment] object ValueWriter extends ValueWriter {
           .addUnsignedInt(currentValueOffset)
           .addAll(valueLengthRemainingBytes)
 
-      case None =>
+      case TupleOrNone.None =>
         //unable to compress valueOffset and valueLength bytes, write them as full bytes.
         val currentValueOffset = builder.nextStartValueOffset
 

core/src/main/scala/swaydb/core/util/Bytes.scala

@@ -17,8 +17,9 @@
 package swaydb.core.util
 
 import swaydb.OK
-import swaydb.slice.utils.ScalaByteOps
 import swaydb.slice.{Slice, SliceReader}
+import swaydb.slice.utils.ScalaByteOps
+import swaydb.utils.TupleOrNone
 
 private[swaydb] object Bytes extends ScalaByteOps {
 
@@ -45,32 +46,38 @@ private[swaydb] object Bytes extends ScalaByteOps {
 
   def compress(previous: Slice[Byte],
                next: Slice[Byte],
-               minimumCommonBytes: Int): Option[(Int, Slice[Byte])] = {
+               minimumCommonBytes: Int): TupleOrNone[Int, Slice[Byte]] = {
     val commonBytes = Bytes.commonPrefixBytesCount(previous, next)
     if (commonBytes < minimumCommonBytes)
-      None
+      TupleOrNone.None
     else
-      Some(commonBytes, next.drop(commonBytes))
+      TupleOrNone.Some(commonBytes, next.drop(commonBytes))
   }
 
   def compressFull(previous: Option[Slice[Byte]],
                    next: Slice[Byte]): Option[OK] =
-    previous flatMap {
-      previous =>
+    previous match {
+      case Some(previous) =>
         compressFull(
           previous = previous,
           next = next
         )
+
+      case None =>
+        None
     }
 
   def compressFull(previous: Slice[Byte],
                    next: Slice[Byte]): Option[OK] =
     if (previous.size < next.size)
       None
     else
-      compress(previous, next, next.size) map {
-        _ =>
-          OK.instance
+      compress(previous, next, next.size) match {
+        case TupleOrNone.None =>
+          None
+
+        case TupleOrNone.Some(_, _) =>
+          OK.someOK
       }
 
   def compressExact(previous: Slice[Byte],

core/src/test/scala/swaydb/core/util/BytesSpec.scala

@@ -37,7 +37,7 @@ class BytesSpec extends AnyWordSpec with Matchers {
       val previous: Slice[Byte] = Slice(Array(1.toByte, 2.toByte, 3.toByte, 4.toByte))
       val next: Slice[Byte] = Slice(Array(1.toByte, 2.toByte, 3.toByte, 4.toByte, 5.toByte, 6.toByte))
 
-      val (commonBytes, compressed) = Bytes.compress(previous, next, 1).value
+      val (commonBytes, compressed) = Bytes.compress(previous, next, 1).toOption().value
 
       compressed shouldBe Slice(5, 6)
       commonBytes shouldBe 4
@@ -47,7 +47,7 @@ class BytesSpec extends AnyWordSpec with Matchers {
       decompress.isFull shouldBe true
       decompress shouldBe next
       //return empty if minimum compressed bytes is not reached
-      Bytes.compress(previous, next, 5) shouldBe empty
+      Bytes.compress(previous, next, 5).toOption() shouldBe empty
 
       Bytes.commonPrefixBytes(previous, next) shouldBe previous
     }
@@ -56,7 +56,7 @@ class BytesSpec extends AnyWordSpec with Matchers {
       val previous: Slice[Byte] = Slice(Array(1.toByte, 2.toByte, 3.toByte, 4.toByte))
       val next: Slice[Byte] = Slice(Array(1.toByte, 2.toByte, 3.toByte))
 
-      val (commonBytes, compressed) = Bytes.compress(previous, next, 1).value
+      val (commonBytes, compressed) = Bytes.compress(previous, next, 1).toOption().value
 
       compressed shouldBe Slice.emptyBytes
       commonBytes shouldBe 3
@@ -66,7 +66,7 @@ class BytesSpec extends AnyWordSpec with Matchers {
       decompress.isFull shouldBe true
       decompress shouldBe next
       //return empty if minimum compressed bytes is not reached
-      Bytes.compress(previous, next, 4) shouldBe empty
+      Bytes.compress(previous, next, 4).toOption() shouldBe empty
 
       Bytes.commonPrefixBytes(previous, next) shouldBe next
     }
@@ -75,7 +75,7 @@ class BytesSpec extends AnyWordSpec with Matchers {
       val previous: Slice[Byte] = Slice(Array(1.toByte, 2.toByte, 3.toByte, 4.toByte))
       val next: Slice[Byte] = Slice(Array(5.toByte, 6.toByte, 7.toByte, 8.toByte, 9.toByte, 10.toByte))
 
-      Bytes.compress(previous, next, 1) shouldBe empty
+      Bytes.compress(previous, next, 1).toOption() shouldBe empty
 
       Bytes.commonPrefixBytes(previous, next) shouldBe Slice.emptyBytes
     }

utils/src/main/scala/swaydb/OK.scala

@@ -19,4 +19,5 @@ package swaydb
 sealed trait OK
 object OK {
   final val instance: OK = new OK {}
+  final val someOK = Some(instance)
 }

utils/src/main/scala/swaydb/utils/TupleOrNone.scala

@@ -16,18 +16,28 @@
 
 package swaydb.utils
 
+/**
+ * Useful for reducing allocations when using `Option[(L, R)]` which when contains a value requires double allocations
+ *  - 1 for scala.Some`
+ *  - And 1 for `Tuple2[L, R]`
+ */
 private[swaydb] sealed trait TupleOrNone[+L, +R] extends SomeOrNoneCovariant[TupleOrNone[L, R], TupleOrNone.Some[L, R]] {
   override def noneC: TupleOrNone[Nothing, Nothing] = TupleOrNone.None
+
+  def toOption(): Option[(L, R)]
+
 }
 
-private[swaydb] object TupleOrNone {
-  final object None extends TupleOrNone[Nothing, Nothing] {
+private[swaydb] case object TupleOrNone {
+  final case object None extends TupleOrNone[Nothing, Nothing] {
     override def isNoneC: Boolean = true
-    override def getC: Some[Nothing, Nothing] = throw new Exception("Tuple is of type Null")
+    override def getC: Some[Nothing, Nothing] = throw new Exception(s"${TupleOrNone.productPrefix} is of type ${None.productPrefix}")
+    override def toOption(): Option[(Nothing, Nothing)] = scala.None
   }
 
   case class Some[+L, +R](left: L, right: R) extends TupleOrNone[L, R] {
     override def isNoneC: Boolean = false
     override def getC: Some[L, R] = this
+    override def toOption(): Option[(L, R)] = scala.Some((left, right))
   }
 }