Add support for switching to Iceberg RESTCatalog (#3225)

This PR adds the support for switching between Hadoop Catalog and REST
Catalog for Iceberg.

### How to switch to REST Catalog
1. change the `storage-config.yaml`, see Iceberg.catalog.type, change it
to "rest"
2. adjust iceberg.catalog.uri accordingly. 

### Tested Iceberg RESTCatalog
https://hub.docker.com/r/tabulario/iceberg-rest. This REST Catalog is
only for testing purpose. We need to test more production-level catalog
implementation.

Author: bobbai00

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

@@ -5,6 +5,9 @@ storage:
     database: "texera_storage"
     commit-batch-size: 1000
   iceberg:
+    catalog:
+      type: hadoop # either hadoop or rest
+      uri: http://localhost:8181 # the uri of the rest catalog
     table:
       namespace: "operator-result"
       commit:

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

@@ -16,18 +16,29 @@ object IcebergCatalogInstance {
   /**
     * Retrieves the singleton Iceberg catalog instance.
     * - If the catalog is not initialized, it is lazily created using the configured properties.
+    *
     * @return the Iceberg catalog instance.
     */
   def getInstance(): Catalog = {
     instance match {
       case Some(catalog) => catalog
       case None =>
-        val hadoopCatalog = IcebergUtil.createHadoopCatalog(
-          "texera_iceberg",
-          StorageConfig.fileStorageDirectoryPath
-        )
-        instance = Some(hadoopCatalog)
-        hadoopCatalog
+        val catalog = StorageConfig.icebergCatalogType match {
+          case "hadoop" =>
+            IcebergUtil.createHadoopCatalog(
+              "texera_iceberg",
+              StorageConfig.fileStorageDirectoryPath
+            )
+          case "rest" =>
+            IcebergUtil.createRestCatalog(
+              "texera_iceberg",
+              StorageConfig.fileStorageDirectoryPath
+            )
+          case unsupported =>
+            throw new IllegalArgumentException(s"Unsupported catalog type: $unsupported")
+        }
+        instance = Some(catalog)
+        catalog
     }
   }
 

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

@@ -16,6 +16,7 @@ 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 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
@@ -35,6 +36,7 @@ object StorageConfig {
                 icebergCommitMap.updated("retry", icebergRetryMap)
               )
             )
+            .updated("catalog", icebergCatalogMap)
         )
         .updated("jdbc", jdbcMap)
     )
@@ -97,6 +99,19 @@ object StorageConfig {
     .asInstanceOf[Map[String, Any]]("max-wait-ms")
     .asInstanceOf[Int]
 
+  // Iceberg catalog configurations
+  val icebergCatalogType: String = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("catalog")
+    .asInstanceOf[Map[String, Any]]("type")
+    .asInstanceOf[String]
+
+  val icebergCatalogUri: String = conf("storage")
+    .asInstanceOf[Map[String, Any]]("iceberg")
+    .asInstanceOf[Map[String, Any]]("catalog")
+    .asInstanceOf[Map[String, Any]]("uri")
+    .asInstanceOf[String]
+
   // JDBC configurations
   val jdbcUrl: String = conf("storage")
     .asInstanceOf[Map[String, Any]]("jdbc")

core/workflow-core/src/main/scala/edu/uci/ics/amber/util/IcebergUtil.scala

@@ -10,6 +10,7 @@ import org.apache.iceberg.data.{GenericRecord, Record}
 import org.apache.iceberg.hadoop.{HadoopCatalog, HadoopFileIO}
 import org.apache.iceberg.io.{CloseableIterable, InputFile}
 import org.apache.iceberg.parquet.{Parquet, ParquetValueReader}
+import org.apache.iceberg.rest.RESTCatalog
 import org.apache.iceberg.types.Type.PrimitiveType
 import org.apache.iceberg.{
   CatalogProperties,
@@ -60,6 +61,38 @@ object IcebergUtil {
     catalog
   }
 
+  /**
+    * Creates and initializes a RESTCatalog with the given parameters.
+    * - Configures the catalog to interact with a REST endpoint.
+    * - The `warehouse` parameter specifies the root directory for storing table data.
+    * - Sets the file I/O implementation to `HadoopFileIO`.
+    * - Authentication support is not implemented yet (see TODO).
+    *
+    * Note: The only tested REST catalog implementation is `tabulario/iceberg-rest`
+    * (https://hub.docker.com/r/tabulario/iceberg-rest).
+    *
+    * TODO: Add authentication support, such as OAuth2, using `OAuth2Properties`.
+    *
+    * @param catalogName the name of the catalog.
+    * @param warehouse   the root path for the warehouse where the tables are stored.
+    * @return the initialized RESTCatalog instance.
+    */
+  def createRestCatalog(
+      catalogName: String,
+      warehouse: Path
+  ): RESTCatalog = {
+    val catalog = new RESTCatalog()
+    catalog.initialize(
+      catalogName,
+      Map(
+        "warehouse" -> warehouse.toString,
+        CatalogProperties.URI -> StorageConfig.icebergCatalogUri,
+        CatalogProperties.FILE_IO_IMPL -> classOf[HadoopFileIO].getName
+      ).asJava
+    )
+    catalog
+  }
+
   /**
     * Creates a new Iceberg table with the specified schema and properties.
     * - Drops the existing table if `overrideIfExists` is true and the table already exists.

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

@@ -1,11 +1,6 @@
 package edu.uci.ics.amber.storage.result.iceberg
 
-import edu.uci.ics.amber.core.storage.{
-  DocumentFactory,
-  IcebergCatalogInstance,
-  StorageConfig,
-  VFSURIFactory
-}
+import edu.uci.ics.amber.core.storage.{DocumentFactory, VFSURIFactory}
 import edu.uci.ics.amber.core.storage.model.{VirtualDocument, VirtualDocumentSpec}
 import edu.uci.ics.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
 import edu.uci.ics.amber.core.virtualidentity.{
@@ -56,13 +51,6 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
     // Initialize serialization and deserialization functions
     serde = IcebergUtil.toGenericRecord
     deserde = (schema, record) => IcebergUtil.fromRecord(record, amberSchema)
-
-    // Initialize the the Iceberg catalog
-    catalog = IcebergUtil.createHadoopCatalog(
-      "iceberg_document_test",
-      StorageConfig.fileStorageDirectoryPath
-    )
-    IcebergCatalogInstance.replaceInstance(catalog)
   }
 
   override def beforeEach(): Unit = {