Add IcebergDocument as one implementation of VirtualDocument

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

@@ -183,6 +183,7 @@ class ComputingUnitMaster extends io.dropwizard.Application[Configuration] with
           // rely on the server-side result cleanup logic.
           case OpResultStorage.MONGODB =>
             MongoDatabaseManager.dropCollection(collectionName)
+          case OpResultStorage.ICEBERG =>
         }
       })
     } catch {

core/amber/src/main/scala/edu/uci/ics/texera/workflow/WorkflowCompiler.scala

@@ -93,7 +93,9 @@ class WorkflowCompiler(
               if (!storage.contains(storageKey)) {
                 // get the schema for result storage in certain mode
                 val sinkStorageSchema: Option[Schema] =
-                  if (storageType == OpResultStorage.MONGODB) {
+                  if (
+                    storageType == OpResultStorage.MONGODB || storageType == OpResultStorage.ICEBERG
+                  ) {
                     // use the output schema on the first output port as the schema for storage
                     Some(schema.right.get)
                   } else {

core/build.sbt

@@ -3,18 +3,43 @@ lazy val WorkflowCore = (project in file("workflow-core"))
   .dependsOn(DAO)
   .configs(Test)
   .dependsOn(DAO % "test->test") // test scope dependency
-lazy val WorkflowOperator = (project in file("workflow-operator")).dependsOn(WorkflowCore)
+lazy val WorkflowOperator = (project in file("workflow-operator"))
+  .dependsOn(WorkflowCore)
+  .settings(
+    dependencyOverrides ++= Seq(
+      "org.apache.commons" % "commons-compress" % "1.23.0", // because of the dependency introduced by iceberg
+    )
+  )
 lazy val WorkflowCompilingService = (project in file("workflow-compiling-service"))
   .dependsOn(WorkflowOperator)
   .settings(
     dependencyOverrides ++= Seq(
       // override it as io.dropwizard 4 require 2.16.1 or higher
-      "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.16.1"
+      "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.16.1",
+      "com.fasterxml.jackson.core" % "jackson-databind" % "2.16.1",
+      "org.glassfish.jersey.core" % "jersey-common" % "3.0.12"
     )
   )
 
 lazy val WorkflowExecutionService = (project in file("amber"))
   .dependsOn(WorkflowOperator)
+  .settings(
+    dependencyOverrides ++= Seq(
+      // override it as io.dropwizard 4 require 2.16.1 or higher
+      "com.fasterxml.jackson.core" % "jackson-core" % "2.15.1",
+      "com.fasterxml.jackson.core" % "jackson-databind" % "2.15.1",
+      "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.15.1",
+      "org.slf4j" % "slf4j-api" % "1.7.26",
+      "org.glassfish.jersey.media" % "jersey-media-multipart" % "2.25.1",
+      "org.glassfish.jersey.core" % "jersey-common" % "2.25.1",
+      "org.glassfish.jersey.core" % "jersey-server" % "2.25.1",
+      "javax.xml.bind" % "jaxb-api" % "2.3.1",
+      "com.sun.xml.bind" % "jaxb-impl" % "2.3.1",
+      "org.eclipse.jetty" % "jetty-server" % "9.4.20.v20190813",
+      "org.eclipse.jetty" % "jetty-servlet" % "9.4.20.v20190813",
+      "org.eclipse.jetty" % "jetty-http" % "9.4.20.v20190813",
+    )
+  )
   .configs(Test)
   .dependsOn(DAO % "test->test") // test scope dependency
 

core/workflow-core/build.sbt

@@ -111,6 +111,25 @@ val arrowDependencies = Seq(
 
 libraryDependencies ++= arrowDependencies
 
+val excludeHadoopJersey = ExclusionRule(organization = "com.sun.jersey")
+val excludeHadoopSlf4j = ExclusionRule(organization = "org.slf4j")
+val excludeHadoopJetty = ExclusionRule(organization = "org.eclipse.jetty")
+val excludeHadoopJsp = ExclusionRule(organization = "javax.servlet.jsp")
+libraryDependencies ++= Seq(
+  "org.apache.iceberg" % "iceberg-api" % "1.7.1",
+  "org.apache.iceberg" % "iceberg-core" % "1.7.1",
+  "org.apache.iceberg" % "iceberg-parquet" % "1.7.1",
+  "org.apache.iceberg" % "iceberg-data" % "1.7.1",
+  "org.apache.hadoop" % "hadoop-common" % "3.3.1" excludeAll(
+    ExclusionRule("javax.xml.bind"),
+    ExclusionRule("org.glassfish.jersey"),
+    excludeHadoopJersey,
+    excludeHadoopSlf4j,
+    excludeHadoopJetty,
+    excludeHadoopJsp,
+  )
+)
+
 /////////////////////////////////////////////////////////////////////////////
 // Additional Dependencies
 /////////////////////////////////////////////////////////////////////////////
@@ -123,5 +142,5 @@ libraryDependencies ++= Seq(
   "com.typesafe.scala-logging" %% "scala-logging" % "3.9.5",          // Scala Logging
   "org.eclipse.jgit" % "org.eclipse.jgit" % "5.13.0.202109080827-r",  // jgit
   "org.yaml" % "snakeyaml" % "1.30",                                  // yaml reader (downgrade to 1.30 due to dropwizard 1.3.23 required by amber)
-  "org.apache.commons" % "commons-vfs2" % "2.9.0"                     // for FileResolver throw VFS-related exceptions
+  "org.apache.commons" % "commons-vfs2" % "2.9.0",                     // for FileResolver throw VFS-related exceptions   // for Kyro serde/deserde
 )

core/workflow-core/src/main/resources/storage-config.yaml

@@ -1,10 +1,27 @@
 storage:
-  result-storage-mode: memory
+  result-storage-mode: iceberg
   mongodb:
     url: "mongodb://localhost:27017"
     database: "texera_storage"
     commit-batch-size: 1000
+  iceberg:
+    catalog:
+      jdbc: # currently we only support storing catalog info via jdbc, i.e. https://iceberg.apache.org/docs/1.7.1/jdbc/
+        url: "jdbc:mysql://0.0.0.0:3306/texera_iceberg?serverTimezone=UTC"
+        username: "root"
+        password: "123456"
+    table:
+      namespace: "operator-result"
+      commit:
+        batch-size: 4096 # decide the buffer size of our IcebergTableWriter
+        retry:
+          # retry configures the OCC parameter for concurrent write operations in Iceberg
+          # Docs about Reliability in Iceberg: https://iceberg.apache.org/docs/1.7.1/reliability/
+          # Docs about full parameter list and their meaning: https://iceberg.apache.org/docs/1.7.1/configuration/#write-properties
+          num-retries: 10
+          min-wait-ms: 100   # 0.1s
+          max-wait-ms: 10000 # 10s
   jdbc:
-    url: "jdbc:mysql://localhost:3306/texera_db?serverTimezone=UTC"
-    username: ""
-    password: ""
+    url: "jdbc:mysql://0.0.0.0:3306/texera_db?serverTimezone=UTC"
+    username: "root"
+    password: "123456"

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

@@ -1,7 +1,9 @@
 package edu.uci.ics.amber.core.storage
 
+import edu.uci.ics.amber.util.PathUtils.corePath
 import org.yaml.snakeyaml.Yaml
 
+import java.net.URI
 import java.util.{Map => JMap}
 import scala.jdk.CollectionConverters._
 
@@ -13,34 +15,123 @@ object StorageConfig {
 
     val storageMap = javaConf("storage").asInstanceOf[JMap[String, Any]].asScala.toMap
     val mongodbMap = storageMap("mongodb").asInstanceOf[JMap[String, Any]].asScala.toMap
+    val icebergMap = storageMap("iceberg").asInstanceOf[JMap[String, Any]].asScala.toMap
+    val icebergCatalogMap = icebergMap("catalog").asInstanceOf[JMap[String, Any]].asScala.toMap
+    val icebergCatalogJdbcMap =
+      icebergCatalogMap("jdbc").asInstanceOf[JMap[String, Any]].asScala.toMap
+    val icebergTableMap = icebergMap("table").asInstanceOf[JMap[String, Any]].asScala.toMap
+    val icebergCommitMap = icebergTableMap("commit").asInstanceOf[JMap[String, Any]].asScala.toMap
+    val icebergRetryMap = icebergCommitMap("retry").asInstanceOf[JMap[String, Any]].asScala.toMap
     val jdbcMap = storageMap("jdbc").asInstanceOf[JMap[String, Any]].asScala.toMap
-    javaConf
-      .updated("storage", storageMap.updated("mongodb", mongodbMap).updated("jdbc", jdbcMap))
+
+    javaConf.updated(
+      "storage",
+      storageMap
+        .updated("mongodb", mongodbMap)
+        .updated(
+          "iceberg",
+          icebergMap
+            .updated(
+              "catalog",
+              icebergCatalogMap.updated("jdbc", icebergCatalogJdbcMap)
+            )
+            .updated(
+              "table",
+              icebergTableMap.updated(
+                "commit",
+                icebergCommitMap.updated("retry", icebergRetryMap)
+              )
+            )
+        )
+        .updated("jdbc", jdbcMap)
+    )
   }
 
+  // Result storage mode
   val resultStorageMode: String =
     conf("storage").asInstanceOf[Map[String, Any]]("result-storage-mode").asInstanceOf[String]
 
-  // For MongoDB specifics
+  // MongoDB configurations
   val mongodbUrl: String = conf("storage")
     .asInstanceOf[Map[String, Any]]("mongodb")
     .asInstanceOf[Map[String, Any]]("url")
     .asInstanceOf[String]
+
   val mongodbDatabaseName: String = conf("storage")
     .asInstanceOf[Map[String, Any]]("mongodb")
     .asInstanceOf[Map[String, Any]]("database")
     .asInstanceOf[String]
+
   val mongodbBatchSize: Int = conf("storage")
     .asInstanceOf[Map[String, Any]]("mongodb")
     .asInstanceOf[Map[String, Any]]("commit-batch-size")
     .asInstanceOf[Int]
 
+  // Iceberg catalog configurations
+  val icebergCatalogUrl: String = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("catalog")
+    .asInstanceOf[Map[String, Any]]("jdbc")
+    .asInstanceOf[Map[String, Any]]("url")
+    .asInstanceOf[String]
+
+  val icebergCatalogUsername: String = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("catalog")
+    .asInstanceOf[Map[String, Any]]("jdbc")
+    .asInstanceOf[Map[String, Any]]("username")
+    .asInstanceOf[String]
+
+  val icebergCatalogPassword: String = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("catalog")
+    .asInstanceOf[Map[String, Any]]("jdbc")
+    .asInstanceOf[Map[String, Any]]("password")
+    .asInstanceOf[String]
+
+  val icebergTableNamespace: String = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("table")
+    .asInstanceOf[Map[String, Any]]("namespace")
+    .asInstanceOf[String]
+
+  val icebergTableCommitBatchSize: Int = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("table")
+    .asInstanceOf[Map[String, Any]]("commit")
+    .asInstanceOf[Map[String, Any]]("batch-size")
+    .asInstanceOf[Int]
+
+  val icebergTableCommitNumRetries: Int = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("table")
+    .asInstanceOf[Map[String, Any]]("commit")
+    .asInstanceOf[Map[String, Any]]("retry")
+    .asInstanceOf[Map[String, Any]]("num-retries")
+    .asInstanceOf[Int]
+
+  val icebergTableCommitMinRetryWaitMs: Int = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("table")
+    .asInstanceOf[Map[String, Any]]("commit")
+    .asInstanceOf[Map[String, Any]]("retry")
+    .asInstanceOf[Map[String, Any]]("min-wait-ms")
+    .asInstanceOf[Int]
+
+  val icebergTableCommitMaxRetryWaitMs: Int = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("table")
+    .asInstanceOf[Map[String, Any]]("commit")
+    .asInstanceOf[Map[String, Any]]("retry")
+    .asInstanceOf[Map[String, Any]]("max-wait-ms")
+    .asInstanceOf[Int]
+
+  // JDBC configurations
   val jdbcUrl: String = conf("storage")
     .asInstanceOf[Map[String, Any]]("jdbc")
     .asInstanceOf[Map[String, Any]]("url")
     .asInstanceOf[String]
 
-  // For jdbc specifics
   val jdbcUsername: String = conf("storage")
     .asInstanceOf[Map[String, Any]]("jdbc")
     .asInstanceOf[Map[String, Any]]("username")
@@ -50,4 +141,8 @@ object StorageConfig {
     .asInstanceOf[Map[String, Any]]("jdbc")
     .asInstanceOf[Map[String, Any]]("password")
     .asInstanceOf[String]
+
+  // File storage configurations
+  val fileStorageDirectoryUri: URI =
+    corePath.resolve("amber").resolve("user-resources").resolve("workflow-results").toUri
 }

core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/model/VirtualDocument.scala

@@ -65,9 +65,10 @@ abstract class VirtualDocument[T] extends ReadonlyVirtualDocument[T] {
 
   /**
     * return a writer that buffers the items and performs the flush operation at close time
+    * @param writerIdentifier the id of the writer, maybe required by some implementations
     * @return a buffered item writer
     */
-  def writer(): BufferedItemWriter[T] =
+  def writer(writerIdentifier: String): BufferedItemWriter[T] =
     throw new NotImplementedError("write method is not implemented")
 
   /**