Merge pull request #48 from scalableminds/multi-get-put

fm3 · web-flow · commit 9177cdd3f27a · 2025-01-14T11:23:50.000+01:00
Add put + get with multiple keys *and* versions
diff --git a/Changelog.md b/Changelog.md
@@ -2,6 +2,8 @@
 
 ## Added
  - New API endpoints `DeleteAllByPrefix` and `PutMultipleVersions`. [#47](https://github.com/scalableminds/fossildb/pull/47)
+ - New API endpoints `GetMultipleKeysByListWithMultipleVersions` and `PutMultipleKeysWithMultipleVersions` for reading and writing multiple keys/versions in one request. [#48](https://github.com/scalableminds/fossildb/pull/48)
+ - `ListKeys` now supports optional `prefix` field
 
 ## Breaking Changes
 
diff --git a/build.sbt b/build.sbt
@@ -21,7 +21,7 @@ ThisBuild / scalacOptions ++= Seq(
 
 version := getVersionFromGit
 
-scalaVersion := "2.13.12"
+scalaVersion := "2.13.15"
 
 libraryDependencies ++= Seq(
   "ch.qos.logback" % "logback-classic" % "1.5.6",
@@ -30,7 +30,7 @@ libraryDependencies ++= Seq(
   "io.grpc" % "grpc-netty" % scalapb.compiler.Version.grpcJavaVersion,
   "io.grpc" % "grpc-services" % scalapb.compiler.Version.grpcJavaVersion,
   "com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapb.compiler.Version.scalapbVersion,
-  "org.rocksdb" % "rocksdbjni" % "8.10.0",
+  "org.rocksdb" % "rocksdbjni" % "9.4.0",
   "com.github.scopt" %% "scopt" % "4.1.0"
 )
 
diff --git a/project/scalapb.sbt b/project/scalapb.sbt
@@ -1,3 +1,3 @@
 addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.2")
 
-libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.11.13"
+libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.11.15"
diff --git a/src/main/protobuf/fossildbapi.proto b/src/main/protobuf/fossildbapi.proto
@@ -2,6 +2,22 @@ syntax = "proto2";
 
 package com.scalableminds.fossildb.proto;
 
+message VersionedKeyValuePairProto {
+    required string key = 1;
+    required uint64 version = 2;
+    required bytes value = 3;
+}
+
+message KeyVersionsValuesPairProto {
+    required string key = 1;
+    repeated VersionValuePairProto versionValuePairs = 2;
+}
+
+message VersionValuePairProto {
+    required uint64 actualVersion = 1;
+    required bytes value = 2;
+}
+
 message HealthRequest {}
 message HealthReply {
     required bool success = 1;
@@ -46,6 +62,16 @@ message PutMultipleVersionsReply {
     optional string errorMessage = 2;
 }
 
+message PutMultipleKeysWithMultipleVersionsRequest {
+    required string collection = 1;
+    repeated VersionedKeyValuePairProto versionedKeyValuePairs = 2;
+}
+
+message PutMultipleKeysWithMultipleVersionsReply {
+    required bool success = 1;
+    optional string errorMessage = 2;
+}
+
 message DeleteRequest {
     required string collection = 1;
     required string key = 2;
@@ -97,6 +123,19 @@ message GetMultipleKeysReply {
     repeated uint64 actualVersions = 5;
 }
 
+message GetMultipleKeysByListWithMultipleVersionsRequest {
+    required string collection = 1;
+    repeated string keys = 2;
+    optional uint64 newestVersion = 3; // Applied to all requested keys
+    optional uint64 oldestVersion = 4; // Applied to all requested keys
+}
+
+message GetMultipleKeysByListWithMultipleVersionsReply {
+    required bool success = 1;
+    optional string errorMessage = 2;
+    repeated KeyVersionsValuesPairProto keyVersionsValuesPairs = 3;
+}
+
 message DeleteMultipleVersionsRequest {
     required string collection = 1;
     required string key = 2;
@@ -113,6 +152,7 @@ message ListKeysRequest {
     required string collection = 1;
     optional uint32 limit = 2;
     optional string startAfterKey = 3;
+    optional string prefix = 4;
 }
 
 message ListKeysReply {
@@ -175,8 +215,10 @@ service FossilDB {
     rpc Get (GetRequest) returns (GetReply) {}
     rpc GetMultipleVersions (GetMultipleVersionsRequest) returns (GetMultipleVersionsReply) {}
     rpc GetMultipleKeys (GetMultipleKeysRequest) returns (GetMultipleKeysReply) {}
+    rpc GetMultipleKeysByListWithMultipleVersions (GetMultipleKeysByListWithMultipleVersionsRequest) returns (GetMultipleKeysByListWithMultipleVersionsReply) {}
     rpc Put (PutRequest) returns (PutReply) {}
     rpc PutMultipleVersions (PutMultipleVersionsRequest) returns (PutMultipleVersionsReply) {}
+    rpc PutMultipleKeysWithMultipleVersions (PutMultipleKeysWithMultipleVersionsRequest) returns (PutMultipleKeysWithMultipleVersionsReply) {}
     rpc Delete (DeleteRequest) returns (DeleteReply) {}
     rpc DeleteMultipleVersions (DeleteMultipleVersionsRequest) returns (DeleteMultipleVersionsReply) {}
     rpc DeleteAllByPrefix (DeleteAllByPrefixRequest) returns (DeleteAllByPrefixReply) {}
diff --git a/src/main/scala/com/scalableminds/fossildb/FossilDBGrpcImpl.scala b/src/main/scala/com/scalableminds/fossildb/FossilDBGrpcImpl.scala
@@ -64,6 +64,27 @@ class FossilDBGrpcImpl(storeManager: StoreManager)
     GetMultipleKeysReply(success = true, None, keys, values.map(ByteString.copyFrom), versions)
   } { errorMsg => GetMultipleKeysReply(success = false, errorMsg) }
 
+  override def getMultipleKeysByListWithMultipleVersions(req: GetMultipleKeysByListWithMultipleVersionsRequest): Future[GetMultipleKeysByListWithMultipleVersionsReply] = withExceptionHandler(req) {
+    val store = storeManager.getStore(req.collection)
+    val keyVersionsValuesPairs = req.keys.map { key =>
+      val (values, versions) = store.withRawRocksIterator{rocksIt => store.getMultipleVersions(rocksIt, key, req.oldestVersion, req.newestVersion)}
+      val versionValuePairs = values.zip(versions).map { case (value, version) =>
+        VersionValuePairProto(version, ByteString.copyFrom(value))
+      }
+      KeyVersionsValuesPairProto(key, versionValuePairs)
+    }
+    GetMultipleKeysByListWithMultipleVersionsReply(success = true, None, keyVersionsValuesPairs)
+  } { errorMsg => GetMultipleKeysByListWithMultipleVersionsReply(success = false, errorMsg) }
+
+  override def putMultipleKeysWithMultipleVersions(req: PutMultipleKeysWithMultipleVersionsRequest): Future[PutMultipleKeysWithMultipleVersionsReply] = withExceptionHandler(req) {
+    val store = storeManager.getStore(req.collection)
+    require(req.versionedKeyValuePairs.forall(_.version >= 0), "Version numbers must be non-negative")
+    req.versionedKeyValuePairs.foreach { pair =>
+      store.put(pair.key, pair.version, pair.value.toByteArray)
+    }
+    PutMultipleKeysWithMultipleVersionsReply(success = true, None)
+  } { errorMsg => PutMultipleKeysWithMultipleVersionsReply(success = false, errorMsg) }
+
   override def deleteMultipleVersions(req: DeleteMultipleVersionsRequest): Future[DeleteMultipleVersionsReply] = withExceptionHandler(req) {
     val store = storeManager.getStore(req.collection)
     store.withRawRocksIterator{rocksIt => store.deleteMultipleVersions(rocksIt, req.key, req.oldestVersion, req.newestVersion)}
@@ -78,7 +99,7 @@ class FossilDBGrpcImpl(storeManager: StoreManager)
 
   override def listKeys(req: ListKeysRequest): Future[ListKeysReply] = withExceptionHandler(req) {
     val store = storeManager.getStore(req.collection)
-    val keys = store.withRawRocksIterator{rocksIt => store.listKeys(rocksIt, req.limit, req.startAfterKey)}
+    val keys = store.withRawRocksIterator{rocksIt => store.listKeys(rocksIt, req.limit, req.startAfterKey, req.prefix)}
     ListKeysReply(success = true, None, keys)
   } { errorMsg => ListKeysReply(success = false, errorMsg) }
 
diff --git a/src/main/scala/com/scalableminds/fossildb/db/VersionedKeyValueStore.scala b/src/main/scala/com/scalableminds/fossildb/db/VersionedKeyValueStore.scala
@@ -7,11 +7,13 @@ import scala.util.Try
 
 
 case class VersionedKey(key: String, version: Long) {
-  override def toString: String = s"$key${VersionedKey.versionSeparator}${(~version).toHexString.toUpperCase}${VersionedKey.versionSeparator}$version"
+  override def toString: String = VersionedKey.asString(key, version)
 }
 
 object VersionedKey {
 
+  def asString(key: String, version: Long) = s"$key${VersionedKey.versionSeparator}${(~version).toHexString.toUpperCase}${VersionedKey.versionSeparator}$version"
+
   val versionSeparator: Char = '@'
 
   def apply(key: String): Option[VersionedKey] = {
@@ -24,6 +26,7 @@ object VersionedKey {
       VersionedKey(key, version)
     }
   }
+
 }
 
 case class VersionedKeyValuePair[T](versionedKey: VersionedKey, value: T) {
@@ -58,7 +61,7 @@ class VersionFilterIterator(it: RocksDBIterator, version: Option[Long]) extends
 
 }
 
-class KeyOnlyIterator[T](rocksIt: RocksIterator, startAfterKey: Option[String]) extends Iterator[String] {
+class KeyOnlyIterator[T](rocksIt: RocksIterator, startAfterKey: Option[String], prefix: Option[String]) extends Iterator[String] {
 
   /*
      Note that seek in the underlying iterators either hits precisely or goes to the
@@ -70,17 +73,18 @@ class KeyOnlyIterator[T](rocksIt: RocksIterator, startAfterKey: Option[String])
 
   private def compositeKeyFor(keyOpt: Option[String]) = keyOpt match {
     case Some(key) => VersionedKey(key, 0).toString
-    case None => ""
+    // If the currentKey is not yet set, seek to the very beginning, or, if set, to the prefix.
+    case None => prefix.getOrElse("")
   }
 
   override def hasNext: Boolean = {
-    val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), None)
+    val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), prefix)
     if (it.hasNext && currentKey.isDefined && currentKey.contains(VersionedKey(it.peek).get.key)) it.next()
     it.hasNext
   }
 
   override def next(): String = {
-    val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), None)
+    val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), prefix)
     if (it.hasNext && currentKey.isDefined && currentKey.contains(VersionedKey(it.peek).get.key)) it.next()
     val nextKey = VersionedKey(it.next()).get.key
     currentKey = Some(nextKey)
@@ -186,16 +190,16 @@ class VersionedKeyValueStore(underlying: RocksDBStore) {
 
   def put(key: String, version: Long, value: Array[Byte]): Unit = {
     requireValidKey(key)
-    underlying.put(VersionedKey(key, version).toString, value)
+    underlying.put(VersionedKey.asString(key, version), value)
   }
 
   def delete(key: String, version: Long): Unit = {
     requireValidKey(key)
-    underlying.delete(VersionedKey(key, version).toString)
+    underlying.delete(VersionedKey.asString(key, version))
   }
 
-  def listKeys(rocksIt: RocksIterator, limit: Option[Int], startAfterKey: Option[String]): Seq[String] = {
-    val iterator = new KeyOnlyIterator(rocksIt, startAfterKey)
+  def listKeys(rocksIt: RocksIterator, limit: Option[Int], startAfterKey: Option[String], prefix: Option[String]): Seq[String] = {
+    val iterator = new KeyOnlyIterator(rocksIt, startAfterKey, prefix)
     iterator.take(limit.getOrElse(Int.MaxValue)).toSeq
   }
 
@@ -207,4 +211,5 @@ class VersionedKeyValueStore(underlying: RocksDBStore) {
   private def requireValidKey(key: String): Unit = {
     require(!key.contains(VersionedKey.versionSeparator), s"keys cannot contain the char ${VersionedKey.versionSeparator}")
   }
+
 }
diff --git a/src/test/scala/com/scalableminds/fossildb/FossilDBSuite.scala b/src/test/scala/com/scalableminds/fossildb/FossilDBSuite.scala
@@ -97,6 +97,17 @@ class FossilDBSuite extends AnyFlatSpec with BeforeAndAfterEach with TestHelpers
     assert(reply.actualVersion == 0)
   }
 
+  "PutMultipleKeysWithMultipleVersions" should "write all versions of all specified keys" in {
+    client.putMultipleKeysWithMultipleVersions(PutMultipleKeysWithMultipleVersionsRequest(collectionA, Seq(VersionedKeyValuePairProto(aKey, 0, testData1), VersionedKeyValuePairProto(aKey, 2, testData2), VersionedKeyValuePairProto(aNotherKey, 5, testData3))))
+    val reply = client.get(GetRequest(collectionA, aKey))
+    assert(reply.actualVersion == 2)
+    val reply2 = client.get(GetRequest(collectionA, aKey, version = Some(0)))
+    assert(reply2.actualVersion == 0)
+    val reply3 = client.get(GetRequest(collectionA, aNotherKey))
+    assert(reply3.actualVersion == 5)
+    assert(reply3.value == testData3)
+  }
+
   "Get" should "return matching value after matching Put" in {
     client.put(PutRequest(collectionA, aKey, Some(0), testData1))
     val reply = client.get(GetRequest(collectionA, aKey, Some(0)))
@@ -195,6 +206,37 @@ class FossilDBSuite extends AnyFlatSpec with BeforeAndAfterEach with TestHelpers
     assert(reply.keys.length == 3)
   }
 
+  it should "respect prefix argument" in {
+    client.put(PutRequest(collectionA, "123456", Some(1), testData1))
+    client.put(PutRequest(collectionA, "123457", Some(123), testData2))
+    client.put(PutRequest(collectionA, "12345800", Some(123), testData3))
+    client.put(PutRequest(collectionA, "12345801", Some(123), testData3))
+    client.put(PutRequest(collectionA, "12345802", Some(123), testData3))
+    client.put(PutRequest(collectionA, "123458", Some(123), testData3))
+    client.put(PutRequest(collectionA, "123459", Some(123), testData3))
+
+    val reply = client.listKeys(ListKeysRequest(collectionA, None, None, prefix = Some("123458")))
+    assert(reply.keys.length == 4)
+    assert(reply.keys(0) == "12345800")
+    assert(reply.keys(1) == "12345801")
+  }
+
+  it should "respect prefix argument and startAfterKey together" in {
+    client.put(PutRequest(collectionA, "123456", Some(1), testData1))
+    client.put(PutRequest(collectionA, "123457", Some(123), testData2))
+    client.put(PutRequest(collectionA, "12345800", Some(123), testData3))
+    client.put(PutRequest(collectionA, "12345801", Some(123), testData3))
+    client.put(PutRequest(collectionA, "12345802", Some(123), testData3))
+    client.put(PutRequest(collectionA, "123458", Some(123), testData3))
+    client.put(PutRequest(collectionA, "123459", Some(123), testData3))
+
+    val reply = client.listKeys(ListKeysRequest(collectionA, None, startAfterKey = Some("12345800"), prefix = Some("123458")))
+    assert(reply.keys.length == 3)
+    assert(reply.keys(0) == "12345801")
+    assert(reply.keys(1) == "12345802")
+    assert(reply.keys(2) == "123458")
+  }
+
   "GetMultipleVersions" should "return all versions in decending order if called without limits" in {
     client.put(PutRequest(collectionA, aKey, Some(0), testData1))
     client.put(PutRequest(collectionA, aKey, Some(1), testData2))
@@ -348,6 +390,46 @@ class FossilDBSuite extends AnyFlatSpec with BeforeAndAfterEach with TestHelpers
     assert(reply.keys.isEmpty)
   }
 
+  "GetMultipleKeysByListWithVersions" should "return selected keys with versions in descending order" in {
+    client.put(PutRequest(collectionA, aKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aNotherKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aThirdKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aNotherKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aThirdKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aKey, Some(2), testData3))
+    client.put(PutRequest(collectionA, aNotherKey, Some(2), testData3))
+    client.put(PutRequest(collectionA, aThirdKey, Some(2), testData3))
+    val reply = client.getMultipleKeysByListWithMultipleVersions(GetMultipleKeysByListWithMultipleVersionsRequest(collectionA, keys = Seq(aNotherKey, aThirdKey)))
+    assert(reply.keyVersionsValuesPairs.map(_.key) == Seq(aNotherKey, aThirdKey))
+    assert(reply.keyVersionsValuesPairs(0).versionValuePairs.length == 3)
+    assert(reply.keyVersionsValuesPairs(1).versionValuePairs.length == 3)
+    assert(reply.keyVersionsValuesPairs(0).versionValuePairs(0) == VersionValuePairProto(2L, testData3))
+    assert(reply.keyVersionsValuesPairs(0).versionValuePairs(1) == VersionValuePairProto(1L, testData2))
+    assert(reply.keyVersionsValuesPairs(0).versionValuePairs(2) == VersionValuePairProto(0L, testData1))
+    assert(reply.keyVersionsValuesPairs(1).versionValuePairs(0) == VersionValuePairProto(2L, testData3))
+    assert(reply.keyVersionsValuesPairs(1).versionValuePairs(1) == VersionValuePairProto(1L, testData2))
+    assert(reply.keyVersionsValuesPairs(1).versionValuePairs(2) == VersionValuePairProto(0L, testData1))
+  }
+
+  it should "limit the versions if specified" in {
+    client.put(PutRequest(collectionA, aKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aNotherKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aThirdKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aNotherKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aThirdKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aKey, Some(2), testData3))
+    client.put(PutRequest(collectionA, aNotherKey, Some(2), testData3))
+    client.put(PutRequest(collectionA, aThirdKey, Some(2), testData3))
+    val reply = client.getMultipleKeysByListWithMultipleVersions(GetMultipleKeysByListWithMultipleVersionsRequest(collectionA, keys = Seq(aNotherKey, aThirdKey), newestVersion = Some(1), oldestVersion = Some(1)))
+    assert(reply.keyVersionsValuesPairs.map(_.key) == Seq(aNotherKey, aThirdKey))
+    assert(reply.keyVersionsValuesPairs(0).versionValuePairs.length == 1)
+    assert(reply.keyVersionsValuesPairs(1).versionValuePairs.length == 1)
+    assert(reply.keyVersionsValuesPairs(0).versionValuePairs(0) == VersionValuePairProto(1L, testData2))
+    assert(reply.keyVersionsValuesPairs(1).versionValuePairs(0) == VersionValuePairProto(1L, testData2))
+  }
+
   "Backup" should "create non-empty backup directory" in {
     client.put(PutRequest(collectionA, aKey, Some(0), testData1))
     client.backup(BackupRequest())

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,3 @@`
`1`	`1`	`addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.2")`
`2`	`2`
`3`		`-libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.11.13"`
	`3`	`+libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.11.15"`
Original file line number	Diff line number	Diff line change
`@@ -7,11 +7,13 @@ import scala.util.Try`
`7`	`7`
`8`	`8`
`9`	`9`	`case class VersionedKey(key: String, version: Long) {`
`10`		`- override def toString: String = s"$key${VersionedKey.versionSeparator}${(~version).toHexString.toUpperCase}${VersionedKey.versionSeparator}$version"`
	`10`	`+ override def toString: String = VersionedKey.asString(key, version)`
`11`	`11`	`}`
`12`	`12`
`13`	`13`	`object VersionedKey {`
`14`	`14`
	`15`	`+ def asString(key: String, version: Long) = s"$key${VersionedKey.versionSeparator}${(~version).toHexString.toUpperCase}${VersionedKey.versionSeparator}$version"`
	`16`	`+`
`15`	`17`	`val versionSeparator: Char = '@'`
`16`	`18`
`17`	`19`	`def apply(key: String): Option[VersionedKey] = {`
`@@ -24,6 +26,7 @@ object VersionedKey {`
`24`	`26`	`VersionedKey(key, version)`
`25`	`27`	`}`
`26`	`28`	`}`
	`29`	`+`
`27`	`30`	`}`
`28`	`31`
`29`	`32`	`case class VersionedKeyValuePair[T](versionedKey: VersionedKey, value: T) {`
`@@ -58,7 +61,7 @@ class VersionFilterIterator(it: RocksDBIterator, version: Option[Long]) extends`
`58`	`61`
`59`	`62`	`}`
`60`	`63`
`61`		`-class KeyOnlyIterator[T](rocksIt: RocksIterator, startAfterKey: Option[String]) extends Iterator[String] {`
	`64`	`+class KeyOnlyIterator[T](rocksIt: RocksIterator, startAfterKey: Option[String], prefix: Option[String]) extends Iterator[String] {`
`62`	`65`
`63`	`66`	`/*`
`64`	`67`	`Note that seek in the underlying iterators either hits precisely or goes to the`
`@@ -70,17 +73,18 @@ class KeyOnlyIterator[T](rocksIt: RocksIterator, startAfterKey: Option[String])`
`70`	`73`
`71`	`74`	`private def compositeKeyFor(keyOpt: Option[String]) = keyOpt match {`
`72`	`75`	`case Some(key) => VersionedKey(key, 0).toString`
`73`		`- case None => ""`
	`76`	`+ // If the currentKey is not yet set, seek to the very beginning, or, if set, to the prefix.`
	`77`	`+ case None => prefix.getOrElse("")`
`74`	`78`	`}`
`75`	`79`
`76`	`80`	`override def hasNext: Boolean = {`
`77`		`- val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), None)`
	`81`	`+ val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), prefix)`
`78`	`82`	`if (it.hasNext && currentKey.isDefined && currentKey.contains(VersionedKey(it.peek).get.key)) it.next()`
`79`	`83`	`it.hasNext`
`80`	`84`	`}`
`81`	`85`
`82`	`86`	`override def next(): String = {`
`83`		`- val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), None)`
	`87`	`+ val it = RocksDBStore.scanKeysOnly(rocksIt, compositeKeyFor(currentKey), prefix)`
`84`	`88`	`if (it.hasNext && currentKey.isDefined && currentKey.contains(VersionedKey(it.peek).get.key)) it.next()`
`85`	`89`	`val nextKey = VersionedKey(it.next()).get.key`
`86`	`90`	`currentKey = Some(nextKey)`
`@@ -186,16 +190,16 @@ class VersionedKeyValueStore(underlying: RocksDBStore) {`
`186`	`190`
`187`	`191`	`def put(key: String, version: Long, value: Array[Byte]): Unit = {`
`188`	`192`	`requireValidKey(key)`
`189`		`- underlying.put(VersionedKey(key, version).toString, value)`
	`193`	`+ underlying.put(VersionedKey.asString(key, version), value)`
`190`	`194`	`}`
`191`	`195`
`192`	`196`	`def delete(key: String, version: Long): Unit = {`
`193`	`197`	`requireValidKey(key)`
`194`		`- underlying.delete(VersionedKey(key, version).toString)`
	`198`	`+ underlying.delete(VersionedKey.asString(key, version))`
`195`	`199`	`}`
`196`	`200`
`197`		`- def listKeys(rocksIt: RocksIterator, limit: Option[Int], startAfterKey: Option[String]): Seq[String] = {`
`198`		`- val iterator = new KeyOnlyIterator(rocksIt, startAfterKey)`
	`201`	`+ def listKeys(rocksIt: RocksIterator, limit: Option[Int], startAfterKey: Option[String], prefix: Option[String]): Seq[String] = {`
	`202`	`+ val iterator = new KeyOnlyIterator(rocksIt, startAfterKey, prefix)`
`199`	`203`	`iterator.take(limit.getOrElse(Int.MaxValue)).toSeq`
`200`	`204`	`}`
`201`	`205`
`@@ -207,4 +211,5 @@ class VersionedKeyValueStore(underlying: RocksDBStore) {`
`207`	`211`	`private def requireValidKey(key: String): Unit = {`
`208`	`212`	`require(!key.contains(VersionedKey.versionSeparator), s"keys cannot contain the char ${VersionedKey.versionSeparator}")`
`209`	`213`	`}`
	`214`	`+`
`210`	`215`	`}`