Implement Parallelized map and optimize Database search API

engine/src/main/java/com/google/android/fhir/Util.kt

@@ -26,6 +26,10 @@ import java.time.ZoneId
 import java.time.format.DateTimeFormatter
 import java.util.Date
 import java.util.Locale
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
 import org.hl7.fhir.r4.model.OperationOutcome
 import org.hl7.fhir.r4.model.Resource
 import org.hl7.fhir.r4.model.ResourceType
@@ -69,6 +73,11 @@ internal fun Resource.isUploadSuccess(): Boolean {
     outcome.issue.all { it.severity.equals(OperationOutcome.IssueSeverity.INFORMATION) }
 }
 
+/** Implementation of a parallelized map for CPU intensive tasks */
+suspend fun <A, B> Iterable<A>.pmapCPU(f: suspend (A) -> B): List<B> = coroutineScope {
+  map { async(Dispatchers.Default) { f(it) } }.awaitAll()
+}
+
 internal class OffsetDateTimeTypeAdapter : TypeAdapter<OffsetDateTime>() {
   override fun write(out: JsonWriter, value: OffsetDateTime) {
     out.value(DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(value))

engine/src/main/java/com/google/android/fhir/db/impl/DatabaseImpl.kt

@@ -21,6 +21,7 @@ import androidx.annotation.VisibleForTesting
 import androidx.room.Room
 import androidx.room.withTransaction
 import androidx.sqlite.db.SimpleSQLiteQuery
+import ca.uhn.fhir.context.FhirContext
 import ca.uhn.fhir.parser.IParser
 import ca.uhn.fhir.util.FhirTerser
 import com.google.android.fhir.DatabaseErrorStrategy
@@ -37,6 +38,7 @@ import com.google.android.fhir.db.impl.entities.LocalChangeEntity
 import com.google.android.fhir.db.impl.entities.ResourceEntity
 import com.google.android.fhir.index.ResourceIndexer
 import com.google.android.fhir.logicalId
+import com.google.android.fhir.pmapCPU
 import com.google.android.fhir.search.SearchQuery
 import com.google.android.fhir.toLocalChange
 import com.google.android.fhir.updateMeta
@@ -227,8 +229,11 @@ internal class DatabaseImpl(
     query: SearchQuery,
   ): List<ResourceWithUUID<R>> {
     return db.withTransaction {
-      resourceDao.getResources(SimpleSQLiteQuery(query.query, query.args.toTypedArray())).map {
-        ResourceWithUUID(it.uuid, iParser.parseResource(it.serializedResource) as R)
+      resourceDao.getResources(SimpleSQLiteQuery(query.query, query.args.toTypedArray())).pmapCPU {
+        ResourceWithUUID(
+          it.uuid,
+          FhirContext.forR4Cached().newJsonParser().parseResource(it.serializedResource) as R,
+        )
       }
     }
   }
@@ -239,11 +244,12 @@ internal class DatabaseImpl(
     return db.withTransaction {
       resourceDao
         .getForwardReferencedResources(SimpleSQLiteQuery(query.query, query.args.toTypedArray()))
-        .map {
+        .pmapCPU {
           ForwardIncludeSearchResult(
             it.matchingIndex,
             it.baseResourceUUID,
-            iParser.parseResource(it.serializedResource) as Resource,
+            FhirContext.forR4Cached().newJsonParser().parseResource(it.serializedResource)
+              as Resource,
           )
         }
     }
@@ -255,11 +261,12 @@ internal class DatabaseImpl(
     return db.withTransaction {
       resourceDao
         .getReverseReferencedResources(SimpleSQLiteQuery(query.query, query.args.toTypedArray()))
-        .map {
+        .pmapCPU {
           ReverseIncludeSearchResult(
             it.matchingIndex,
             it.baseResourceTypeAndId,
-            iParser.parseResource(it.serializedResource) as Resource,
+            FhirContext.forR4Cached().newJsonParser().parseResource(it.serializedResource)
+              as Resource,
           )
         }
     }

engine/src/test/java/com/google/android/fhir/UtilTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 Google LLC
+ * Copyright 2022-2024 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,9 @@ package com.google.android.fhir
 import android.os.Build
 import com.google.common.truth.Truth.assertThat
 import junit.framework.TestCase
+import kotlin.system.measureTimeMillis
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
 import org.hl7.fhir.r4.model.IdType
 import org.hl7.fhir.r4.model.OperationOutcome
 import org.hl7.fhir.r4.model.Patient
@@ -111,6 +114,24 @@ class UtilTest : TestCase() {
     assertThat(percentOf(25, 50)).isEqualTo(0.5)
   }
 
+  @Test
+  fun `pmap() should execute mapping tasks in parallel`() = runBlocking {
+    val numberList = listOf(2, 3)
+    var squaredNumberList: List<Int>
+
+    val timeTaken = measureTimeMillis {
+      squaredNumberList =
+        numberList.pmapCPU {
+          delay(1000L)
+          it * 2
+        }
+    }
+    assertTrue(squaredNumberList.isNotEmpty())
+    assertThat(squaredNumberList.first()).isEqualTo(4)
+    assertThat(squaredNumberList.last()).isEqualTo(6)
+    assertTrue(timeTaken < 2000)
+  }
+
   companion object {
     val TEST_OPERATION_OUTCOME_ERROR = OperationOutcome()
     val TEST_OPERATION_OUTCOME_INFO = OperationOutcome()