feat(storage): bound the per-warehouse catalog cache and release evicted catalogs

amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py

@@ -35,6 +35,11 @@ class IcebergCatalogInstance:
       testing or reconfiguration.
     """
 
+    # Deliberately unbounded, unlike the Scala side's bounded cache (#7290): a Python
+    # worker process is spawned per worker and destroyed when the execution ends
+    # (PythonWorkflowWorker.postStop), so this dict holds the single warehouse that
+    # execution used and dies with the process. If PVMs ever become long-lived (e.g.
+    # pooled across executions), this needs the same bounding the JVM singleton has.
     _catalogs: dict = {}
     _POSTGRES_KEY = "__postgres__"
 

amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala

@@ -24,7 +24,6 @@ import org.apache.texera.amber.core.state.State
 import org.apache.texera.amber.core.storage.model.{VirtualDocument, VirtualDocumentSpec}
 import org.apache.texera.amber.core.storage.{DocumentFactory, IcebergCatalogInstance, VFSURIFactory}
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
-import org.apache.iceberg.Table
 import org.apache.texera.amber.core.virtualidentity.{
   ExecutionIdentity,
   OperatorIdentity,
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
     val (batch1, batch2) = items.splitAt(batchSize)
 
     // Write two separate batches to produce two committed data files.
-    // This also initialises `document.catalog` (lazy val) with the real catalog, which
-    // is why we open a fresh reader document below after injecting the spy.
     val writer1 = document.writer(UUID.randomUUID().toString)
     writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
 
     val writer2 = document.writer(UUID.randomUUID().toString)
     writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
 
-    val refreshCount = new AtomicInteger(0)
+    val metadataLoadCount = new AtomicInteger(0)
     val realCatalog = IcebergCatalogInstance.getInstance()
-    IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
-    // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
-    // will pick up the spy catalog on first access inside seekToUsableFile.
+    IcebergCatalogInstance.replaceInstance(
+      catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+    )
+    // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+    // load inside seekToUsableFile goes through the spy installed above.
     val readerDoc = getDocument
     try {
       val retrieved = readerDoc.get().toList
       assert(
         retrieved.toSet == items.toSet,
         "All records from both files should be read correctly"
       )
-      // With lazy file advancement seekToUsableFile() (and therefore table.refresh()) is called:
-      //   once on iterator creation, once when the last file is exhausted → 2 total.
-      // Without the fix it would be called once per hasNext() on the last file → O(batchSize).
+      // With lazy file advancement the table is (re-)resolved once at iterator
+      // construction plus once per seekToUsableFile — construction seek and the
+      // final exhausted-files seek → 3 total. Without lazy advancement it would be
+      // once per hasNext() on the last file → O(batchSize).
       assert(
-        refreshCount.get() <= 4,
-        s"table.refresh() should be called at most 4 times (lazy advancement), but was ${refreshCount.get()}"
+        metadataLoadCount.get() <= 4,
+        s"the table should be loaded at most 4 times (lazy advancement), but was ${metadataLoadCount.get()}"
       )
     } finally {
       IcebergCatalogInstance.replaceInstance(realCatalog)
@@ -309,35 +309,18 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
     }
   }
 
-  /** Returns a dynamic proxy for `realTable` that increments `counter` on every `refresh()` call. */
-  private def tableWithRefreshSpy(realTable: Table, counter: AtomicInteger): Table =
-    Proxy
-      .newProxyInstance(
-        classOf[Table].getClassLoader,
-        Array(classOf[Table]),
-        new InvocationHandler {
-          override def invoke(proxy: Object, method: Method, args: Array[Object]): Object = {
-            if (method.getName == "refresh") counter.incrementAndGet()
-            if (args == null) method.invoke(realTable) else method.invoke(realTable, args: _*)
-          }
-        }
-      )
-      .asInstanceOf[Table]
-
-  /** Returns a dynamic proxy for `realCatalog` that wraps every loaded `Table` with a refresh spy. */
-  private def catalogWithRefreshSpy(realCatalog: Catalog, counter: AtomicInteger): Catalog =
+  /** Returns a dynamic proxy for `realCatalog` that counts `loadTable` calls. */
+  private def catalogWithMetadataLoadSpy(realCatalog: Catalog, counter: AtomicInteger): Catalog =
     Proxy
       .newProxyInstance(
         classOf[Catalog].getClassLoader,
         Array(classOf[Catalog]),
         new InvocationHandler {
           override def invoke(proxy: Object, method: Method, args: Array[Object]): Object = {
-            val result =
-              if (args == null) method.invoke(realCatalog) else method.invoke(realCatalog, args: _*)
-            if (method.getName == "loadTable" && result != null)
-              tableWithRefreshSpy(result.asInstanceOf[Table], counter)
-            else
-              result
+            // The reader re-resolves its table per seek (#7290) instead of refreshing a
+            // pinned one, so metadata loads now surface as `loadTable` calls here.
+            if (method.getName == "loadTable") counter.incrementAndGet()
+            if (args == null) method.invoke(realCatalog) else method.invoke(realCatalog, args: _*)
           }
         }
       )

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala

@@ -66,6 +66,30 @@ object DocumentFactory {
     }
   }
 
+  /**
+    * The iceberg coordinates a VFS URI resolves to: which warehouse's catalog, which
+    * namespace, and which table (storage key). One resolver shared by every VFS entry
+    * point below, so the decode steps cannot drift apart (promised in #6944 review).
+    */
+  private case class IcebergLocation(
+      warehouse: Option[String],
+      namespace: String,
+      storageKey: String
+  )
+
+  private def resolveIcebergLocation(uri: URI): IcebergLocation = {
+    val components = decodeURI(uri)
+    IcebergLocation(
+      components.warehouse,
+      resolveNamespace(components.resourceType),
+      sanitizeURIPath(uri)
+    )
+  }
+
+  private val tupleSerde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
+  private val tupleDeserde: (IcebergSchema, Record) => Tuple = (schema, record) =>
+    IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema))
+
   /**
     * Create a document for storage specified by the uri.
     * This document is suitable for storing structural data, i.e. the schema is required to create such document.
@@ -76,11 +100,7 @@ object DocumentFactory {
   def createDocument(uri: URI, schema: Schema): VirtualDocument[_] = {
     uri.getScheme match {
       case VFS_FILE_URI_SCHEME =>
-        val components = decodeURI(uri)
-        val warehouse = components.warehouse
-        val resourceType = components.resourceType
-        val storageKey = sanitizeURIPath(uri)
-        val namespace = resolveNamespace(resourceType)
+        val IcebergLocation(warehouse, namespace, storageKey) = resolveIcebergLocation(uri)
 
         val icebergSchema = IcebergUtil.toIcebergSchema(schema)
         IcebergUtil.createTable(
@@ -90,16 +110,12 @@ object DocumentFactory {
           icebergSchema,
           overrideIfExists = true
         )
-        val serde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
-        val deserde: (IcebergSchema, Record) => Tuple = (schema, record) =>
-          IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema))
-
         new IcebergDocument[Tuple](
           namespace,
           storageKey,
           icebergSchema,
-          serde,
-          deserde,
+          tupleSerde,
+          tupleDeserde,
           warehouse
         )
       case unsupportedScheme =>
@@ -122,11 +138,7 @@ object DocumentFactory {
   def documentExists(uri: URI): Boolean = {
     uri.getScheme match {
       case VFS_FILE_URI_SCHEME =>
-        val components = decodeURI(uri)
-        val warehouse = components.warehouse
-        val resourceType = components.resourceType
-        val storageKey = sanitizeURIPath(uri)
-        val namespace = resolveNamespace(resourceType)
+        val IcebergLocation(warehouse, namespace, storageKey) = resolveIcebergLocation(uri)
         IcebergCatalogInstance
           .getInstance(warehouse)
           .tableExists(TableIdentifier.of(namespace, storageKey))
@@ -170,11 +182,7 @@ object DocumentFactory {
     uri.getScheme match {
       case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None)
       case VFS_FILE_URI_SCHEME =>
-        val components = decodeURI(uri)
-        val warehouse = components.warehouse
-        val resourceType = components.resourceType
-        val storageKey = sanitizeURIPath(uri)
-        val namespace = resolveNamespace(resourceType)
+        val IcebergLocation(warehouse, namespace, storageKey) = resolveIcebergLocation(uri)
 
         val table = IcebergUtil
           .loadTableMetadata(
@@ -187,17 +195,13 @@ object DocumentFactory {
           )
 
         val amberSchema = IcebergUtil.fromIcebergSchema(table.schema())
-        val serde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
-        val deserde: (IcebergSchema, Record) => Tuple = (schema, record) =>
-          IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema))
-
         (
           new IcebergDocument[Tuple](
             namespace,
             storageKey,
             table.schema(),
-            serde,
-            deserde,
+            tupleSerde,
+            tupleDeserde,
             warehouse
           ),
           Some(amberSchema)

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala

@@ -19,11 +19,23 @@
 
 package org.apache.texera.amber.core.storage
 
+import com.google.common.base.Ticker
+import com.google.common.cache.{
+  Cache,
+  CacheBuilder,
+  RemovalCause,
+  RemovalListener,
+  RemovalNotification
+}
+import com.google.common.util.concurrent.{ExecutionError, UncheckedExecutionException}
+import com.typesafe.scalalogging.LazyLogging
 import org.apache.texera.common.config.StorageConfig
 import org.apache.texera.amber.util.IcebergUtil
 import org.apache.iceberg.catalog.Catalog
 
-import scala.collection.mutable
+import java.time.Duration
+import java.util.concurrent.{Callable, ExecutionException}
+import scala.util.Try
 
 /**
   * IcebergCatalogInstance manages the Iceberg catalog clients used across the Texera application.
@@ -36,11 +48,65 @@ import scala.collection.mutable
   * Only the REST catalog varies by warehouse; the hadoop and postgres catalogs are warehouse-agnostic
   * and ignore the warehouse argument.
   *
-  * Access is synchronized because the same JVM serves multiple warehouses concurrently.
+  * The cache is bounded (#7290): per-user warehouses (#6870) make the set of catalogs a
+  * long-lived JVM touches unbounded, and each REST catalog holds an HTTP client. An entry
+  * idle for the expiry window is closed -- nothing can be using it, and idle entries are
+  * exactly what a long-lived JVM accumulates. An entry evicted by *size* is only dropped,
+  * never closed: size pressure means more simultaneously hot warehouses than the bound,
+  * and closing a hot catalog would fail the operations still using it. Load degrades into
+  * rebuild churn (the dropped catalog decays once its in-flight users finish), not errors.
+  *
+  * Callers must therefore resolve their catalog per logical operation instead of holding
+  * one across an execution -- that is also what keeps a dropped catalog's lifetime bounded
+  * by the operation using it (see IcebergDocument / IcebergTableWriter).
+  *
+  * Only *evicted* entries are closed. A catalog displaced by [[replaceInstance]] is the
+  * caller's to manage: whoever replaces an entry may still hold (and restore) the old
+  * reference -- tests wrap-and-restore the shared catalog, and endpoint reconfiguration
+  * (#7358) will swap catalogs the same way.
   */
-object IcebergCatalogInstance {
+object IcebergCatalogInstance extends LazyLogging {
+
+  // Sizing mirrors HuggingFaceModelResource's bounded-cache precedent: generous enough
+  // that eviction never hits a warehouse in active use, small enough to bound the JVM.
+  private val CatalogCacheMaxSize = 64L
+  private val CatalogCacheExpireAfterAccess = Duration.ofMinutes(60)
+
+  /**
+    * Builds a catalog cache with the eviction wiring `getInstance` relies on.
+    * Package-private so the spec can exercise size and idle eviction on isolated
+    * instances with a manual ticker, instead of flooding the JVM-wide cache below.
+    */
+  private[storage] def buildCatalogCache(
+      maximumSize: Long,
+      expireAfterAccess: Duration,
+      ticker: Ticker
+  ): Cache[String, Catalog] =
+    CacheBuilder
+      .newBuilder()
+      .maximumSize(maximumSize)
+      .expireAfterAccess(expireAfterAccess)
+      .ticker(ticker)
+      .removalListener(new RemovalListener[String, Catalog] {
+        override def onRemoval(notification: RemovalNotification[String, Catalog]): Unit =
+          // Close ONLY idle-expired entries. A size-evicted catalog may be mid-operation
+          // (overload = more hot warehouses than the bound) and a replaced one is still
+          // the replacing caller's (wrap-and-restore in tests, reconfiguration later);
+          // both are dropped un-closed and decay once their last user finishes.
+          if (notification.getCause == RemovalCause.EXPIRED) {
+            notification.getValue match {
+              case closeable: AutoCloseable =>
+                Try(closeable.close()).failed.foreach(error =>
+                  logger.warn(s"failed to close expired catalog '${notification.getKey}'", error)
+                )
+              case _ =>
+            }
+          }
+      })
+      .build[String, Catalog]()
 
-  private val catalogs = mutable.Map.empty[String, Catalog]
+  private val catalogs: Cache[String, Catalog] =
+    buildCatalogCache(CatalogCacheMaxSize, CatalogCacheExpireAfterAccess, Ticker.systemTicker())
 
   // Cache key for the warehouse-agnostic catalog types. Not a legal warehouse name,
   // so it cannot collide with a REST warehouse.
@@ -70,11 +136,35 @@ object IcebergCatalogInstance {
     */
   def getInstance(warehouse: Option[String] = None): Catalog = {
     val name = warehouse.getOrElse(defaultWarehouse)
-    synchronized {
-      catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
-    }
+    getOrLoad(catalogs, cacheKey(name), () => createCatalog(name))
   }
 
+  /**
+    * `Cache.get` wraps loader failures (`UncheckedExecutionException`, `ExecutionException`,
+    * `ExecutionError`); unwrap them so `createCatalog` failures keep the types they had
+    * before the cache existed. Package-private so the spec can pin the unwrapping against
+    * an isolated cache with a throwing loader.
+    */
+  private[storage] def getOrLoad(
+      cache: Cache[String, Catalog],
+      key: String,
+      loader: () => Catalog
+  ): Catalog =
+    try {
+      // get(key, loader) locks per key, not globally: a cache miss's REST config
+      // round trip no longer blocks lookups of other warehouses.
+      cache.get(
+        key,
+        new Callable[Catalog] {
+          override def call(): Catalog = loader()
+        }
+      )
+    } catch {
+      case e: UncheckedExecutionException => throw e.getCause
+      case e: ExecutionException          => throw e.getCause
+      case e: ExecutionError              => throw e.getCause
+    }
+
   private def createCatalog(warehouse: String): Catalog =
     StorageConfig.icebergCatalogType match {
       case "hadoop" =>
@@ -103,7 +193,5 @@ object IcebergCatalogInstance {
     * @param warehouse the warehouse to cache it under; `None` uses the configured default.
     */
   def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit =
-    synchronized {
-      catalogs(cacheKey(warehouse.getOrElse(defaultWarehouse))) = catalog
-    }
+    catalogs.put(cacheKey(warehouse.getOrElse(defaultWarehouse)), catalog)
 }