From 452a8e6093ca1985992fc29b3a5c54aca12e1d6a Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Tue, 14 Jul 2026 15:42:52 -0700 Subject: [PATCH 1/9] feat(storage): per-warehouse Iceberg catalog and warehouse-scoped URIs Replace the single-warehouse catalog singleton with a per-warehouse cache on both the JVM and Python sides, and encode the warehouse as a leading /wh/ segment in VFS URIs so a URI fully identifies where its tables live. The read path derives the warehouse from the URI; non-BYO URIs omit the segment and fall back to the configured warehouse, leaving existing behavior unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../python/core/storage/document_factory.py | 19 +++- .../iceberg/iceberg_catalog_instance.py | 62 +++++++------ .../core/storage/iceberg/iceberg_document.py | 3 +- .../python/core/storage/vfs_uri_factory.py | 14 +++ .../amber/core/storage/DocumentFactory.scala | 28 ++++-- .../core/storage/IcebergCatalogInstance.scala | 90 ++++++++++--------- .../amber/core/storage/VFSURIFactory.scala | 39 ++++++-- .../result/iceberg/IcebergDocument.scala | 7 +- .../amber/core/workflow/WorkflowContext.scala | 3 +- 9 files changed, 175 insertions(+), 90 deletions(-) diff --git a/amber/src/main/python/core/storage/document_factory.py b/amber/src/main/python/core/storage/document_factory.py index 5078ca7dd54..7e33fed3c45 100644 --- a/amber/src/main/python/core/storage/document_factory.py +++ b/amber/src/main/python/core/storage/document_factory.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations # under the License. +import re import typing import urllib from typing import Optional @@ -67,7 +68,12 @@ def sanitize_uri_path(uri): :param uri: Result of urllib.parse.urlparse(). Could be quoted. :return: Unquoted and sanitized format of uri. """ - return urllib.parse.unquote(uri.path).lstrip("/").replace("/", "_") + path = urllib.parse.unquote(uri.path).lstrip("/") + # Strip the optional leading "wh//" segment so the storage key is + # identical regardless of which warehouse the URI targets (matches + # DocumentFactory.scala). + path = re.sub(r"^wh/[^/]+/", "", path) + return path.replace("/", "_") @staticmethod def create_document(uri: str, schema: Schema) -> VirtualDocument: @@ -76,12 +82,13 @@ def create_document(uri: str, schema: Schema) -> VirtualDocument: _, _, _, resource_type = VFSURIFactory.decode_uri(uri) namespace = DocumentFactory._resolve_namespace(resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) + warehouse = VFSURIFactory.warehouse_from_uri(uri) # Convert Amber Schema to Iceberg Schema with LARGE_BINARY # field name encoding iceberg_schema = amber_schema_to_iceberg_schema(schema) create_table( - IcebergCatalogInstance.get_instance(), + IcebergCatalogInstance.get_instance(warehouse), namespace, storage_key, iceberg_schema, @@ -94,6 +101,7 @@ def create_document(uri: str, schema: Schema) -> VirtualDocument: iceberg_schema, amber_tuples_to_arrow_table, arrow_table_to_amber_tuples, + warehouse, ) else: @@ -119,7 +127,8 @@ def document_exists(uri: str) -> bool: _, _, _, resource_type = VFSURIFactory.decode_uri(uri) namespace = DocumentFactory._resolve_namespace(resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) - return IcebergCatalogInstance.get_instance().table_exists( + warehouse = VFSURIFactory.warehouse_from_uri(uri) + return IcebergCatalogInstance.get_instance(warehouse).table_exists( f"{namespace}.{storage_key}" ) @@ -134,9 +143,10 @@ def open_document(uri: str) -> typing.Tuple[VirtualDocument, Optional[Schema]]: _, _, _, resource_type = VFSURIFactory.decode_uri(uri) namespace = DocumentFactory._resolve_namespace(resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) + warehouse = VFSURIFactory.warehouse_from_uri(uri) table = load_table_metadata( - IcebergCatalogInstance.get_instance(), + IcebergCatalogInstance.get_instance(warehouse), namespace, storage_key, ) @@ -152,6 +162,7 @@ def open_document(uri: str) -> typing.Tuple[VirtualDocument, Optional[Schema]]: table.schema(), amber_tuples_to_arrow_table, arrow_table_to_amber_tuples, + warehouse, ) return document, amber_schema diff --git a/amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py b/amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py index 65a4f6beec3..61987c2594d 100644 --- a/amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py +++ b/amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py @@ -27,49 +27,59 @@ class IcebergCatalogInstance: """ - IcebergCatalogInstance is a singleton that manages the Iceberg catalog instance. - Supports postgres SQL catalog and REST catalog. - - Provides a single shared catalog for all Iceberg table-related operations. - - Lazily initializes the catalog on first access. - - Supports replacing the catalog instance for testing or reconfiguration. + Manages Iceberg catalog instances, cached per warehouse. + - REST catalogs are keyed by warehouse so one process can read/write tables in + many warehouses (Design 2, warehouse-per-execution). + - The postgres catalog has no warehouse concept and shares a single entry. + - Catalogs are lazily created on first access; entries can be replaced for + testing or reconfiguration. """ - _instance: Optional[Catalog] = None + _catalogs: dict = {} + _POSTGRES_KEY = "__postgres__" @classmethod - def get_instance(cls): + def get_instance(cls, warehouse: Optional[str] = None) -> Catalog: """ - Retrieves the singleton Iceberg catalog instance. - - If the catalog is not initialized, it is lazily created using the configured - properties. - - Supports "postgres" and "rest" catalog types. + Retrieves the Iceberg catalog for the given warehouse, creating and caching + it on first use. For REST catalogs, `warehouse` selects which warehouse's + catalog to use (defaults to the configured warehouse when None). For the + postgres catalog, `warehouse` is ignored. + :param warehouse: the warehouse name (REST only); None uses the default. :return: the Iceberg catalog instance. """ - if cls._instance is None: - catalog_type = StorageConfig.ICEBERG_CATALOG_TYPE - if catalog_type == "postgres": - cls._instance = create_postgres_catalog( + catalog_type = StorageConfig.ICEBERG_CATALOG_TYPE + if catalog_type == "postgres": + if cls._POSTGRES_KEY not in cls._catalogs: + cls._catalogs[cls._POSTGRES_KEY] = create_postgres_catalog( "texera_iceberg", StorageConfig.ICEBERG_FILE_STORAGE_DIRECTORY_PATH, StorageConfig.ICEBERG_POSTGRES_CATALOG_URI_WITHOUT_SCHEME, StorageConfig.ICEBERG_POSTGRES_CATALOG_USERNAME, StorageConfig.ICEBERG_POSTGRES_CATALOG_PASSWORD, ) - elif catalog_type == "rest": - cls._instance = create_rest_catalog( + return cls._catalogs[cls._POSTGRES_KEY] + elif catalog_type == "rest": + key = warehouse or StorageConfig.ICEBERG_REST_CATALOG_WAREHOUSE_NAME + if key not in cls._catalogs: + cls._catalogs[key] = create_rest_catalog( "texera_iceberg", - StorageConfig.ICEBERG_REST_CATALOG_WAREHOUSE_NAME, + key, StorageConfig.ICEBERG_REST_CATALOG_URI, ) - else: - raise ValueError(f"Unsupported catalog type: {catalog_type}") - return cls._instance + return cls._catalogs[key] + else: + raise ValueError(f"Unsupported catalog type: {catalog_type}") @classmethod - def replace_instance(cls, catalog: Catalog): + def replace_instance(cls, catalog: Catalog, warehouse: Optional[str] = None): """ - Replaces the existing Iceberg catalog instance. - - This method is useful for testing or dynamically updating the catalog. - :param catalog: the new Iceberg catalog instance to replace the current one. + Replaces the cached catalog for a warehouse (testing or reconfiguration). + :param catalog: the new Iceberg catalog instance. + :param warehouse: the warehouse to replace (REST only); None uses default. """ - cls._instance = catalog + if StorageConfig.ICEBERG_CATALOG_TYPE == "postgres": + key = cls._POSTGRES_KEY + else: + key = warehouse or StorageConfig.ICEBERG_REST_CATALOG_WAREHOUSE_NAME + cls._catalogs[key] = catalog diff --git a/amber/src/main/python/core/storage/iceberg/iceberg_document.py b/amber/src/main/python/core/storage/iceberg/iceberg_document.py index 7a5beda916d..797c1ec4e4d 100644 --- a/amber/src/main/python/core/storage/iceberg/iceberg_document.py +++ b/amber/src/main/python/core/storage/iceberg/iceberg_document.py @@ -63,6 +63,7 @@ def __init__( table_schema: Schema, serde: Callable[[Schema, Iterable[T]], pa.Table], deserde: Callable[[Schema, pa.Table], Iterable[T]], + warehouse: Optional[str] = None, ): self.table_namespace = table_namespace self.table_name = table_name @@ -71,7 +72,7 @@ def __init__( self.deserde = deserde self.lock = rwlock.RWLockFair() - self.catalog = IcebergCatalogInstance.get_instance() + self.catalog = IcebergCatalogInstance.get_instance(warehouse) def get_uri(self) -> ParseResult: """Returns the URI of the table location.""" diff --git a/amber/src/main/python/core/storage/vfs_uri_factory.py b/amber/src/main/python/core/storage/vfs_uri_factory.py index 883450abf2b..28733dfbbda 100644 --- a/amber/src/main/python/core/storage/vfs_uri_factory.py +++ b/amber/src/main/python/core/storage/vfs_uri_factory.py @@ -88,6 +88,20 @@ def extract_value(key: str) -> str: resource_type, ) + @staticmethod + def warehouse_from_uri(uri: str) -> Optional[str]: + """ + Extracts the warehouse name from the optional leading "/wh/" segment + of a VFS URI, or None if absent. Mirrors VFSURIFactory.warehouseFromURI + (Scala) so a URI fully identifies which warehouse its tables live in. + """ + segments = urlparse(uri).path.lstrip("/").split("/") + try: + idx = segments.index("wh") + except ValueError: + return None + return segments[idx + 1] if idx + 1 < len(segments) else None + @staticmethod def create_port_base_uri(workflow_id, execution_id, global_port_id) -> str: """Base URI for a port. Result and state URIs derive from it via diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala index a26340e79cc..afe3e7a6317 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala @@ -22,7 +22,11 @@ package org.apache.texera.amber.core.storage import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.core.storage.FileResolver.DATASET_FILE_URI_SCHEME import org.apache.texera.amber.core.storage.VFSResourceType._ -import org.apache.texera.amber.core.storage.VFSURIFactory.{VFS_FILE_URI_SCHEME, decodeURI} +import org.apache.texera.amber.core.storage.VFSURIFactory.{ + VFS_FILE_URI_SCHEME, + decodeURI, + warehouseFromURI +} import org.apache.texera.amber.core.storage.model._ import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument import org.apache.texera.amber.core.tuple.{Schema, Tuple} @@ -38,7 +42,12 @@ object DocumentFactory { val ICEBERG = "iceberg" private def sanitizeURIPath(uri: URI): String = - uri.getPath.stripPrefix("/").replace("/", "_") + uri.getPath.stripPrefix("/").replaceFirst("^wh/[^/]+/", "").replace("/", "_") + + // The warehouse a URI's table lives in, encoded as a leading `/wh/` path + // segment; absent for non-BYO storage, which falls back to the configured warehouse. + private def warehouseOf(uri: URI): String = + warehouseFromURI(uri).getOrElse(StorageConfig.icebergRESTCatalogWarehouseName) private def resolveNamespace(resourceType: VFSResourceType.Value): String = resourceType match { @@ -74,6 +83,7 @@ object DocumentFactory { * @return the created document */ def createDocument(uri: URI, schema: Schema): VirtualDocument[_] = { + val warehouse = warehouseOf(uri) uri.getScheme match { case VFS_FILE_URI_SCHEME => val (_, _, _, resourceType) = decodeURI(uri) @@ -82,7 +92,7 @@ object DocumentFactory { val icebergSchema = IcebergUtil.toIcebergSchema(schema) IcebergUtil.createTable( - IcebergCatalogInstance.getInstance(), + IcebergCatalogInstance.getInstance(warehouse), namespace, storageKey, icebergSchema, @@ -97,7 +107,8 @@ object DocumentFactory { storageKey, icebergSchema, serde, - deserde + deserde, + warehouse ) case unsupportedScheme => throw new UnsupportedOperationException( @@ -117,13 +128,14 @@ object DocumentFactory { * iceberg namespace mapping. */ def documentExists(uri: URI): Boolean = { + val warehouse = warehouseOf(uri) uri.getScheme match { case VFS_FILE_URI_SCHEME => val (_, _, _, resourceType) = decodeURI(uri) val storageKey = sanitizeURIPath(uri) val namespace = resolveNamespace(resourceType) IcebergCatalogInstance - .getInstance() + .getInstance(warehouse) .tableExists(TableIdentifier.of(namespace, storageKey)) case unsupportedScheme => @@ -162,6 +174,7 @@ object DocumentFactory { * @return the VirtualDocument, which is the handler of the data; the Schema, which is the schema of the data stored in the document */ def openDocument(uri: URI): (VirtualDocument[_], Option[Schema]) = { + val warehouse = warehouseOf(uri) uri.getScheme match { case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None) case VFS_FILE_URI_SCHEME => @@ -171,7 +184,7 @@ object DocumentFactory { val table = IcebergUtil .loadTableMetadata( - IcebergCatalogInstance.getInstance(), + IcebergCatalogInstance.getInstance(warehouse), namespace, storageKey ) @@ -190,7 +203,8 @@ object DocumentFactory { storageKey, table.schema(), serde, - deserde + deserde, + warehouse ), Some(amberSchema) ) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala index cd4b3c8796f..30b2ef749c0 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala @@ -23,57 +23,67 @@ import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.util.IcebergUtil import org.apache.iceberg.catalog.Catalog +import scala.collection.mutable + /** - * IcebergCatalogInstance is a singleton that manages the Iceberg catalog instance. - * - Provides a single shared catalog for all Iceberg table-related operations in the Texera application. - * - Lazily initializes the catalog on first access. - * - Supports replacing the catalog instance primarily for testing or reconfiguration. + * IcebergCatalogInstance manages the Iceberg catalog clients used across the Texera application. + * + * Catalogs are cached per warehouse: each distinct warehouse name gets its own lazily-created + * catalog client, so a single JVM may hold several catalogs, one per warehouse it touches. Callers + * that do not specify a warehouse use the configured default, preserving single-warehouse (non-BYO) + * behavior. + * + * 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. */ object IcebergCatalogInstance { - private var instance: Option[Catalog] = None + private val catalogs = mutable.Map.empty[String, Catalog] + + private def defaultWarehouse: String = StorageConfig.icebergRESTCatalogWarehouseName /** - * Retrieves the singleton Iceberg catalog instance. - * - If the catalog is not initialized, it is lazily created using the configured properties. + * Retrieves the catalog for the given warehouse, creating and caching it on first access. * - * @return the Iceberg catalog instance. + * @param warehouse the warehouse to obtain a catalog for; defaults to the configured warehouse. + * @return the Iceberg catalog for that warehouse. */ - def getInstance(): Catalog = { - instance match { - case Some(catalog) => catalog - case None => - val catalog = StorageConfig.icebergCatalogType match { - case "hadoop" => - IcebergUtil.createHadoopCatalog( - "texera_iceberg", - StorageConfig.fileStorageDirectoryPath - ) - case "rest" => - IcebergUtil.createRestCatalog( - "texera_iceberg", - StorageConfig.icebergRESTCatalogWarehouseName - ) - case "postgres" => - IcebergUtil.createPostgresCatalog( - "texera_iceberg", - StorageConfig.fileStorageDirectoryPath - ) - case unsupported => - throw new IllegalArgumentException(s"Unsupported catalog type: $unsupported") - } - instance = Some(catalog) - catalog + def getInstance(warehouse: String = defaultWarehouse): Catalog = + synchronized { + catalogs.getOrElseUpdate(warehouse, createCatalog(warehouse)) + } + + private def createCatalog(warehouse: String): Catalog = + StorageConfig.icebergCatalogType match { + case "hadoop" => + IcebergUtil.createHadoopCatalog( + "texera_iceberg", + StorageConfig.fileStorageDirectoryPath + ) + case "rest" => + IcebergUtil.createRestCatalog( + "texera_iceberg", + warehouse + ) + case "postgres" => + IcebergUtil.createPostgresCatalog( + "texera_iceberg", + StorageConfig.fileStorageDirectoryPath + ) + case unsupported => + throw new IllegalArgumentException(s"Unsupported catalog type: $unsupported") } - } /** - * Replaces the existing Iceberg catalog instance. - * - This method is useful for testing or dynamically updating the catalog. + * Replaces the cached catalog for a warehouse, primarily for testing or reconfiguration. * - * @param catalog the new Iceberg catalog instance to replace the current one. + * @param catalog the catalog to cache. + * @param warehouse the warehouse to cache it under; defaults to the configured warehouse. */ - def replaceInstance(catalog: Catalog): Unit = { - instance = Some(catalog) - } + def replaceInstance(catalog: Catalog, warehouse: String = defaultWarehouse): Unit = + synchronized { + catalogs(warehouse) = catalog + } } diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala index 291c31896b0..f9bf58528e5 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala @@ -40,6 +40,20 @@ object VFSResourceType extends Enumeration { object VFSURIFactory { val VFS_FILE_URI_SCHEME = "vfs" + // Warehouse is carried as a leading `/wh/` path segment so a storage URI + // fully identifies which warehouse its table lives in. Absent for non-BYO storage. + private def warehousePathSegment(warehouse: Option[String]): String = + warehouse.map(name => s"/wh/$name").getOrElse("") + + /** + * Extracts the warehouse name encoded in a VFS URI, if present. + */ + def warehouseFromURI(uri: URI): Option[String] = { + val segments = uri.getPath.stripPrefix("/").split("/").toList + val idx = segments.indexOf("wh") + if (idx >= 0 && idx + 1 < segments.length) Some(segments(idx + 1)) else None + } + /** * Parses a VFS URI and extracts its components * @@ -90,10 +104,11 @@ object VFSURIFactory { def createPortBaseURI( workflowId: WorkflowIdentity, executionId: ExecutionIdentity, - globalPortId: GlobalPortIdentity + globalPortId: GlobalPortIdentity, + warehouse: Option[String] = None ): URI = new URI( - s"$VFS_FILE_URI_SCHEME:///wid/${workflowId.id}/eid/${executionId.id}" + + s"$VFS_FILE_URI_SCHEME://${warehousePathSegment(warehouse)}/wid/${workflowId.id}/eid/${executionId.id}" + s"/globalportid/${globalPortId.serializeAsString}" ) @@ -109,12 +124,14 @@ object VFSURIFactory { */ def createRuntimeStatisticsURI( workflowId: WorkflowIdentity, - executionId: ExecutionIdentity + executionId: ExecutionIdentity, + warehouse: Option[String] = None ): URI = { createNonResultVFSURI( VFSResourceType.RUNTIME_STATISTICS, workflowId, - executionId + executionId, + warehouse = warehouse ) } @@ -124,13 +141,15 @@ object VFSURIFactory { def createConsoleMessagesURI( workflowId: WorkflowIdentity, executionId: ExecutionIdentity, - operatorId: OperatorIdentity + operatorId: OperatorIdentity, + warehouse: Option[String] = None ): URI = { createNonResultVFSURI( VFSResourceType.CONSOLE_MESSAGES, workflowId, executionId, - Some(operatorId) + Some(operatorId), + warehouse ) } @@ -150,7 +169,8 @@ object VFSURIFactory { resourceType: VFSResourceType.Value, workflowId: WorkflowIdentity, executionId: ExecutionIdentity, - operatorId: Option[OperatorIdentity] = None + operatorId: Option[OperatorIdentity] = None, + warehouse: Option[String] = None ): URI = { if (resourceType == VFSResourceType.RESULT) { @@ -171,10 +191,11 @@ object VFSURIFactory { ) } + val whSegment = warehousePathSegment(warehouse) val baseUri = operatorId match { case Some(opId) => - s"$VFS_FILE_URI_SCHEME:///wid/${workflowId.id}/eid/${executionId.id}/opid/${opId.id}" - case None => s"$VFS_FILE_URI_SCHEME:///wid/${workflowId.id}/eid/${executionId.id}" + s"$VFS_FILE_URI_SCHEME://$whSegment/wid/${workflowId.id}/eid/${executionId.id}/opid/${opId.id}" + case None => s"$VFS_FILE_URI_SCHEME://$whSegment/wid/${workflowId.id}/eid/${executionId.id}" } new URI(s"$baseUri/${resourceType.toString.toLowerCase}") diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala index 182da2baac2..74a58224d7f 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.core.storage.result.iceberg +import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.core.storage.IcebergCatalogInstance import org.apache.texera.amber.core.storage.model.{BufferedItemWriter, VirtualDocument} import org.apache.texera.amber.core.storage.util.StorageUtil.{withLock, withReadLock, withWriteLock} @@ -57,6 +58,7 @@ object Constants { * @param tableSchema schema of the table. * @param serde function to serialize T into an Iceberg Record. * @param deserde function to deserialize an Iceberg Record into T. + * @param warehouse the warehouse whose catalog backs this table; defaults to the configured warehouse. * @tparam T type of the data items stored in the Iceberg table. */ private[storage] class IcebergDocument[T >: Null <: AnyRef]( @@ -64,13 +66,14 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef]( val tableName: String, val tableSchema: org.apache.iceberg.Schema, val serde: (org.apache.iceberg.Schema, T) => Record, - val deserde: (org.apache.iceberg.Schema, Record) => T + val deserde: (org.apache.iceberg.Schema, Record) => T, + val warehouse: String = StorageConfig.icebergRESTCatalogWarehouseName ) extends VirtualDocument[T] with OnIceberg { private val lock = new ReentrantReadWriteLock() - @transient lazy val catalog: Catalog = IcebergCatalogInstance.getInstance() + @transient lazy val catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse) /** * Returns the URI of the table location. diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala index df6944ed888..5d12daf18a6 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala @@ -35,5 +35,6 @@ class WorkflowContext( var workflowId: WorkflowIdentity = DEFAULT_WORKFLOW_ID, var executionId: ExecutionIdentity = DEFAULT_EXECUTION_ID, var workflowSettings: WorkflowSettings = DEFAULT_WORKFLOW_SETTINGS, - var cuid: Option[Int] = None + var cuid: Option[Int] = None, + var warehouse: Option[String] = None ) From acad885064403e46fe1499897ec5e4273b52a33c Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:01:27 -0700 Subject: [PATCH 2/9] test(storage): cover per-warehouse catalog cache and warehouse-scoped URIs Add unit tests for the new warehouse-aware storage behavior: VFSURIFactory warehouseFromURI extraction and the /wh/ URI segment (present with a warehouse, absent without), that a warehouse-scoped URI still round-trips through decodeURI, and that DocumentFactory creates/finds a document for a warehouse-scoped URI (the /wh/ prefix is stripped from the storage key). Python VFSURIFactory.warehouse_from_uri gets matching coverage. Also update the existing state-materialization test to save/restore the IcebergCatalogInstance per-warehouse cache instead of the removed _instance singleton field. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../test_state_materialization_e2e.py | 9 +++-- .../core/storage/test_vfs_uri_factory.py | 23 ++++++++++++ .../core/storage/DocumentFactorySpec.scala | 21 +++++++++++ .../core/storage/VFSURIFactorySpec.scala | 37 +++++++++++++++++++ 4 files changed, 86 insertions(+), 4 deletions(-) diff --git a/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py b/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py index 5db9bdf9036..84c626f58bb 100644 --- a/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py +++ b/amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py @@ -89,8 +89,8 @@ def sqlite_iceberg_catalog(): catalog-agnostic, so the sqlite backend exercises the same code path. Module-scoped so all tests in this file share one warehouse, and so - namespace creation only happens once. We save/restore the original - `IcebergCatalogInstance` singleton so other test modules that expect + namespace creation only happens once. We save/restore the + `IcebergCatalogInstance` cache so other test modules that expect a real postgres-backed catalog (e.g. test_iceberg_document.py) are not affected by our replacement. """ @@ -117,7 +117,7 @@ def sqlite_iceberg_catalog(): s3_large_binaries_base_uri="s3://texera-large-binaries/objects/0/", ) - original_instance = IcebergCatalogInstance._instance + original_catalogs = dict(IcebergCatalogInstance._catalogs) db_path = f"{_WAREHOUSE_DIR}/catalog.sqlite" catalog = SqlCatalog( "texera_iceberg_e2e", @@ -134,7 +134,8 @@ def sqlite_iceberg_catalog(): try: yield catalog finally: - IcebergCatalogInstance.replace_instance(original_instance) + IcebergCatalogInstance._catalogs.clear() + IcebergCatalogInstance._catalogs.update(original_catalogs) def _fresh_base_uri() -> str: diff --git a/amber/src/test/python/core/storage/test_vfs_uri_factory.py b/amber/src/test/python/core/storage/test_vfs_uri_factory.py index 90ea44a52d5..8eda55b1dd8 100644 --- a/amber/src/test/python/core/storage/test_vfs_uri_factory.py +++ b/amber/src/test/python/core/storage/test_vfs_uri_factory.py @@ -138,3 +138,26 @@ def test_rejects_key_present_but_value_missing(self): def test_rejects_unknown_resource_type(self): with pytest.raises(ValueError, match="Unknown resource type: bogus"): VFSURIFactory.decode_uri("vfs:///wid/1/eid/1/bogus") + + +class TestWarehouseFromUri: + def test_extracts_warehouse_from_wh_segment(self): + assert ( + VFSURIFactory.warehouse_from_uri("vfs:///wh/user-2-foo/wid/7/eid/3/result") + == "user-2-foo" + ) + + def test_returns_none_when_no_wh_segment(self): + # Non-BYO URIs have no /wh/ segment, so the warehouse is absent. + assert VFSURIFactory.warehouse_from_uri("vfs:///wid/7/eid/3/result") is None + + def test_decode_round_trips_a_warehouse_scoped_uri(self): + # The leading /wh/ segment must not break decode_uri, which finds + # wid/eid by key rather than by position. + wid, eid, port, resource_type = VFSURIFactory.decode_uri( + "vfs:///wh/user-2-foo/wid/11/eid/22/result" + ) + assert wid.id == 11 + assert eid.id == 22 + assert port is None + assert resource_type == VFSResourceType.RESULT diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala index ad8b5580c53..d62965c567b 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala @@ -235,6 +235,27 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA DocumentFactory.documentExists(stateUri) shouldBe true } + it should "create + find a document for a warehouse-scoped vfs URI (the /wh/ segment is stripped from the storage key)" in { + val base = VFSURIFactory.createPortBaseURI( + WorkflowIdentity(0), + ExecutionIdentity(0), + GlobalPortIdentity( + PhysicalOpIdentity( + logicalOpId = OperatorIdentity(s"op-${UUID.randomUUID().toString.replace("-", "")}"), + layerName = "main" + ), + PortIdentity() + ), + warehouse = Some("wh-test") + ) + val vfsUri = VFSURIFactory.resultURI(base) + vfsUri.getPath should startWith("/wh/wh-test/") + + val doc = DocumentFactory.createDocument(vfsUri, vfsSchema) + doc shouldBe an[IcebergDocument[_]] + DocumentFactory.documentExists(vfsUri) shouldBe true + } + "documentExists" should "report false before creation and true after for a vfs URI" in { val vfsUri = freshResultURI() DocumentFactory.documentExists(vfsUri) shouldBe false diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala index f08bd292d25..06323d3e688 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala @@ -120,4 +120,41 @@ class VFSURIFactorySpec extends AnyFlatSpec { VFSURIFactory.decodeURI(new URI("vfs:///eid/2/wid")) } } + + "VFSURIFactory.warehouseFromURI" should "extract the warehouse from a leading /wh/ segment" in { + val uri = + VFSURIFactory.createPortBaseURI( + workflowId, + executionId, + portId, + warehouse = Some("user-2-foo") + ) + assert(uri.getPath.startsWith("/wh/user-2-foo/")) + assert(VFSURIFactory.warehouseFromURI(uri).contains("user-2-foo")) + } + + it should "return None when no /wh/ segment is present (non-BYO URIs are unchanged)" in { + val uri = VFSURIFactory.createPortBaseURI(workflowId, executionId, portId) + assert(!uri.getPath.contains("/wh/")) + assert(VFSURIFactory.warehouseFromURI(uri).isEmpty) + } + + "A warehouse-scoped URI" should + "still round-trip through decodeURI (wid/eid/port/resource resolved despite the /wh/ prefix)" in { + val base = + VFSURIFactory.createPortBaseURI( + workflowId, + executionId, + portId, + warehouse = Some("user-2-foo") + ) + val resultURI = VFSURIFactory.resultURI(base) + assert(VFSURIFactory.warehouseFromURI(resultURI).contains("user-2-foo")) + + val (wid, eid, globalPortIdOpt, resourceType) = VFSURIFactory.decodeURI(resultURI) + assert(wid == workflowId) + assert(eid == executionId) + assert(globalPortIdOpt.contains(portId)) + assert(resourceType == VFSResourceType.RESULT) + } } From fc31023953eb12b334375c7d44d9fab2377e1da7 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 27 Jul 2026 16:40:01 -0700 Subject: [PATCH 3/9] fix(storage): anchor the warehouse URI segment and share warehouse-agnostic catalogs Three defects in the per-warehouse storage foundation, plus the test gap that let them through. warehouseFromURI scanned the whole path for a `wh` segment while the writer emits it as a leading prefix and DocumentFactory.sanitizeURIPath strips only a leading one. Any later `wh` segment therefore selected a warehouse the URI was never built for, and the storage key disagreed with it. Operator ids reach that path verbatim (they come straight off the workflow JSON), so an id such as `a/wh//b` in a console-messages URI redirected the write to another warehouse -- and the read path applies no ownership check, since the warehouse arrives in the path rather than through the API. Anchor the parse to the first segment, and reject '/' in operator ids and in the GlobalPortIdentity components alongside the existing '_' guard, which was added for this same reason. The catalog cache keyed every entry by warehouse name before dispatching on catalog type, but createCatalog ignores the warehouse for hadoop and postgres. Two warehouse names in a postgres deployment therefore built two equivalent JdbcCatalogs, each with its own connection pool against the same database. Key the warehouse-agnostic types under a constant instead, matching the Python side, which already did this. On the Python side warehouse_from_uri read urlparse().path without unquoting while Scala reads java.net.URI.getPath, which decodes -- so the two languages resolved a percent-encoded name to different warehouses, splitting one execution's data across two of them. create_port_base_uri also had no warehouse parameter, so Python could read the segment but never write it. The warehouse case in DocumentFactorySpec looked up an unregistered warehouse name, which under the default `rest` catalog type missed the cache and tried to build a live REST catalog; it passed only where the environment happened to configure a postgres catalog. Register the name with the local test catalog so the suite exercises the code under test rather than the machine's config. --- .../python/core/storage/vfs_uri_factory.py | 29 +++++++++----- .../core/storage/test_vfs_uri_factory.py | 39 +++++++++++++++++++ .../core/storage/IcebergCatalogInstance.scala | 21 +++++++++- .../amber/core/storage/VFSURIFactory.scala | 27 ++++++++++--- .../util/serde/GlobalPortIdentitySerde.scala | 11 ++++++ .../core/storage/DocumentFactorySpec.scala | 6 ++- .../storage/LocalHadoopIcebergCatalog.scala | 26 +++++++++---- .../core/storage/VFSURIFactorySpec.scala | 25 ++++++++++++ 8 files changed, 160 insertions(+), 24 deletions(-) diff --git a/amber/src/main/python/core/storage/vfs_uri_factory.py b/amber/src/main/python/core/storage/vfs_uri_factory.py index 28733dfbbda..40e61155c8f 100644 --- a/amber/src/main/python/core/storage/vfs_uri_factory.py +++ b/amber/src/main/python/core/storage/vfs_uri_factory.py @@ -17,7 +17,7 @@ from enum import Enum from typing import Optional -from urllib.parse import urlparse +from urllib.parse import unquote, urlparse from core.util.virtual_identity import ( serialize_global_port_identity, @@ -94,21 +94,32 @@ def warehouse_from_uri(uri: str) -> Optional[str]: Extracts the warehouse name from the optional leading "/wh/" segment of a VFS URI, or None if absent. Mirrors VFSURIFactory.warehouseFromURI (Scala) so a URI fully identifies which warehouse its tables live in. + + Anchored to the leading segment, matching Scala and the leading-only strip + in document_factory.sanitize_uri_path: a later segment that happens to be + "wh" must not select a warehouse. The path is unquoted first because + java.net.URI.getPath decodes on the Scala side while urlparse does not -- + without it the two languages disagree on a percent-encoded name. """ - segments = urlparse(uri).path.lstrip("/").split("/") - try: - idx = segments.index("wh") - except ValueError: - return None - return segments[idx + 1] if idx + 1 < len(segments) else None + segments = unquote(urlparse(uri).path).lstrip("/").split("/") + if len(segments) >= 2 and segments[0] == "wh" and segments[1]: + return segments[1] + return None @staticmethod - def create_port_base_uri(workflow_id, execution_id, global_port_id) -> str: + def create_port_base_uri( + workflow_id, execution_id, global_port_id, warehouse: Optional[str] = None + ) -> str: """Base URI for a port. Result and state URIs derive from it via `result_uri` / `state_uri`. + + `warehouse` is written as the leading "/wh/" segment, mirroring the + Scala side; when None the URI is byte-for-byte what it was before warehouses + existed. """ + wh_segment = f"/wh/{warehouse}" if warehouse else "" return ( - f"{VFSURIFactory.VFS_FILE_URI_SCHEME}:///wid/{workflow_id.id}" + f"{VFSURIFactory.VFS_FILE_URI_SCHEME}://{wh_segment}/wid/{workflow_id.id}" f"/eid/{execution_id.id}/globalportid/" f"{serialize_global_port_identity(global_port_id)}" ) diff --git a/amber/src/test/python/core/storage/test_vfs_uri_factory.py b/amber/src/test/python/core/storage/test_vfs_uri_factory.py index 8eda55b1dd8..106433c22d3 100644 --- a/amber/src/test/python/core/storage/test_vfs_uri_factory.py +++ b/amber/src/test/python/core/storage/test_vfs_uri_factory.py @@ -151,6 +151,45 @@ def test_returns_none_when_no_wh_segment(self): # Non-BYO URIs have no /wh/ segment, so the warehouse is absent. assert VFSURIFactory.warehouse_from_uri("vfs:///wid/7/eid/3/result") is None + def test_only_a_leading_wh_segment_counts(self): + # A `wh` deeper in the path -- e.g. inside an operator id -- must not select + # a warehouse; it would disagree with sanitize_uri_path, which strips only a + # leading one, and would route the write to another user's warehouse. + assert ( + VFSURIFactory.warehouse_from_uri( + "vfs:///wid/1/eid/2/opid/a/wh/victim/b/consolemessages" + ) + is None + ) + assert ( + VFSURIFactory.warehouse_from_uri("vfs:///wid/1/eid/2/opid/wh/consolemessages") + is None + ) + + def test_percent_encoded_name_decodes_like_java(self): + # java.net.URI.getPath decodes on the Scala side; urlparse does not. Without + # unquoting here the two languages would resolve the same URI to different + # warehouses, splitting one execution's data across two of them. + assert ( + VFSURIFactory.warehouse_from_uri("vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result") + == "user-2-foo" + ) + + def test_round_trips_a_warehouse_written_by_the_python_builder(self): + # Python can now write the segment, not just read it. + uri = VFSURIFactory.create_port_base_uri( + WorkflowIdentity(id=7), ExecutionIdentity(id=3), _gpi(), "user-2-foo" + ) + assert uri.startswith("vfs:///wh/user-2-foo/wid/7/eid/3/") + assert VFSURIFactory.warehouse_from_uri(uri) == "user-2-foo" + + def test_builder_without_warehouse_is_unchanged(self): + uri = VFSURIFactory.create_port_base_uri( + WorkflowIdentity(id=7), ExecutionIdentity(id=3), _gpi() + ) + assert "/wh/" not in uri + assert VFSURIFactory.warehouse_from_uri(uri) is None + def test_decode_round_trips_a_warehouse_scoped_uri(self): # The leading /wh/ segment must not break decode_uri, which finds # wid/eid by key rather than by position. diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala index 30b2ef749c0..9f791b9c01f 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala @@ -42,8 +42,25 @@ object IcebergCatalogInstance { private val catalogs = mutable.Map.empty[String, Catalog] + // Cache key for the warehouse-agnostic catalog types. Not a legal warehouse name, + // so it cannot collide with a REST warehouse. + private val SharedCatalogKey = "" + private def defaultWarehouse: String = StorageConfig.icebergRESTCatalogWarehouseName + /** + * The cache key for a warehouse. Only the REST catalog is scoped to a warehouse; + * hadoop and postgres ignore it, so they must share one entry. Keying them by + * warehouse name would build a second, fully equivalent catalog per distinct name + * -- for postgres a second JdbcCatalog with its own connection pool, all pointing + * at the same database. Mirrors the Python side, which keys those under a constant. + */ + private def cacheKey(warehouse: String): String = + StorageConfig.icebergCatalogType match { + case "rest" => warehouse + case _ => SharedCatalogKey + } + /** * Retrieves the catalog for the given warehouse, creating and caching it on first access. * @@ -52,7 +69,7 @@ object IcebergCatalogInstance { */ def getInstance(warehouse: String = defaultWarehouse): Catalog = synchronized { - catalogs.getOrElseUpdate(warehouse, createCatalog(warehouse)) + catalogs.getOrElseUpdate(cacheKey(warehouse), createCatalog(warehouse)) } private def createCatalog(warehouse: String): Catalog = @@ -84,6 +101,6 @@ object IcebergCatalogInstance { */ def replaceInstance(catalog: Catalog, warehouse: String = defaultWarehouse): Unit = synchronized { - catalogs(warehouse) = catalog + catalogs(cacheKey(warehouse)) = catalog } } diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala index f9bf58528e5..1916636ed6d 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala @@ -47,12 +47,18 @@ object VFSURIFactory { /** * Extracts the warehouse name encoded in a VFS URI, if present. + * + * Anchored to the leading segment on purpose. The warehouse is written as a + * leading `/wh/` prefix and `DocumentFactory` strips only a leading one, + * so scanning the whole path would disagree with the stripper: a later segment + * that happens to be `wh` -- e.g. inside a user-chosen operator id in a + * console-messages URI -- would select a warehouse the URI was never built for. */ - def warehouseFromURI(uri: URI): Option[String] = { - val segments = uri.getPath.stripPrefix("/").split("/").toList - val idx = segments.indexOf("wh") - if (idx >= 0 && idx + 1 < segments.length) Some(segments(idx + 1)) else None - } + def warehouseFromURI(uri: URI): Option[String] = + uri.getPath.stripPrefix("/").split("/").toList match { + case "wh" :: name :: _ if name.nonEmpty => Some(name) + case _ => None + } /** * Parses a VFS URI and extracts its components @@ -191,6 +197,17 @@ object VFSURIFactory { ) } + // The operator id is user-supplied (it comes straight off the workflow JSON) and + // is interpolated into the URI path below. A '/' in it would add path segments, + // letting it forge structure the URI never meant to have -- e.g. a `wh/` + // pair that would then be read back as a warehouse. + operatorId.foreach { opId => + require( + !opId.id.contains('/'), + s"operatorId must not contain '/' (VFS URI parsing relies on this): ${opId.id}" + ) + } + val whSegment = warehousePathSegment(warehouse) val baseUri = operatorId match { case Some(opId) => diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/serde/GlobalPortIdentitySerde.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/serde/GlobalPortIdentitySerde.scala index c8fd8e1a363..65a836bf7a3 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/serde/GlobalPortIdentitySerde.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/serde/GlobalPortIdentitySerde.scala @@ -53,6 +53,17 @@ object GlobalPortIdentitySerde { !layerName.contains('_'), s"layerName must not contain '_' (VFS URI parsing relies on this): $layerName" ) + // A '/' would add path segments to the VFS URI these ids are interpolated + // into, letting a user-chosen id forge structure the URI never meant to have + // (e.g. a `wh/` pair). Rejected for the same reason as '_'. + require( + !logicalOpId.contains('/'), + s"logicalOpId must not contain '/' (VFS URI parsing relies on this): $logicalOpId" + ) + require( + !layerName.contains('/'), + s"layerName must not contain '/' (VFS URI parsing relies on this): $layerName" + ) require(portId >= 0, s"portId must be non-negative: $portId") s"(logicalOpId=$logicalOpId,layerName=$layerName,portId=$portId,isInternal=$isInternal,isInput=$isInput)" } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala index d62965c567b..ecb5f5c54f5 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala @@ -61,7 +61,11 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA override def beforeAll(): Unit = { super.beforeAll() - LocalHadoopIcebergCatalog.ensure() + // "wh-test" is registered explicitly so the warehouse-scoped case resolves to + // this local catalog instead of reaching for a live REST catalog under the + // default (`rest`) config -- otherwise the case would pass or fail according to + // the machine's catalog type rather than the code under test. + LocalHadoopIcebergCatalog.ensure("wh-test") } // --------------------------------------------------------------------------- diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala index 3b92b763a30..74dcb6abd8e 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.core.storage +import org.apache.iceberg.catalog.Catalog import org.apache.texera.amber.util.IcebergUtil import java.nio.file.{Files, Path} @@ -42,11 +43,21 @@ import java.util.Comparator */ object LocalHadoopIcebergCatalog { - private var initialized = false + private var catalog: Option[Catalog] = None - def ensure(): Unit = + /** + * Installs the shared local catalog, additionally registering it under each of + * `warehouses`. + * + * A suite that exercises a warehouse-scoped URI must name that warehouse here: + * under the configured `rest` catalog type the cache is keyed by warehouse name, + * so an unregistered name would miss and try to build a *live* REST catalog, + * making the suite pass or fail according to the machine's catalog config rather + * than the code under test. + */ + def ensure(warehouses: String*): Unit = synchronized { - if (!initialized) { + val installed = catalog.getOrElse { val warehouse = Files.createTempDirectory("wfcore-iceberg-shared") // Best-effort recursive cleanup of the temp warehouse on JVM exit so test runs // don't leave wfcore-iceberg-shared* directories behind on dev machines / CI. @@ -57,10 +68,11 @@ object LocalHadoopIcebergCatalog { .forEach((p: Path) => Files.deleteIfExists(p)) catch { case _: Throwable => () } } - IcebergCatalogInstance.replaceInstance( - IcebergUtil.createHadoopCatalog("wfcore-test", warehouse) - ) - initialized = true + val created = IcebergUtil.createHadoopCatalog("wfcore-test", warehouse) + catalog = Some(created) + IcebergCatalogInstance.replaceInstance(created) + created } + warehouses.foreach(name => IcebergCatalogInstance.replaceInstance(installed, name)) } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala index 06323d3e688..ee03f8e0775 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala @@ -139,6 +139,31 @@ class VFSURIFactorySpec extends AnyFlatSpec { assert(VFSURIFactory.warehouseFromURI(uri).isEmpty) } + it should "only honour a LEADING wh segment, never one deeper in the path" in { + // A `wh` appearing later -- e.g. inside an operator id -- must not select a + // warehouse: it would disagree with DocumentFactory, which strips only a + // leading `wh//`, and would route the write to another user's warehouse. + assert( + VFSURIFactory + .warehouseFromURI(new URI("vfs:///wid/1/eid/2/opid/a/wh/victim/b/consolemessages")) + .isEmpty + ) + // An operator literally named `wh` is likewise not a warehouse. + assert( + VFSURIFactory.warehouseFromURI(new URI("vfs:///wid/1/eid/2/opid/wh/consolemessages")).isEmpty + ) + } + + "VFSURIFactory" should "reject an operatorId containing '/' rather than let it forge URI segments" in { + assertThrows[IllegalArgumentException] { + VFSURIFactory.createConsoleMessagesURI( + workflowId, + executionId, + OperatorIdentity("a/wh/victim/b") + ) + } + } + "A warehouse-scoped URI" should "still round-trip through decodeURI (wid/eid/port/resource resolved despite the /wh/ prefix)" in { val base = From 3d8ea0b5f2e74e180b780715454c5221c2462df3 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:00:50 -0700 Subject: [PATCH 4/9] style(storage): apply ruff formatting to the warehouse URI tests --- .../src/test/python/core/storage/test_vfs_uri_factory.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/amber/src/test/python/core/storage/test_vfs_uri_factory.py b/amber/src/test/python/core/storage/test_vfs_uri_factory.py index 106433c22d3..9b7fb37f603 100644 --- a/amber/src/test/python/core/storage/test_vfs_uri_factory.py +++ b/amber/src/test/python/core/storage/test_vfs_uri_factory.py @@ -162,7 +162,9 @@ def test_only_a_leading_wh_segment_counts(self): is None ) assert ( - VFSURIFactory.warehouse_from_uri("vfs:///wid/1/eid/2/opid/wh/consolemessages") + VFSURIFactory.warehouse_from_uri( + "vfs:///wid/1/eid/2/opid/wh/consolemessages" + ) is None ) @@ -171,7 +173,9 @@ def test_percent_encoded_name_decodes_like_java(self): # unquoting here the two languages would resolve the same URI to different # warehouses, splitting one execution's data across two of them. assert ( - VFSURIFactory.warehouse_from_uri("vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result") + VFSURIFactory.warehouse_from_uri( + "vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result" + ) == "user-2-foo" ) From edd11d2cd1e6a142bc422e21de604f1366fb8094 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:26:34 -0700 Subject: [PATCH 5/9] fix(storage): split the raw URI path so an encoded slash cannot forge segments The `/` guards added for operator ids only reject a literal slash, and the warehouse name was interpolated into the URI path with no validation at all. Since `java.net.URI.getPath` percent-decodes, a `%2F` slipped past both: a warehouse named `a%2Fwid%2F999` produced a path that reads back as `/wh/a/wid/999/wid/1/...`, so the warehouse resolved to `a` and `decodeURI`'s key search found the injected `wid` rather than the real one -- one execution's data landing under another's storage key. Python was not affected there, because its parser splits the undecoded path, so the two languages also disagreed. Fix it structurally rather than by blacklisting more characters: split the RAW path in both languages, so a percent-encoded slash stays inside the segment that contains it and cannot become a separator, and require the warehouse segment to be a legal name, so a URI that `warehousePathSegment` could never have written resolves to no warehouse instead of to a wrong one. Validate the name on the way in as well -- registration applies a stricter rule, but the URI layer should not depend on that. Reported by Copilot on #6944. --- .../python/core/storage/vfs_uri_factory.py | 39 ++++++++++++++++--- .../core/storage/test_vfs_uri_factory.py | 27 ++++++++++--- .../amber/core/storage/VFSURIFactory.scala | 32 +++++++++++++-- .../core/storage/VFSURIFactorySpec.scala | 23 +++++++++++ 4 files changed, 106 insertions(+), 15 deletions(-) diff --git a/amber/src/main/python/core/storage/vfs_uri_factory.py b/amber/src/main/python/core/storage/vfs_uri_factory.py index 40e61155c8f..d7a4d9db1c2 100644 --- a/amber/src/main/python/core/storage/vfs_uri_factory.py +++ b/amber/src/main/python/core/storage/vfs_uri_factory.py @@ -17,7 +17,8 @@ from enum import Enum from typing import Optional -from urllib.parse import unquote, urlparse +import re +from urllib.parse import urlparse from core.util.virtual_identity import ( serialize_global_port_identity, @@ -37,6 +38,10 @@ class VFSResourceType(str, Enum): STATE = "state" +# See VFSURIFactory._is_valid_warehouse_name. +_WAREHOUSE_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]*") + + class VFSURIFactory: VFS_FILE_URI_SCHEME = "vfs" @@ -88,6 +93,14 @@ def extract_value(key: str) -> str: resource_type, ) + @staticmethod + def _is_valid_warehouse_name(name: str) -> bool: + """A warehouse name becomes a URI path segment, so it may not carry + characters that have meaning there: no "/" to add segments, and no "%" to + smuggle one in percent-encoded form. Mirrors VFSURIFactory (Scala). + """ + return _WAREHOUSE_NAME_RE.fullmatch(name) is not None + @staticmethod def warehouse_from_uri(uri: str) -> Optional[str]: """ @@ -97,12 +110,19 @@ def warehouse_from_uri(uri: str) -> Optional[str]: Anchored to the leading segment, matching Scala and the leading-only strip in document_factory.sanitize_uri_path: a later segment that happens to be - "wh" must not select a warehouse. The path is unquoted first because - java.net.URI.getPath decodes on the Scala side while urlparse does not -- - without it the two languages disagree on a percent-encoded name. + "wh" must not select a warehouse. The RAW path is split -- never decoded + first -- so a percent-encoded slash stays inside its own segment instead of + becoming a separator, and the name must be a legal warehouse name, so + anything that could not have been written by create_port_base_uri resolves + to no warehouse rather than to a wrong one. Scala splits its raw path the + same way, so both languages read a URI identically. """ - segments = unquote(urlparse(uri).path).lstrip("/").split("/") - if len(segments) >= 2 and segments[0] == "wh" and segments[1]: + segments = urlparse(uri).path.lstrip("/").split("/") + if ( + len(segments) >= 2 + and segments[0] == "wh" + and VFSURIFactory._is_valid_warehouse_name(segments[1]) + ): return segments[1] return None @@ -117,6 +137,13 @@ def create_port_base_uri( Scala side; when None the URI is byte-for-byte what it was before warehouses existed. """ + if warehouse is not None and not VFSURIFactory._is_valid_warehouse_name( + warehouse + ): + raise ValueError( + f"warehouse name must match {_WAREHOUSE_NAME_RE.pattern} " + f"(it becomes a URI path segment): {warehouse}" + ) wh_segment = f"/wh/{warehouse}" if warehouse else "" return ( f"{VFSURIFactory.VFS_FILE_URI_SCHEME}://{wh_segment}/wid/{workflow_id.id}" diff --git a/amber/src/test/python/core/storage/test_vfs_uri_factory.py b/amber/src/test/python/core/storage/test_vfs_uri_factory.py index 9b7fb37f603..3a0ad3ee801 100644 --- a/amber/src/test/python/core/storage/test_vfs_uri_factory.py +++ b/amber/src/test/python/core/storage/test_vfs_uri_factory.py @@ -168,17 +168,34 @@ def test_only_a_leading_wh_segment_counts(self): is None ) - def test_percent_encoded_name_decodes_like_java(self): - # java.net.URI.getPath decodes on the Scala side; urlparse does not. Without - # unquoting here the two languages would resolve the same URI to different - # warehouses, splitting one execution's data across two of them. + def test_percent_encoded_name_is_rejected_not_decoded(self): + # The raw path is split and the name must be a legal warehouse name, so a + # percent-encoded name resolves to no warehouse rather than to a decoded + # one. Decoding first would let "%2F" become a separator and pick the wrong + # warehouse; it would also diverge from Scala, which splits its raw path. assert ( VFSURIFactory.warehouse_from_uri( "vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result" ) - == "user-2-foo" + is None + ) + + def test_encoded_slash_in_name_cannot_forge_segments(self): + # "%2F" must stay inside its own segment: decoded-then-split, this URI would + # read as /wh/a/wid/999/... and hand back "a" while also shifting wid. + assert ( + VFSURIFactory.warehouse_from_uri( + "vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result" + ) + is None ) + def test_builder_rejects_an_unsafe_warehouse_name(self): + with pytest.raises(ValueError, match="warehouse name must match"): + VFSURIFactory.create_port_base_uri( + WorkflowIdentity(id=7), ExecutionIdentity(id=3), _gpi(), "a/b" + ) + def test_round_trips_a_warehouse_written_by_the_python_builder(self): # Python can now write the segment, not just read it. uri = VFSURIFactory.create_port_base_uri( diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala index 1916636ed6d..993885b17b6 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala @@ -40,10 +40,28 @@ object VFSResourceType extends Enumeration { object VFSURIFactory { val VFS_FILE_URI_SCHEME = "vfs" + // A warehouse name becomes a URI path segment, so it is restricted to characters + // that carry no meaning there: no '/' to add segments, and no '%' to smuggle one + // in percent-encoded form. Registration applies a stricter rule still; this is the + // invariant the URI layer itself depends on. + private val warehouseNamePattern = "[A-Za-z0-9][A-Za-z0-9_-]*".r + + private def isValidWarehouseName(name: String): Boolean = + warehouseNamePattern.pattern.matcher(name).matches() + // Warehouse is carried as a leading `/wh/` path segment so a storage URI // fully identifies which warehouse its table lives in. Absent for non-BYO storage. private def warehousePathSegment(warehouse: Option[String]): String = - warehouse.map(name => s"/wh/$name").getOrElse("") + warehouse + .map { name => + require( + isValidWarehouseName(name), + s"warehouse name must match ${warehouseNamePattern.regex} " + + s"(it becomes a URI path segment): $name" + ) + s"/wh/$name" + } + .getOrElse("") /** * Extracts the warehouse name encoded in a VFS URI, if present. @@ -53,11 +71,17 @@ object VFSURIFactory { * so scanning the whole path would disagree with the stripper: a later segment * that happens to be `wh` -- e.g. inside a user-chosen operator id in a * console-messages URI -- would select a warehouse the URI was never built for. + * + * Splits the RAW path rather than the decoded one, so a percent-encoded slash + * stays inside the segment that contains it instead of becoming a separator; + * decoding first would let `%2F` forge path structure. The name is then required + * to be a legal warehouse name, so anything that could not have been written by + * `warehousePathSegment` resolves to no warehouse rather than to a wrong one. */ def warehouseFromURI(uri: URI): Option[String] = - uri.getPath.stripPrefix("/").split("/").toList match { - case "wh" :: name :: _ if name.nonEmpty => Some(name) - case _ => None + Option(uri.getRawPath).toList.flatMap(_.stripPrefix("/").split("/")) match { + case "wh" :: name :: _ if isValidWarehouseName(name) => Some(name) + case _ => None } /** diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala index ee03f8e0775..132c92a5f00 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala @@ -154,6 +154,19 @@ class VFSURIFactorySpec extends AnyFlatSpec { ) } + it should "not let a percent-encoded slash in the name forge extra segments" in { + // Decoded before splitting, this path would read as /wh/a/wid/999/... -- handing + // back "a" as the warehouse and shifting which `wid` the parser sees. Splitting + // the raw path keeps `%2F` inside its own segment, and the name is then rejected + // as illegal, so the URI resolves to no warehouse rather than to a wrong one. + assert( + VFSURIFactory.warehouseFromURI(new URI("vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result")).isEmpty + ) + assert( + VFSURIFactory.warehouseFromURI(new URI("vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result")).isEmpty + ) + } + "VFSURIFactory" should "reject an operatorId containing '/' rather than let it forge URI segments" in { assertThrows[IllegalArgumentException] { VFSURIFactory.createConsoleMessagesURI( @@ -164,6 +177,16 @@ class VFSURIFactorySpec extends AnyFlatSpec { } } + it should "reject a warehouse name that is not safe as a URI path segment" in { + Seq("a/b", "a%2Fb", "", "-lead", "sp ace").foreach { bad => + withClue(s"warehouse name '$bad' should be rejected: ") { + assertThrows[IllegalArgumentException] { + VFSURIFactory.createPortBaseURI(workflowId, executionId, portId, Some(bad)) + } + } + } + } + "A warehouse-scoped URI" should "still round-trip through decodeURI (wid/eid/port/resource resolved despite the /wh/ prefix)" in { val base = From d264e35f4ccf634ef9f7b5e836165cb181c568e7 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Tue, 28 Jul 2026 13:45:58 -0700 Subject: [PATCH 6/9] fix(storage): reject '/' in port-identity fields on the Python side too The Scala serde was given a '/' guard alongside its existing '_' guard, because both characters would otherwise let a user-chosen operator id forge structure in the VFS URI the id is interpolated into. Python's serialize_global_port_identity mirrors that serializer but only ever checked '_', so the contract held in one language and not the other. No production Python code builds these URIs today, so this closes the gap rather than fixing a reachable bug -- but the guard is the contract both sides parse against, and it should not depend on which language wrote the string. --- .../main/python/core/util/virtual_identity.py | 25 ++++++++++--------- .../python/core/util/test_virtual_identity.py | 11 ++++++++ 2 files changed, 24 insertions(+), 12 deletions(-) diff --git a/amber/src/main/python/core/util/virtual_identity.py b/amber/src/main/python/core/util/virtual_identity.py index 6893e7e8f05..0a9de0ccc48 100644 --- a/amber/src/main/python/core/util/virtual_identity.py +++ b/amber/src/main/python/core/util/virtual_identity.py @@ -64,24 +64,25 @@ def serialize_global_port_identity(obj: GlobalPortIdentity) -> str: ``(logicalOpId=,layerName=, portId=,isInternal=,isInput=)`` - Raises ValueError if `logicalOpId` or `layerName` contains an underscore - (VFS URI parsing relies on the absence of '_'), or if `portId` is negative. + Raises ValueError if `logicalOpId` or `layerName` contains an underscore or a + slash (VFS URI parsing relies on the absence of both), or if `portId` is + negative. Mirrors GlobalPortIdentitySerde (Scala). """ logical_op_id = obj.op_id.logical_op_id.id layer_name = obj.op_id.layer_name port_id = obj.port_id.id is_internal = obj.port_id.internal is_input_port = obj.input - if "_" in logical_op_id: - raise ValueError( - f"logicalOpId must not contain '_' " - f"(VFS URI parsing relies on this): {logical_op_id}" - ) - if "_" in layer_name: - raise ValueError( - f"layerName must not contain '_' " - f"(VFS URI parsing relies on this): {layer_name}" - ) + for field, value in (("logicalOpId", logical_op_id), ("layerName", layer_name)): + # '_' would collide with the separator the storage key is built from, and + # '/' would add path segments to the VFS URI this string is interpolated + # into -- letting an id forge structure the URI never meant to have. + for forbidden in ("_", "/"): + if forbidden in value: + raise ValueError( + f"{field} must not contain '{forbidden}' " + f"(VFS URI parsing relies on this): {value}" + ) if port_id < 0: raise ValueError(f"portId must be non-negative: {port_id}") return ( diff --git a/amber/src/test/python/core/util/test_virtual_identity.py b/amber/src/test/python/core/util/test_virtual_identity.py index 75c8e30a74b..ad60391c287 100644 --- a/amber/src/test/python/core/util/test_virtual_identity.py +++ b/amber/src/test/python/core/util/test_virtual_identity.py @@ -159,6 +159,17 @@ def test_rejects_underscore_in_layer_name(self): with pytest.raises(ValueError, match="layerName must not contain"): serialize_global_port_identity(_gpi(layer="main_source_0_op")) + def test_rejects_slash_in_logical_op_id(self): + # A '/' would add path segments to the VFS URI this string is interpolated + # into, letting an id forge structure the URI never meant to have. Matches + # the guard in GlobalPortIdentitySerde (Scala). + with pytest.raises(ValueError, match="logicalOpId must not contain"): + serialize_global_port_identity(_gpi(op_id="a/wh/victim/b")) + + def test_rejects_slash_in_layer_name(self): + with pytest.raises(ValueError, match="layerName must not contain"): + serialize_global_port_identity(_gpi(layer="main/wh/victim")) + def test_rejects_negative_port_id(self): # Port ids are array indices and must be non-negative. with pytest.raises(ValueError, match="portId must be non-negative"): From 75660d6a4e16f22dbf6c84230f9bc701d6b5acfd Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:19:59 -0700 Subject: [PATCH 7/9] fix(storage): locate URI keys by raw segment in decodeURI as well warehouseFromURI was changed to split the raw path so a percent-encoded slash cannot forge segments, but decodeURI -- which finds `wid`/`eid` by searching the same segments -- was left splitting the decoded path. A URI whose warehouse name contained `%2F` therefore read as `/wh/a/wid/999/wid/1/...` there, and the key search returned the injected `wid` rather than the real one, putting one execution's data under another's storage key. Python's decode_uri already split the raw path, so the two languages also disagreed about which execution a URI identified. Split the raw path here too. Legitimate URIs are unaffected: the characters the serialized port identity contributes -- '(', ')', ',', '=' -- are all legal in a path segment and are never percent-encoded, so raw and decoded are identical for anything the factory writes. --- .../texera/amber/core/storage/VFSURIFactory.scala | 6 +++++- .../amber/core/storage/VFSURIFactorySpec.scala | 12 ++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala index 993885b17b6..160d6f79677 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala @@ -101,7 +101,11 @@ object VFSURIFactory { throw new IllegalArgumentException(s"Invalid URI scheme: ${uri.getScheme}") } - val segments = uri.getPath.stripPrefix("/").split("/").toList + // Raw path, for the same reason as warehouseFromURI: keys are located by + // searching the segments, so a percent-encoded slash inside a segment must not + // split it and shift which `wid`/`eid` the search finds. Python's decode_uri + // splits the raw path too, so both languages read a URI identically. + val segments = uri.getRawPath.stripPrefix("/").split("/").toList def extractValue(key: String): String = { val index = segments.indexOf(key) diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala index 132c92a5f00..850294d1555 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala @@ -167,6 +167,18 @@ class VFSURIFactorySpec extends AnyFlatSpec { ) } + it should "locate wid/eid by raw segment, so an encoded slash cannot shift them" in { + // Decoded before splitting, this path reads as /wh/a/wid/999/wid/1/... and the + // key search finds the injected `wid` first -- resolving to execution 999 and + // landing this execution's data under another's storage key. Python's + // decode_uri splits the raw path, so decoding here would also make the two + // languages disagree about which execution a URI belongs to. + val (wid, eid, _, _) = + VFSURIFactory.decodeURI(new URI("vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result")) + assert(wid == WorkflowIdentity(1)) + assert(eid == ExecutionIdentity(2)) + } + "VFSURIFactory" should "reject an operatorId containing '/' rather than let it forge URI segments" in { assertThrows[IllegalArgumentException] { VFSURIFactory.createConsoleMessagesURI( From 006d47aa6d93bcfb2ebda1e91c96cdb775abc0d0 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:52:15 -0700 Subject: [PATCH 8/9] test(storage): use the named VFSUriComponents fields after the main merge --- .../amber/core/storage/VFSURIFactorySpec.scala | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala index 324e44ae17b..bbb96a90bf0 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala @@ -174,10 +174,10 @@ class VFSURIFactorySpec extends AnyFlatSpec { // landing this execution's data under another's storage key. Python's // decode_uri splits the raw path, so decoding here would also make the two // languages disagree about which execution a URI belongs to. - val (wid, eid, _, _) = + val components = VFSURIFactory.decodeURI(new URI("vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result")) - assert(wid == WorkflowIdentity(1)) - assert(eid == ExecutionIdentity(2)) + assert(components.workflowId == WorkflowIdentity(1)) + assert(components.executionId == ExecutionIdentity(2)) } "VFSURIFactory" should "reject an operatorId containing '/' rather than let it forge URI segments" in { @@ -212,10 +212,10 @@ class VFSURIFactorySpec extends AnyFlatSpec { val resultURI = VFSURIFactory.resultURI(base) assert(VFSURIFactory.warehouseFromURI(resultURI).contains("user-2-foo")) - val (wid, eid, globalPortIdOpt, resourceType) = VFSURIFactory.decodeURI(resultURI) - assert(wid == workflowId) - assert(eid == executionId) - assert(globalPortIdOpt.contains(portId)) - assert(resourceType == VFSResourceType.RESULT) + val components = VFSURIFactory.decodeURI(resultURI) + assert(components.workflowId == workflowId) + assert(components.executionId == executionId) + assert(components.globalPortId.contains(portId)) + assert(components.resourceType == VFSResourceType.RESULT) } } From 99bac9c1dfdf1a389f135058c8cb1fc8a63c047f Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:04:59 -0700 Subject: [PATCH 9/9] refactor(storage): report the warehouse as part of VFSUriComponents The URI had two parsers: decodeURI, which located wid/eid/port/resource by searching the segments, and a separate warehouseFromURI. Keeping them apart has already cost two bugs -- they disagreed first about whether a `wh` segment had to be leading, then about whether to split the raw or the decoded path -- and the reason to keep them apart is gone now that decodeURI returns a named structure (#7020): adding a field costs no call-site churn. Fold the warehouse into VFSUriComponents as a trailing field, so one parse yields every component and there is a single place where the URI's shape is interpreted. Scala and Python both derive it from the segments the decoder already split. This also lets DocumentFactory resolve the warehouse inside the VFS branch rather than before the scheme dispatch, where it also ran -- harmlessly, but pointlessly -- for dataset URIs that have no warehouse. --- .../python/core/storage/document_factory.py | 18 +++--- .../python/core/storage/vfs_uri_factory.py | 32 +++++----- .../core/storage/test_vfs_uri_factory.py | 59 ++++++++++--------- .../amber/core/storage/DocumentFactory.scala | 27 ++++----- .../amber/core/storage/VFSURIFactory.scala | 31 ++++++---- .../core/storage/VFSURIFactorySpec.scala | 35 +++++++---- 6 files changed, 115 insertions(+), 87 deletions(-) diff --git a/amber/src/main/python/core/storage/document_factory.py b/amber/src/main/python/core/storage/document_factory.py index 55a0849daff..0db1b413922 100644 --- a/amber/src/main/python/core/storage/document_factory.py +++ b/amber/src/main/python/core/storage/document_factory.py @@ -79,10 +79,10 @@ def sanitize_uri_path(uri): def create_document(uri: str, schema: Schema) -> VirtualDocument: parsed_uri = urlparse(uri) if parsed_uri.scheme == VFSURIFactory.VFS_FILE_URI_SCHEME: - resource_type = VFSURIFactory.decode_uri(uri).resource_type - namespace = DocumentFactory._resolve_namespace(resource_type) + components = VFSURIFactory.decode_uri(uri) + namespace = DocumentFactory._resolve_namespace(components.resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) - warehouse = VFSURIFactory.warehouse_from_uri(uri) + warehouse = components.warehouse # Convert Amber Schema to Iceberg Schema with LARGE_BINARY # field name encoding iceberg_schema = amber_schema_to_iceberg_schema(schema) @@ -124,10 +124,10 @@ def document_exists(uri: str) -> bool: """ parsed_uri = urlparse(uri) if parsed_uri.scheme == VFSURIFactory.VFS_FILE_URI_SCHEME: - resource_type = VFSURIFactory.decode_uri(uri).resource_type - namespace = DocumentFactory._resolve_namespace(resource_type) + components = VFSURIFactory.decode_uri(uri) + namespace = DocumentFactory._resolve_namespace(components.resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) - warehouse = VFSURIFactory.warehouse_from_uri(uri) + warehouse = components.warehouse return IcebergCatalogInstance.get_instance(warehouse).table_exists( f"{namespace}.{storage_key}" ) @@ -140,10 +140,10 @@ def document_exists(uri: str) -> bool: def open_document(uri: str) -> typing.Tuple[VirtualDocument, Optional[Schema]]: parsed_uri = urlparse(uri) if parsed_uri.scheme == VFSURIFactory.VFS_FILE_URI_SCHEME: - resource_type = VFSURIFactory.decode_uri(uri).resource_type - namespace = DocumentFactory._resolve_namespace(resource_type) + components = VFSURIFactory.decode_uri(uri) + namespace = DocumentFactory._resolve_namespace(components.resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) - warehouse = VFSURIFactory.warehouse_from_uri(uri) + warehouse = components.warehouse table = load_table_metadata( IcebergCatalogInstance.get_instance(warehouse), diff --git a/amber/src/main/python/core/storage/vfs_uri_factory.py b/amber/src/main/python/core/storage/vfs_uri_factory.py index 3ae2f42db3b..37a28cbbf0d 100644 --- a/amber/src/main/python/core/storage/vfs_uri_factory.py +++ b/amber/src/main/python/core/storage/vfs_uri_factory.py @@ -51,6 +51,10 @@ class VFSUriComponents(NamedTuple): execution_id: ExecutionIdentity global_port_id: Optional[GlobalPortIdentity] resource_type: VFSResourceType + # The warehouse whose catalog holds this URI's table, from the optional leading + # "/wh/" segment; None for non-BYO storage, which uses the configured + # default. Last so positional unpacking of the earlier fields still works. + warehouse: Optional[str] = None class VFSURIFactory: @@ -95,6 +99,7 @@ def extract_value(key: str) -> str: execution_id, global_port_id, resource_type, + VFSURIFactory._warehouse_from(segments), ) @staticmethod @@ -106,22 +111,21 @@ def _is_valid_warehouse_name(name: str) -> bool: return _WAREHOUSE_NAME_RE.fullmatch(name) is not None @staticmethod - def warehouse_from_uri(uri: str) -> Optional[str]: + def _warehouse_from(segments: list) -> Optional[str]: """ - Extracts the warehouse name from the optional leading "/wh/" segment - of a VFS URI, or None if absent. Mirrors VFSURIFactory.warehouseFromURI - (Scala) so a URI fully identifies which warehouse its tables live in. - - Anchored to the leading segment, matching Scala and the leading-only strip - in document_factory.sanitize_uri_path: a later segment that happens to be - "wh" must not select a warehouse. The RAW path is split -- never decoded - first -- so a percent-encoded slash stays inside its own segment instead of - becoming a separator, and the name must be a legal warehouse name, so - anything that could not have been written by create_port_base_uri resolves - to no warehouse rather than to a wrong one. Scala splits its raw path the - same way, so both languages read a URI identically. + The warehouse encoded in a VFS URI, if present. Reported as part of + VFSUriComponents by decode_uri, which is the only way in. Mirrors + VFSURIFactory.warehouseFrom (Scala). + + Anchored to the leading segment: a later segment that happens to be "wh" + -- e.g. inside a user-chosen operator id -- must not select a warehouse; it + would disagree with document_factory.sanitize_uri_path, which strips only a + leading one, and would route the write to another user's warehouse. The + segments come from the RAW path (see decode_uri), so a percent-encoded slash + stays inside its own segment, and the name must be a legal warehouse name, + so anything create_port_base_uri could not have written resolves to no + warehouse rather than to a wrong one. """ - segments = urlparse(uri).path.lstrip("/").split("/") if ( len(segments) >= 2 and segments[0] == "wh" diff --git a/amber/src/test/python/core/storage/test_vfs_uri_factory.py b/amber/src/test/python/core/storage/test_vfs_uri_factory.py index e490b379ef1..24ad583ae6e 100644 --- a/amber/src/test/python/core/storage/test_vfs_uri_factory.py +++ b/amber/src/test/python/core/storage/test_vfs_uri_factory.py @@ -135,31 +135,27 @@ def test_rejects_unknown_resource_type(self): VFSURIFactory.decode_uri("vfs:///wid/1/eid/1/bogus") -class TestWarehouseFromUri: - def test_extracts_warehouse_from_wh_segment(self): - assert ( - VFSURIFactory.warehouse_from_uri("vfs:///wh/user-2-foo/wid/7/eid/3/result") - == "user-2-foo" - ) +class TestWarehouseInDecodedUri: + def test_reports_warehouse_from_wh_segment(self): + components = VFSURIFactory.decode_uri("vfs:///wh/user-2-foo/wid/7/eid/3/result") + assert components.warehouse == "user-2-foo" - def test_returns_none_when_no_wh_segment(self): + def test_warehouse_is_none_when_no_wh_segment(self): # Non-BYO URIs have no /wh/ segment, so the warehouse is absent. - assert VFSURIFactory.warehouse_from_uri("vfs:///wid/7/eid/3/result") is None + assert VFSURIFactory.decode_uri("vfs:///wid/7/eid/3/result").warehouse is None def test_only_a_leading_wh_segment_counts(self): # A `wh` deeper in the path -- e.g. inside an operator id -- must not select # a warehouse; it would disagree with sanitize_uri_path, which strips only a # leading one, and would route the write to another user's warehouse. assert ( - VFSURIFactory.warehouse_from_uri( - "vfs:///wid/1/eid/2/opid/a/wh/victim/b/consolemessages" - ) + VFSURIFactory.decode_uri( + "vfs:///wid/1/eid/2/opid/a/wh/victim/b/result" + ).warehouse is None ) assert ( - VFSURIFactory.warehouse_from_uri( - "vfs:///wid/1/eid/2/opid/wh/consolemessages" - ) + VFSURIFactory.decode_uri("vfs:///wid/1/eid/2/opid/wh/result").warehouse is None ) @@ -169,21 +165,22 @@ def test_percent_encoded_name_is_rejected_not_decoded(self): # one. Decoding first would let "%2F" become a separator and pick the wrong # warehouse; it would also diverge from Scala, which splits its raw path. assert ( - VFSURIFactory.warehouse_from_uri( + VFSURIFactory.decode_uri( "vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result" - ) + ).warehouse is None ) def test_encoded_slash_in_name_cannot_forge_segments(self): # "%2F" must stay inside its own segment: decoded-then-split, this URI would - # read as /wh/a/wid/999/... and hand back "a" while also shifting wid. - assert ( - VFSURIFactory.warehouse_from_uri( - "vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result" - ) - is None + # read as /wh/a/wid/999/... handing back "a" as the warehouse, and the key + # search would find the injected `wid` instead of the real one. + components = VFSURIFactory.decode_uri( + "vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result" ) + assert components.warehouse is None + assert components.workflow_id.id == 1 + assert components.execution_id.id == 2 def test_builder_rejects_an_unsafe_warehouse_name(self): with pytest.raises(ValueError, match="warehouse name must match"): @@ -197,22 +194,26 @@ def test_round_trips_a_warehouse_written_by_the_python_builder(self): WorkflowIdentity(id=7), ExecutionIdentity(id=3), _gpi(), "user-2-foo" ) assert uri.startswith("vfs:///wh/user-2-foo/wid/7/eid/3/") - assert VFSURIFactory.warehouse_from_uri(uri) == "user-2-foo" + assert ( + VFSURIFactory.decode_uri(VFSURIFactory.result_uri(uri)).warehouse + == "user-2-foo" + ) def test_builder_without_warehouse_is_unchanged(self): uri = VFSURIFactory.create_port_base_uri( WorkflowIdentity(id=7), ExecutionIdentity(id=3), _gpi() ) assert "/wh/" not in uri - assert VFSURIFactory.warehouse_from_uri(uri) is None + assert VFSURIFactory.decode_uri(VFSURIFactory.result_uri(uri)).warehouse is None def test_decode_round_trips_a_warehouse_scoped_uri(self): # The leading /wh/ segment must not break decode_uri, which finds # wid/eid by key rather than by position. - wid, eid, port, resource_type = VFSURIFactory.decode_uri( + components = VFSURIFactory.decode_uri( "vfs:///wh/user-2-foo/wid/11/eid/22/result" ) - assert wid.id == 11 - assert eid.id == 22 - assert port is None - assert resource_type == VFSResourceType.RESULT + assert components.workflow_id.id == 11 + assert components.execution_id.id == 22 + assert components.global_port_id is None + assert components.resource_type == VFSResourceType.RESULT + assert components.warehouse == "user-2-foo" diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala index cbcb4304ec3..f9051e4324c 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala @@ -22,11 +22,7 @@ package org.apache.texera.amber.core.storage import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.core.storage.FileResolver.DATASET_FILE_URI_SCHEME import org.apache.texera.amber.core.storage.VFSResourceType._ -import org.apache.texera.amber.core.storage.VFSURIFactory.{ - VFS_FILE_URI_SCHEME, - decodeURI, - warehouseFromURI -} +import org.apache.texera.amber.core.storage.VFSURIFactory.{VFS_FILE_URI_SCHEME, decodeURI} import org.apache.texera.amber.core.storage.model._ import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument import org.apache.texera.amber.core.tuple.{Schema, Tuple} @@ -44,10 +40,10 @@ object DocumentFactory { private def sanitizeURIPath(uri: URI): String = uri.getPath.stripPrefix("/").replaceFirst("^wh/[^/]+/", "").replace("/", "_") - // The warehouse a URI's table lives in, encoded as a leading `/wh/` path + // The warehouse a URI's table lives in, carried as a leading `/wh/` path // segment; absent for non-BYO storage, which falls back to the configured warehouse. - private def warehouseOf(uri: URI): String = - warehouseFromURI(uri).getOrElse(StorageConfig.icebergRESTCatalogWarehouseName) + private def warehouseOf(components: VFSUriComponents): String = + components.warehouse.getOrElse(StorageConfig.icebergRESTCatalogWarehouseName) private def resolveNamespace(resourceType: VFSResourceType.Value): String = resourceType match { @@ -83,10 +79,11 @@ object DocumentFactory { * @return the created document */ def createDocument(uri: URI, schema: Schema): VirtualDocument[_] = { - val warehouse = warehouseOf(uri) uri.getScheme match { case VFS_FILE_URI_SCHEME => - val resourceType = decodeURI(uri).resourceType + val components = decodeURI(uri) + val warehouse = warehouseOf(components) + val resourceType = components.resourceType val storageKey = sanitizeURIPath(uri) val namespace = resolveNamespace(resourceType) @@ -128,10 +125,11 @@ object DocumentFactory { * iceberg namespace mapping. */ def documentExists(uri: URI): Boolean = { - val warehouse = warehouseOf(uri) uri.getScheme match { case VFS_FILE_URI_SCHEME => - val resourceType = decodeURI(uri).resourceType + val components = decodeURI(uri) + val warehouse = warehouseOf(components) + val resourceType = components.resourceType val storageKey = sanitizeURIPath(uri) val namespace = resolveNamespace(resourceType) IcebergCatalogInstance @@ -174,11 +172,12 @@ object DocumentFactory { * @return the VirtualDocument, which is the handler of the data; the Schema, which is the schema of the data stored in the document */ def openDocument(uri: URI): (VirtualDocument[_], Option[Schema]) = { - val warehouse = warehouseOf(uri) uri.getScheme match { case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None) case VFS_FILE_URI_SCHEME => - val resourceType = decodeURI(uri).resourceType + val components = decodeURI(uri) + val warehouse = warehouseOf(components) + val resourceType = components.resourceType val storageKey = sanitizeURIPath(uri) val namespace = resolveNamespace(resourceType) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala index 49fdafcb4b0..79e36e77f4d 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala @@ -45,7 +45,11 @@ case class VFSUriComponents( workflowId: WorkflowIdentity, executionId: ExecutionIdentity, globalPortId: Option[GlobalPortIdentity], - resourceType: VFSResourceType.Value + resourceType: VFSResourceType.Value, + // The warehouse whose catalog holds this URI's table, from the optional leading + // `/wh/` segment; None for non-BYO storage, which uses the configured + // default. Last so positional unpacking of the earlier fields still works. + warehouse: Option[String] = None ) object VFSURIFactory { @@ -75,7 +79,8 @@ object VFSURIFactory { .getOrElse("") /** - * Extracts the warehouse name encoded in a VFS URI, if present. + * The warehouse encoded in a VFS URI, if present. Reported as part of + * [[VFSUriComponents]] by [[decodeURI]], which is the only way in. * * Anchored to the leading segment on purpose. The warehouse is written as a * leading `/wh/` prefix and `DocumentFactory` strips only a leading one, @@ -83,14 +88,14 @@ object VFSURIFactory { * that happens to be `wh` -- e.g. inside a user-chosen operator id in a * console-messages URI -- would select a warehouse the URI was never built for. * - * Splits the RAW path rather than the decoded one, so a percent-encoded slash - * stays inside the segment that contains it instead of becoming a separator; - * decoding first would let `%2F` forge path structure. The name is then required - * to be a legal warehouse name, so anything that could not have been written by - * `warehousePathSegment` resolves to no warehouse rather than to a wrong one. + * The segments come from the RAW path (see `decodeURI`), so a percent-encoded + * slash stays inside the segment that contains it instead of becoming a + * separator. The name is then required to be a legal warehouse name, so anything + * that could not have been written by `warehousePathSegment` resolves to no + * warehouse rather than to a wrong one. */ - def warehouseFromURI(uri: URI): Option[String] = - Option(uri.getRawPath).toList.flatMap(_.stripPrefix("/").split("/")) match { + private def warehouseFrom(segments: List[String]): Option[String] = + segments match { case "wh" :: name :: _ if isValidWarehouseName(name) => Some(name) case _ => None } @@ -134,7 +139,13 @@ object VFSURIFactory { .find(_.toString.toLowerCase == resourceTypeStr) .getOrElse(throw new IllegalArgumentException(s"Unknown resource type: $resourceTypeStr")) - VFSUriComponents(workflowId, executionId, globalPortIdOption, resourceType) + VFSUriComponents( + workflowId, + executionId, + globalPortIdOption, + resourceType, + warehouseFrom(segments) + ) } /** diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala index bbb96a90bf0..f163a0658ea 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala @@ -60,7 +60,7 @@ class VFSURIFactorySpec extends AnyFlatSpec { assert(resultURI.getPath.endsWith("/result")) assert(stateURI.getPath.endsWith("/state")) - val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType) = + val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType, _) = VFSURIFactory.decodeURI(resultURI) assert(wid == workflowId) assert(eid == executionId) @@ -75,7 +75,7 @@ class VFSURIFactorySpec extends AnyFlatSpec { assert(path.endsWith("/runtimestatistics")) assert(!path.contains("/opid/")) - val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType) = VFSURIFactory.decodeURI(uri) + val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType, _) = VFSURIFactory.decodeURI(uri) assert(wid == workflowId) assert(eid == executionId) assert(globalPortIdOpt.isEmpty) @@ -90,7 +90,7 @@ class VFSURIFactorySpec extends AnyFlatSpec { // The current `decodeURI` does not extract the operator id (it has no // "opid" branch), so we only round-trip wid/eid/resourceType here. - val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType) = VFSURIFactory.decodeURI(uri) + val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType, _) = VFSURIFactory.decodeURI(uri) assert(wid == workflowId) assert(eid == executionId) assert(globalPortIdOpt.isEmpty) @@ -122,7 +122,7 @@ class VFSURIFactorySpec extends AnyFlatSpec { } } - "VFSURIFactory.warehouseFromURI" should "extract the warehouse from a leading /wh/ segment" in { + "decodeURI" should "report the warehouse from a leading /wh/ segment" in { val uri = VFSURIFactory.createPortBaseURI( workflowId, @@ -131,13 +131,16 @@ class VFSURIFactorySpec extends AnyFlatSpec { warehouse = Some("user-2-foo") ) assert(uri.getPath.startsWith("/wh/user-2-foo/")) - assert(VFSURIFactory.warehouseFromURI(uri).contains("user-2-foo")) + // A base URI has no resource segment, so decode the derived result URI. + assert( + VFSURIFactory.decodeURI(VFSURIFactory.resultURI(uri)).warehouse.contains("user-2-foo") + ) } it should "return None when no /wh/ segment is present (non-BYO URIs are unchanged)" in { val uri = VFSURIFactory.createPortBaseURI(workflowId, executionId, portId) assert(!uri.getPath.contains("/wh/")) - assert(VFSURIFactory.warehouseFromURI(uri).isEmpty) + assert(VFSURIFactory.decodeURI(VFSURIFactory.resultURI(uri)).warehouse.isEmpty) } it should "only honour a LEADING wh segment, never one deeper in the path" in { @@ -146,12 +149,16 @@ class VFSURIFactorySpec extends AnyFlatSpec { // leading `wh//`, and would route the write to another user's warehouse. assert( VFSURIFactory - .warehouseFromURI(new URI("vfs:///wid/1/eid/2/opid/a/wh/victim/b/consolemessages")) + .decodeURI(new URI("vfs:///wid/1/eid/2/opid/a/wh/victim/b/consolemessages")) + .warehouse .isEmpty ) // An operator literally named `wh` is likewise not a warehouse. assert( - VFSURIFactory.warehouseFromURI(new URI("vfs:///wid/1/eid/2/opid/wh/consolemessages")).isEmpty + VFSURIFactory + .decodeURI(new URI("vfs:///wid/1/eid/2/opid/wh/consolemessages")) + .warehouse + .isEmpty ) } @@ -161,10 +168,16 @@ class VFSURIFactorySpec extends AnyFlatSpec { // the raw path keeps `%2F` inside its own segment, and the name is then rejected // as illegal, so the URI resolves to no warehouse rather than to a wrong one. assert( - VFSURIFactory.warehouseFromURI(new URI("vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result")).isEmpty + VFSURIFactory + .decodeURI(new URI("vfs:///wh/a%2Fwid%2F999/wid/1/eid/2/result")) + .warehouse + .isEmpty ) assert( - VFSURIFactory.warehouseFromURI(new URI("vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result")).isEmpty + VFSURIFactory + .decodeURI(new URI("vfs:///wh/user-2%2Dfoo/wid/7/eid/3/result")) + .warehouse + .isEmpty ) } @@ -210,7 +223,7 @@ class VFSURIFactorySpec extends AnyFlatSpec { warehouse = Some("user-2-foo") ) val resultURI = VFSURIFactory.resultURI(base) - assert(VFSURIFactory.warehouseFromURI(resultURI).contains("user-2-foo")) + assert(VFSURIFactory.decodeURI(resultURI).warehouse.contains("user-2-foo")) val components = VFSURIFactory.decodeURI(resultURI) assert(components.workflowId == workflowId)