Remove MemoryDocument from result storage (#3201)

This PR removes the `MemoryDocument` class and its usage. Additionally,
it updates the fallback mechanism for `MongoDocument`, changing it from
`Memory` to `Iceberg`.

core/amber/src/main/scala/edu/uci/ics/texera/web/ComputingUnitMaster.scala

@@ -179,7 +179,7 @@ class ComputingUnitMaster extends io.dropwizard.Application[Configuration] with
         val storageType = collection.get("storageType").asText()
         val collectionName = collection.get("storageKey").asText()
         storageType match {
-          case OpResultStorage.MEMORY | OpResultStorage.ICEBERG =>
+          case OpResultStorage.ICEBERG =>
           // rely on the server-side result cleanup logic.
           case OpResultStorage.MONGODB =>
             MongoDatabaseManager.dropCollection(collectionName)

core/amber/src/main/scala/edu/uci/ics/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala

@@ -23,7 +23,6 @@ import edu.uci.ics.texera.dao.jooq.generated.tables.pojos.{
 import edu.uci.ics.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource._
 import edu.uci.ics.texera.web.service.ExecutionsMetadataPersistService
 import io.dropwizard.auth.Auth
-import org.jooq.impl.DSL._
 import org.jooq.types.{UInteger, ULong}
 
 import java.net.URI

core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/result/OpResultStorage.scala

@@ -20,7 +20,6 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala
   */
 object OpResultStorage {
   val defaultStorageMode: String = StorageConfig.resultStorageMode.toLowerCase
-  val MEMORY: String = "memory"
   val MONGODB: String = "mongodb"
   val ICEBERG = "iceberg"
 
@@ -115,9 +114,7 @@ class OpResultStorage extends Serializable with LazyLogging {
       schema: Schema
   ): VirtualDocument[Tuple] = {
     val storage: VirtualDocument[Tuple] =
-      if (mode == OpResultStorage.MEMORY) {
-        new MemoryDocument[Tuple](key)
-      } else if (mode == OpResultStorage.MONGODB) {
+      if (mode == OpResultStorage.MONGODB) {
         try {
           new MongoDocument[Tuple](
             executionId + key,
@@ -127,22 +124,13 @@ class OpResultStorage extends Serializable with LazyLogging {
         } catch {
           case t: Throwable =>
             logger.warn("Failed to create MongoDB storage", t)
-            logger.info(s"Falling back to memory storage for $key")
-            new MemoryDocument[Tuple](key)
+            logger.info(s"Falling back to Iceberg storage for $key")
+            createIcebergDocument(executionId, key, schema)
         }
+      } else if (mode == OpResultStorage.ICEBERG) {
+        createIcebergDocument(executionId, key, schema)
       } else {
-        val icebergSchema = IcebergUtil.toIcebergSchema(schema)
-        val serde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
-        val deserde: (IcebergSchema, Record) => Tuple = (_, record) =>
-          IcebergUtil.fromRecord(record, schema)
-
-        new IcebergDocument[Tuple](
-          StorageConfig.icebergTableNamespace,
-          executionId + key,
-          icebergSchema,
-          serde,
-          deserde
-        )
+        throw new IllegalArgumentException(s"Unsupported storage mode: $mode")
       }
     cache.put(key, (storage, schema))
     storage
@@ -172,4 +160,23 @@ class OpResultStorage extends Serializable with LazyLogging {
   def getAllKeys: Set[String] = {
     cache.keySet().iterator().asScala.toSet
   }
+
+  private def createIcebergDocument(
+      executionId: String,
+      key: String,
+      schema: Schema
+  ): IcebergDocument[Tuple] = {
+    val icebergSchema = IcebergUtil.toIcebergSchema(schema)
+    val serde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
+    val deserde: (IcebergSchema, Record) => Tuple = (_, record) =>
+      IcebergUtil.fromRecord(record, schema)
+
+    new IcebergDocument[Tuple](
+      StorageConfig.icebergTableNamespace,
+      executionId + key,
+      icebergSchema,
+      serde,
+      deserde
+    )
+  }
 }

core/workflow-core/src/test/scala/edu/uci/ics/amber/storage/result/iceberg/IcebergDocumentSpec.scala

@@ -5,7 +5,6 @@ import edu.uci.ics.amber.core.storage.model.VirtualDocumentSpec
 import edu.uci.ics.amber.core.storage.result.iceberg.IcebergDocument
 import edu.uci.ics.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
 import edu.uci.ics.amber.util.IcebergUtil
-import edu.uci.ics.texera.dao.MockTexeraDB
 import org.apache.iceberg.catalog.Catalog
 import org.apache.iceberg.data.Record
 import org.apache.iceberg.{Schema => IcebergSchema}