kvutils: Use VersionedOffsetBuilder where possible, and introduce VersionedOffset. [KVL-1154]

ledger/indexer-benchmark/src/app/scala/ledger/indexerbenchmark/Main.scala

@@ -21,7 +21,7 @@ import com.daml.ledger.participant.state.kvutils.api.{
   LedgerRecord,
 }
 import com.daml.ledger.participant.state.kvutils.export.ProtobufBasedLedgerDataImporter
-import com.daml.ledger.participant.state.kvutils.{OffsetBuilder, Raw}
+import com.daml.ledger.participant.state.kvutils.{Raw, VersionedOffsetBuilder}
 import com.daml.ledger.participant.state.v2.Update
 import com.daml.metrics.Metrics
 
@@ -38,14 +38,15 @@ object Main {
     }
     val importer = ProtobufBasedLedgerDataImporter(path)
 
+    val offsetBuilder = new VersionedOffsetBuilder(0)
     val dataSource: Source[LedgerRecord, NotUsed] = Source
       .fromIterator(() => importer.read().iterator)
       .statefulMapConcat { () =>
         val nextOffset = new AtomicLong(0)
 
         { case (_, writeSet) =>
           writeSet.map { case (key, value) =>
-            val offset = OffsetBuilder.fromLong(nextOffset.getAndIncrement())
+            val offset = offsetBuilder.of(nextOffset.getAndIncrement())
             val logEntryId = Raw.LogEntryId(key.bytes) // `key` is of an unknown type.
             LedgerRecord(offset, logEntryId, value)
           }

ledger/ledger-on-memory/src/app/scala/com/daml/ledger/on/memory/InMemoryLedgerFactory.scala

@@ -31,6 +31,7 @@ private[memory] class InMemoryLedgerFactory(dispatcher: Dispatcher[Index], state
       readerWriter <- new InMemoryLedgerReaderWriter.Owner(
         ledgerId = config.ledgerId,
         participantId = participantConfig.participantId,
+        offsetVersion = 0,
         keySerializationStrategy = DefaultStateKeySerializationStrategy,
         metrics = metrics,
         stateValueCache = caching.WeightedCache.from(

ledger/ledger-on-memory/src/main/scala/com/daml/ledger/on/memory/InMemoryLedgerReader.scala

@@ -8,7 +8,7 @@ import akka.stream.scaladsl.Source
 import com.daml.ledger.api.health.{HealthStatus, Healthy}
 import com.daml.ledger.configuration.LedgerId
 import com.daml.ledger.offset.Offset
-import com.daml.ledger.participant.state.kvutils.OffsetBuilder
+import com.daml.ledger.participant.state.kvutils.VersionedOffsetBuilder
 import com.daml.ledger.participant.state.kvutils.api.{LedgerReader, LedgerRecord}
 import com.daml.metrics.{Metrics, Timed}
 import com.daml.platform.akkastreams.dispatcher.Dispatcher
@@ -17,14 +17,15 @@ import com.daml.platform.akkastreams.dispatcher.SubSource.RangeSource
 class InMemoryLedgerReader(
     override val ledgerId: LedgerId,
     dispatcher: Dispatcher[Index],
+    offsetBuilder: VersionedOffsetBuilder,
     state: InMemoryState,
     metrics: Metrics,
 ) extends LedgerReader {
   override def events(startExclusive: Option[Offset]): Source[LedgerRecord, NotUsed] =
     dispatcher
       .startingAt(
         startExclusive
-          .map(OffsetBuilder.highestIndex(_).toInt)
+          .map(offsetBuilder.highestIndex(_).toInt)
           .getOrElse(StartIndex),
         RangeSource((startExclusive, endInclusive) =>
           Source.fromIterator(() => {

ledger/ledger-on-memory/src/main/scala/com/daml/ledger/on/memory/InMemoryLedgerReaderWriter.scala

@@ -6,6 +6,7 @@ package com.daml.ledger.on.memory
 import com.daml.api.util.TimeProvider
 import com.daml.caching.Cache
 import com.daml.ledger.configuration.LedgerId
+import com.daml.ledger.participant.state.kvutils.VersionedOffsetBuilder
 import com.daml.ledger.participant.state.kvutils.api._
 import com.daml.ledger.resources.{Resource, ResourceContext, ResourceOwner}
 import com.daml.ledger.validator.StateKeySerializationStrategy
@@ -21,6 +22,7 @@ object InMemoryLedgerReaderWriter {
   final class Owner(
       ledgerId: LedgerId,
       participantId: Ref.ParticipantId,
+      offsetVersion: Byte,
       keySerializationStrategy: StateKeySerializationStrategy,
       metrics: Metrics,
       timeProvider: TimeProvider = InMemoryLedgerWriter.DefaultTimeProvider,
@@ -31,7 +33,8 @@ object InMemoryLedgerReaderWriter {
       committerExecutionContext: ExecutionContext,
   ) extends ResourceOwner[KeyValueLedger] {
     override def acquire()(implicit context: ResourceContext): Resource[KeyValueLedger] = {
-      val reader = new InMemoryLedgerReader(ledgerId, dispatcher, state, metrics)
+      val offsetBuilder = new VersionedOffsetBuilder(offsetVersion)
+      val reader = new InMemoryLedgerReader(ledgerId, dispatcher, offsetBuilder, state, metrics)
       for {
         writer <- new InMemoryLedgerWriter.Owner(
           participantId,
@@ -40,6 +43,7 @@ object InMemoryLedgerReaderWriter {
           timeProvider,
           stateValueCache,
           dispatcher,
+          offsetBuilder,
           state,
           engine,
           committerExecutionContext,

ledger/ledger-on-memory/src/main/scala/com/daml/ledger/on/memory/InMemoryLedgerStateAccess.scala

@@ -3,6 +3,7 @@
 
 package com.daml.ledger.on.memory
 
+import com.daml.ledger.participant.state.kvutils.VersionedOffsetBuilder
 import com.daml.ledger.validator.{
   LedgerStateAccess,
   LedgerStateOperations,
@@ -13,15 +14,23 @@ import com.daml.metrics.Metrics
 
 import scala.concurrent.{ExecutionContext, Future}
 
-final class InMemoryLedgerStateAccess(state: InMemoryState, metrics: Metrics)
-    extends LedgerStateAccess[Index] {
+final class InMemoryLedgerStateAccess(
+    offsetBuilder: VersionedOffsetBuilder,
+    state: InMemoryState,
+    metrics: Metrics,
+) extends LedgerStateAccess[Index] {
   override def inTransaction[T](
       body: LedgerStateOperations[Index] => Future[T]
   )(implicit
       executionContext: ExecutionContext,
       loggingContext: LoggingContext,
   ): Future[T] =
     state.write { (log, state) =>
-      body(new TimedLedgerStateOperations(new InMemoryLedgerStateOperations(log, state), metrics))
+      body(
+        new TimedLedgerStateOperations(
+          new InMemoryLedgerStateOperations(offsetBuilder, log, state),
+          metrics,
+        )
+      )
     }
 }

ledger/ledger-on-memory/src/main/scala/com/daml/ledger/on/memory/InMemoryLedgerStateOperations.scala

@@ -4,20 +4,19 @@
 package com.daml.ledger.on.memory
 
 import com.daml.ledger.offset.Offset
-import com.daml.ledger.on.memory.InMemoryLedgerStateOperations._
 import com.daml.ledger.on.memory.InMemoryState.MutableLog
 import com.daml.ledger.participant.state.kvutils.api.LedgerRecord
-import com.daml.ledger.participant.state.kvutils.{OffsetBuilder, Raw}
+import com.daml.ledger.participant.state.kvutils.{Raw, VersionedOffsetBuilder}
 import com.daml.ledger.validator.BatchingLedgerStateOperations
 import com.daml.logging.LoggingContext
 
 import scala.concurrent.{ExecutionContext, Future}
 
 final class InMemoryLedgerStateOperations(
+    offsetBuilder: VersionedOffsetBuilder,
     log: InMemoryState.MutableLog,
     state: InMemoryState.MutableState,
 ) extends BatchingLedgerStateOperations[Index] {
-
   override def readState(
       keys: Iterable[Raw.StateKey]
   )(implicit
@@ -39,16 +38,11 @@ final class InMemoryLedgerStateOperations(
   )(implicit executionContext: ExecutionContext, loggingContext: LoggingContext): Future[Index] =
     Future.successful(appendEntry(log, LedgerRecord(_, key, value)))
 
-}
-
-object InMemoryLedgerStateOperations {
-
-  private[memory] def appendEntry(log: MutableLog, createEntry: Offset => LedgerRecord): Index = {
+  private def appendEntry(log: MutableLog, createEntry: Offset => LedgerRecord): Index = {
     val entryAtIndex = log.size
-    val offset = OffsetBuilder.fromLong(entryAtIndex.toLong)
+    val offset = offsetBuilder.of(entryAtIndex.toLong)
     val entry = createEntry(offset)
     log += entry
     entryAtIndex
   }
-
 }

ledger/ledger-on-memory/src/main/scala/com/daml/ledger/on/memory/InMemoryLedgerWriter.scala

@@ -12,7 +12,7 @@ import com.daml.ledger.on.memory.InMemoryLedgerWriter._
 import com.daml.ledger.participant.state.kvutils.api.{CommitMetadata, LedgerWriter}
 import com.daml.ledger.participant.state.kvutils.export.LedgerDataExporter
 import com.daml.ledger.participant.state.kvutils.store.{DamlStateKey, DamlStateValue}
-import com.daml.ledger.participant.state.kvutils.{KeyValueCommitting, Raw}
+import com.daml.ledger.participant.state.kvutils.{KeyValueCommitting, Raw, VersionedOffsetBuilder}
 import com.daml.ledger.participant.state.v2.SubmissionResult
 import com.daml.ledger.resources.{Resource, ResourceContext, ResourceOwner}
 import com.daml.ledger.validator.caching.{CachingStateReader, ImmutablesOnlyCacheUpdatePolicy}
@@ -39,6 +39,7 @@ import scala.util.Success
 final class InMemoryLedgerWriter private[memory] (
     override val participantId: Ref.ParticipantId,
     dispatcher: Dispatcher[Index],
+    offsetBuilder: VersionedOffsetBuilder,
     now: () => Instant,
     state: InMemoryState,
     committer: Committer,
@@ -57,7 +58,7 @@ final class InMemoryLedgerWriter private[memory] (
         correlationId,
         envelope,
         exportRecordTime = now(),
-        ledgerStateAccess = new InMemoryLedgerStateAccess(state, metrics),
+        ledgerStateAccess = new InMemoryLedgerStateAccess(offsetBuilder, state, metrics),
       )(committerExecutionContext)
       .andThen { case Success(SubmissionResult.Acknowledged) =>
         dispatcher.signalNewHead(state.newHeadSinceLastWrite())
@@ -85,6 +86,7 @@ object InMemoryLedgerWriter {
       timeProvider: TimeProvider = DefaultTimeProvider,
       stateValueCache: StateValueCache = Cache.none,
       dispatcher: Dispatcher[Index],
+      offsetBuilder: VersionedOffsetBuilder,
       state: InMemoryState,
       engine: Engine,
       committerExecutionContext: ExecutionContext,
@@ -97,6 +99,7 @@ object InMemoryLedgerWriter {
       } yield new InMemoryLedgerWriter(
         participantId,
         dispatcher,
+        offsetBuilder,
         now,
         state,
         newCommitter(ledgerDataExporter),

ledger/ledger-on-memory/src/test/suite/scala/com/daml/ledger/on/memory/InMemoryLedgerReaderWriterIntegrationSpec.scala

@@ -37,6 +37,7 @@ class InMemoryLedgerReaderWriterIntegrationSpec
       readerWriter <- new InMemoryLedgerReaderWriter.Owner(
         ledgerId = ledgerId,
         participantId = participantId,
+        offsetVersion = 0,
         keySerializationStrategy = StateKeySerializationStrategy.createDefault(),
         metrics = metrics,
         dispatcher = dispatcher,

ledger/ledger-on-memory/src/test/suite/scala/com/daml/ledger/on/memory/InMemoryLedgerWriterSpec.scala

@@ -7,7 +7,7 @@ import java.time.Instant
 
 import com.codahale.metrics.MetricRegistry
 import com.daml.ledger.api.testing.utils.AkkaBeforeAndAfterAll
-import com.daml.ledger.participant.state.kvutils.Raw
+import com.daml.ledger.participant.state.kvutils.{Raw, VersionedOffsetBuilder}
 import com.daml.ledger.participant.state.kvutils.api.CommitMetadata
 import com.daml.ledger.participant.state.v2.SubmissionResult
 import com.daml.ledger.validator.LedgerStateAccess
@@ -52,6 +52,7 @@ class InMemoryLedgerWriterSpec
       val instance = new InMemoryLedgerWriter(
         participantId = Ref.ParticipantId.assertFromString("participant ID"),
         dispatcher = mockDispatcher,
+        offsetBuilder = new VersionedOffsetBuilder(0),
         now = () => Instant.EPOCH,
         state = InMemoryState.empty,
         committer = mockCommitter,

ledger/ledger-on-sql/src/main/scala/com/daml/ledger/on/sql/Database.scala

@@ -16,14 +16,17 @@ import com.daml.metrics.{Metrics, Timed}
 import com.daml.resources.ProgramResource.StartupException
 import com.zaxxer.hikari.HikariDataSource
 import javax.sql.DataSource
+
+import com.daml.ledger.participant.state.kvutils.VersionedOffsetBuilder
 import org.flywaydb.core.Flyway
 import scalaz.syntax.bind._
 
 import scala.jdk.CollectionConverters._
 import scala.util.{Failure, Success}
 
 final class Database(
-    queries: Connection => Queries,
+    queries: QueriesFactory,
+    offsetBuilder: VersionedOffsetBuilder,
     metrics: Metrics,
 )(implicit
     readerConnectionPool: ConnectionPool[Reader],
@@ -54,7 +57,7 @@ final class Database(
     Timed.future(
       metrics.daml.ledger.database.transactions.run(name),
       Future[X] {
-        body(new TimedQueries(queries(connection), metrics))
+        body(new TimedQueries(queries(offsetBuilder, connection), metrics))
           .andThen {
             case Success(_) => connection.commit()
             case Failure(_) => connection.rollback()
@@ -81,26 +84,28 @@ object Database {
   // entries missing.
   private val MaximumWriterConnectionPoolSize: Int = 1
 
-  def owner(jdbcUrl: String, metrics: Metrics)(implicit
-      loggingContext: LoggingContext
-  ): ResourceOwner[UninitializedDatabase] =
+  def owner(
+      jdbcUrl: String,
+      offsetBuilder: VersionedOffsetBuilder,
+      metrics: Metrics,
+  )(implicit loggingContext: LoggingContext): ResourceOwner[UninitializedDatabase] =
     (jdbcUrl match {
       case "jdbc:h2:mem:" =>
         throw new InvalidDatabaseException(
           "Unnamed in-memory H2 databases are not supported. Please name the database using the format \"jdbc:h2:mem:NAME\"."
         )
       case url if url.startsWith("jdbc:h2:mem:") =>
-        SingleConnectionDatabase.owner(RDBMS.H2, jdbcUrl, metrics)
+        SingleConnectionDatabase.owner(RDBMS.H2, jdbcUrl, offsetBuilder, metrics)
       case url if url.startsWith("jdbc:h2:") =>
-        MultipleConnectionDatabase.owner(RDBMS.H2, jdbcUrl, metrics)
+        MultipleConnectionDatabase.owner(RDBMS.H2, jdbcUrl, offsetBuilder, metrics)
       case url if url.startsWith("jdbc:postgresql:") =>
-        MultipleConnectionDatabase.owner(RDBMS.PostgreSQL, jdbcUrl, metrics)
+        MultipleConnectionDatabase.owner(RDBMS.PostgreSQL, jdbcUrl, offsetBuilder, metrics)
       case url if url.startsWith("jdbc:sqlite::memory:") =>
         throw new InvalidDatabaseException(
           "Unnamed in-memory SQLite databases are not supported. Please name the database using the format \"jdbc:sqlite:file:NAME?mode=memory&cache=shared\"."
         )
       case url if url.startsWith("jdbc:sqlite:") =>
-        SingleConnectionDatabase.owner(RDBMS.SQLite, jdbcUrl, metrics)
+        SingleConnectionDatabase.owner(RDBMS.SQLite, jdbcUrl, offsetBuilder, metrics)
       case _ =>
         throw new InvalidDatabaseException(s"Unknown database")
     }).map { database =>
@@ -112,6 +117,7 @@ object Database {
     def owner(
         system: RDBMS,
         jdbcUrl: String,
+        offsetBuilder: VersionedOffsetBuilder,
         metrics: Metrics,
     ): ResourceOwner[UninitializedDatabase] =
       for {
@@ -139,14 +145,15 @@ object Database {
           new ConnectionPool(writerDataSource)
         implicit val adminConnectionPool: ConnectionPool[Migrator] =
           new ConnectionPool(adminDataSource)(DirectExecutionContext)
-        new UninitializedDatabase(system = system, metrics = metrics)
+        new UninitializedDatabase(system, offsetBuilder, metrics)
       }
   }
 
   object SingleConnectionDatabase {
     def owner(
         system: RDBMS,
         jdbcUrl: String,
+        offsetBuilder: VersionedOffsetBuilder,
         metrics: Metrics,
     ): ResourceOwner[UninitializedDatabase] =
       for {
@@ -164,7 +171,7 @@ object Database {
           new ConnectionPool(readerWriterDataSource)
         implicit val adminConnectionPool: ConnectionPool[Migrator] =
           new ConnectionPool(adminDataSource)(DirectExecutionContext)
-        new UninitializedDatabase(system = system, metrics = metrics)
+        new UninitializedDatabase(system, offsetBuilder, metrics)
       }
   }
 
@@ -184,31 +191,34 @@ object Database {
   sealed trait RDBMS {
     val name: String
 
-    val queries: Connection => Queries
+    val queries: QueriesFactory
   }
 
   object RDBMS {
     object H2 extends RDBMS {
       override val name: String = "h2"
 
-      override val queries: Connection => Queries = H2Queries.apply
+      override val queries: QueriesFactory = H2Queries.apply
     }
 
     object PostgreSQL extends RDBMS {
       override val name: String = "postgresql"
 
-      override val queries: Connection => Queries = PostgresqlQueries.apply
+      override val queries: QueriesFactory = PostgresqlQueries.apply
     }
 
     object SQLite extends RDBMS {
       override val name: String = "sqlite"
 
-      override val queries: Connection => Queries = SqliteQueries.apply
+      override val queries: QueriesFactory = SqliteQueries.apply
     }
-
   }
 
-  class UninitializedDatabase(system: RDBMS, metrics: Metrics)(implicit
+  class UninitializedDatabase(
+      system: RDBMS,
+      offsetBuilder: VersionedOffsetBuilder,
+      metrics: Metrics,
+  )(implicit
       readerConnectionPool: ConnectionPool[Reader],
       writerConnectionPool: ConnectionPool[Writer],
       adminConnectionPool: ConnectionPool[Migrator],
@@ -224,7 +234,7 @@ object Database {
 
     def migrate(): Database = {
       flyway.migrate()
-      new Database(queries = system.queries, metrics = metrics)
+      new Database(system.queries, offsetBuilder, metrics)
     }
 
     def migrateAndReset()(implicit