Merge pull request #52 from scalableminds/multi-get

Add new api call GetMultipleKeysByList

Author: fm3

Changelog.md

@@ -4,6 +4,7 @@
  - 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
+ - New API endpoint `GetMultipleKeysByList`. [#52](https://github.com/scalableminds/fossildb/pull/52)
 
 ## Breaking Changes
 

src/main/protobuf/fossildbapi.proto

@@ -18,6 +18,11 @@ message VersionValuePairProto {
     required bytes value = 2;
 }
 
+message VersionValueBoxProto {
+    optional VersionValuePairProto versionValuePair = 1;
+    optional string errorMessage = 2;
+}
+
 message HealthRequest {}
 message HealthReply {
     required bool success = 1;
@@ -123,6 +128,18 @@ message GetMultipleKeysReply {
     repeated uint64 actualVersions = 5;
 }
 
+message GetMultipleKeysByListRequest {
+    required string collection = 1;
+    repeated string keys = 2;
+    optional uint64 version = 3; // Applied to all requested keys
+}
+
+message GetMultipleKeysByListReply {
+    required bool success = 1;
+    optional string errorMessage = 2;
+    repeated VersionValueBoxProto versionValueBoxes = 3;
+}
+
 message GetMultipleKeysByListWithMultipleVersionsRequest {
     required string collection = 1;
     repeated string keys = 2;
@@ -215,6 +232,7 @@ service FossilDB {
     rpc Get (GetRequest) returns (GetReply) {}
     rpc GetMultipleVersions (GetMultipleVersionsRequest) returns (GetMultipleVersionsReply) {}
     rpc GetMultipleKeys (GetMultipleKeysRequest) returns (GetMultipleKeysReply) {}
+    rpc GetMultipleKeysByList (GetMultipleKeysByListRequest) returns (GetMultipleKeysByListReply) {}
     rpc GetMultipleKeysByListWithMultipleVersions (GetMultipleKeysByListWithMultipleVersionsRequest) returns (GetMultipleKeysByListWithMultipleVersionsReply) {}
     rpc Put (PutRequest) returns (PutReply) {}
     rpc PutMultipleVersions (PutMultipleVersionsRequest) returns (PutMultipleVersionsReply) {}

src/main/scala/com/scalableminds/fossildb/FossilDBGrpcImpl.scala

@@ -76,6 +76,18 @@ class FossilDBGrpcImpl(storeManager: StoreManager)
     GetMultipleKeysByListWithMultipleVersionsReply(success = true, None, keyVersionsValuesPairs)
   } { errorMsg => GetMultipleKeysByListWithMultipleVersionsReply(success = false, errorMsg) }
 
+  override def getMultipleKeysByList(req: GetMultipleKeysByListRequest): Future[GetMultipleKeysByListReply] = withExceptionHandler(req) {
+    val store = storeManager.getStore(req.collection)
+    val versionValueBoxes = req.keys.map { key =>
+      val versionedKeyValuePairOpt = store.withRawRocksIterator { rocksIt => store.get(rocksIt, key, req.version) }
+      versionedKeyValuePairOpt match {
+        case Some(pair) => VersionValueBoxProto(Some(VersionValuePairProto(pair.version, ByteString.copyFrom(pair.value))), errorMessage = None)
+        case None => VersionValueBoxProto(None, errorMessage = None)
+      }
+    }
+    GetMultipleKeysByListReply(success = true, None, versionValueBoxes)
+  } { errorMsg => GetMultipleKeysByListReply(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")

src/test/scala/com/scalableminds/fossildb/FossilDBSuite.scala

@@ -441,6 +441,41 @@ class FossilDBSuite extends AnyFlatSpec with BeforeAndAfterEach with TestHelpers
     assert(reply.keyVersionsValuesPairs.isEmpty)
   }
 
+  "GetMultipleKeysByList" should "return version-value tuples for existing, and empty for missing keys" in {
+    client.put(PutRequest(collectionA, aKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aNotherKey, Some(0), testData2))
+    client.put(PutRequest(collectionA, aNotherKey, Some(1), testData3))
+    val reply = client.getMultipleKeysByList(GetMultipleKeysByListRequest(collectionA, keys = Seq(aKey, aNotherKey, aThirdKey)))
+    assert(reply.versionValueBoxes.length == 3)
+    assert(reply.versionValueBoxes(0).versionValuePair.exists(_.value == testData1))
+    assert(reply.versionValueBoxes(1).versionValuePair.exists(_.value == testData3))
+    assert(reply.versionValueBoxes(1).versionValuePair.exists(_.actualVersion == 1))
+    assert(reply.versionValueBoxes(2).versionValuePair.isEmpty)
+  }
+
+  it should "not return something newer than the requested version" in {
+    client.put(PutRequest(collectionA, aKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aNotherKey, Some(0), testData1))
+    client.put(PutRequest(collectionA, aNotherKey, Some(1), testData2))
+    client.put(PutRequest(collectionA, aNotherKey, Some(2), testData3))
+    client.put(PutRequest(collectionA, aThirdKey, Some(2), testData3))
+    val reply = client.getMultipleKeysByList(GetMultipleKeysByListRequest(collectionA, keys = Seq(aKey, aNotherKey, aThirdKey), version = Some(1)))
+    assert(reply.versionValueBoxes.length == 3)
+    assert(reply.versionValueBoxes(0).versionValuePair.exists(_.value == testData1))
+    assert(reply.versionValueBoxes(0).versionValuePair.exists(_.actualVersion == 0))
+    assert(reply.versionValueBoxes(1).versionValuePair.exists(_.value == testData2))
+    assert(reply.versionValueBoxes(1).versionValuePair.exists(_.actualVersion == 1))
+    assert(reply.versionValueBoxes(2).versionValuePair.isEmpty)
+  }
+
+  it should "return only empty boxes if nothing matches" in {
+    client.put(PutRequest(collectionA, aKey, Some(2), testData1))
+    client.put(PutRequest(collectionA, aNotherKey, Some(2), testData1))
+    val reply = client.getMultipleKeysByList(GetMultipleKeysByListRequest(collectionA, keys = Seq(aKey, aNotherKey, aThirdKey), version = Some(1)))
+    assert(reply.versionValueBoxes.length == 3)
+    assert(reply.versionValueBoxes.forall(_.versionValuePair.isEmpty))
+  }
+
   "Backup" should "create non-empty backup directory" in {
     client.put(PutRequest(collectionA, aKey, Some(0), testData1))
     client.backup(BackupRequest())