This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7020-d7cb2183869800461d07a93539289fa684bee94d in repository https://gitbox.apache.org/repos/asf/texera.git
commit a61702fd10ccde130f0a24a1298adff24c4a38a5 Author: Meng Wang <[email protected]> AuthorDate: Tue Jul 28 22:53:43 2026 -0700 refactor(storage): return named VFSUriComponents from decodeURI (#7020) ### What changes were proposed in this PR? `VFSURIFactory.decodeURI` returned a positional 4-tuple `(WorkflowIdentity, ExecutionIdentity, Option[GlobalPortIdentity], VFSResourceType)` in both Scala and Python, so its nine production call sites destructured positionally — mostly binding a single field through patterns like `val (_, _, _, resourceType) = decodeURI(uri)` — where nothing catches a swapped `_` position, and adding a field means touching every site. The Scala doc already advertised `@return A VFSUriComponents object`, but no such type existed. - **Scala**: introduce `case class VFSUriComponents(workflowId, executionId, globalPortId, resourceType)` in `VFSURIFactory.scala`; `decodeURI` now returns it. All six production call sites (`DocumentFactory` ×3, `ExecutionResultService` ×2, `WorkflowExecutionsResource` ×1) read named fields; the compiler forced every site to be updated, so none can be missed. - **Python**: introduce a matching `VFSUriComponents(NamedTuple)` in `vfs_uri_factory.py`; `decode_uri` returns it. A `NamedTuple` keeps positional unpacking working alongside named access, so this is non-breaking for any external unpacking. The three `document_factory.py` sites read `.resource_type`. - **Tests**: the Scala and Python suites' destructurings updated to named access; `test_document_factory`'s mocked `decode_uri` now returns a real `VFSUriComponents` (production accesses the result by name, so a bare tuple would no longer satisfy it). Pure refactor, no behavior change, symmetric across both languages. ### Any related issues, documentation, discussions? Closes #6979. ### How was this PR tested? - No new tests — the existing round-trip suites pin the decoded values in both languages, and the refactor is validated by them: - Scala: `VFSURIFactorySpec` 8/8, `WorkflowExecutionsResourceSpec` 22/22, `scalafmtCheck` (WorkflowCore + WorkflowExecutionService, main + Test) all pass locally. - Python: the two storage suites pass 21/21 locally; `ruff check` and `ruff format --check` clean. - The Scala side is additionally compile-verified: a tuple destructure of a case class does not compile, so every call site was necessarily updated. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-8) --- .../main/python/core/storage/document_factory.py | 6 ++-- .../main/python/core/storage/vfs_uri_factory.py | 24 ++++++++------- .../user/workflow/WorkflowExecutionsResource.scala | 6 ++-- .../web/service/ExecutionResultService.scala | 4 +-- .../python/core/storage/test_document_factory.py | 4 +-- .../python/core/storage/test_vfs_uri_factory.py | 35 ++++++++++------------ .../workflow/WorkflowExecutionsResourceSpec.scala | 9 ++++-- .../amber/core/storage/DocumentFactory.scala | 6 ++-- .../texera/amber/core/storage/VFSURIFactory.scala | 20 ++++++++----- .../amber/core/storage/VFSURIFactorySpec.scala | 9 +++--- 10 files changed, 66 insertions(+), 57 deletions(-) diff --git a/amber/src/main/python/core/storage/document_factory.py b/amber/src/main/python/core/storage/document_factory.py index 5078ca7dd5..25e1d08211 100644 --- a/amber/src/main/python/core/storage/document_factory.py +++ b/amber/src/main/python/core/storage/document_factory.py @@ -73,7 +73,7 @@ class DocumentFactory: 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 = VFSURIFactory.decode_uri(uri).resource_type namespace = DocumentFactory._resolve_namespace(resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) # Convert Amber Schema to Iceberg Schema with LARGE_BINARY @@ -116,7 +116,7 @@ class DocumentFactory: """ parsed_uri = urlparse(uri) if parsed_uri.scheme == VFSURIFactory.VFS_FILE_URI_SCHEME: - _, _, _, resource_type = VFSURIFactory.decode_uri(uri) + resource_type = VFSURIFactory.decode_uri(uri).resource_type namespace = DocumentFactory._resolve_namespace(resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) return IcebergCatalogInstance.get_instance().table_exists( @@ -131,7 +131,7 @@ class DocumentFactory: 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 = VFSURIFactory.decode_uri(uri).resource_type namespace = DocumentFactory._resolve_namespace(resource_type) storage_key = DocumentFactory.sanitize_uri_path(parsed_uri) 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 883450abf2..d0eb3eab7a 100644 --- a/amber/src/main/python/core/storage/vfs_uri_factory.py +++ b/amber/src/main/python/core/storage/vfs_uri_factory.py @@ -16,7 +16,7 @@ # under the License. from enum import Enum -from typing import Optional +from typing import NamedTuple, Optional from urllib.parse import urlparse from core.util.virtual_identity import ( @@ -37,18 +37,22 @@ class VFSResourceType(str, Enum): STATE = "state" +class VFSUriComponents(NamedTuple): + """The named components encoded in a VFS URI, as returned by + `VFSURIFactory.decode_uri`. A NamedTuple, so positional unpacking keeps + working alongside named access.""" + + workflow_id: WorkflowIdentity + execution_id: ExecutionIdentity + global_port_id: Optional[GlobalPortIdentity] + resource_type: VFSResourceType + + class VFSURIFactory: VFS_FILE_URI_SCHEME = "vfs" @staticmethod - def decode_uri( - uri: str, - ) -> ( - WorkflowIdentity, - ExecutionIdentity, - Optional[GlobalPortIdentity], - VFSResourceType, - ): + def decode_uri(uri: str) -> "VFSUriComponents": """ Parses a VFS URI and extracts its components. """ @@ -81,7 +85,7 @@ class VFSURIFactory: except ValueError: raise ValueError(f"Unknown resource type: {resource_type_str}") - return ( + return VFSUriComponents( workflow_id, execution_id, global_port_id, diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala index 5eab3dab39..784d678f37 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala @@ -471,12 +471,12 @@ object WorkflowExecutionsResource { portId: PortIdentity ): Option[URI] = { def isMatchingExternalPortURI(uri: URI): Boolean = { - val (_, _, globalPortIdOption, resourceType) = VFSURIFactory.decodeURI(uri) - globalPortIdOption.exists { globalPortId => + val components = VFSURIFactory.decodeURI(uri) + components.globalPortId.exists { globalPortId => !globalPortId.portId.internal && globalPortId.opId.logicalOpId == opId && globalPortId.portId == portId && - resourceType == VFSResourceType.RESULT + components.resourceType == VFSResourceType.RESULT } } diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala index f52a17f8ad..c2ffadb9f1 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala @@ -438,7 +438,7 @@ class ExecutionResultService( ) if (storageUri.nonEmpty) { - val (_, _, globalPortIdOption, _) = VFSURIFactory.decodeURI(storageUri.get) + val globalPortIdOption = VFSURIFactory.decodeURI(storageUri.get).globalPortId val opStorage = DocumentFactory.openDocument(storageUri.get)._1 allTableStats(opId.id) = opStorage.getTableStatistics @@ -524,7 +524,7 @@ class ExecutionResultService( .map(uri => { val count = DocumentFactory.openDocument(uri)._1.getCount.toInt - val (_, _, globalPortIdOption, _) = VFSURIFactory.decodeURI(uri) + val globalPortIdOption = VFSURIFactory.decodeURI(uri).globalPortId // Retrieve the mode of the specified output port val mode = physicalPlan diff --git a/amber/src/test/python/core/storage/test_document_factory.py b/amber/src/test/python/core/storage/test_document_factory.py index 359b0c0aed..654025006a 100644 --- a/amber/src/test/python/core/storage/test_document_factory.py +++ b/amber/src/test/python/core/storage/test_document_factory.py @@ -22,7 +22,7 @@ import pytest from core.models import Schema from core.storage.document_factory import DocumentFactory from core.storage.storage_config import StorageConfig -from core.storage.vfs_uri_factory import VFSResourceType +from core.storage.vfs_uri_factory import VFSResourceType, VFSUriComponents # Avoid initializing the real config (only initializable once per process). @@ -39,7 +39,7 @@ def schema(): def _decode_returning(resource_type): """Helper: build a VFSURIFactory.decode_uri side_effect.""" - return lambda _uri: (None, None, None, resource_type) + return lambda _uri: VFSUriComponents(None, None, None, resource_type) @patch("core.storage.document_factory.IcebergDocument") 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 90ea44a52d..5422f2ad68 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 @@ -80,13 +80,12 @@ class TestDecodeUriRoundTrip: base = VFSURIFactory.create_port_base_uri(wid, eid, gpi) uri = VFSURIFactory.result_uri(base) - decoded_wid, decoded_eid, decoded_gpi, resource_type = VFSURIFactory.decode_uri( - uri - ) + components = VFSURIFactory.decode_uri(uri) - assert decoded_wid.id == 42 - assert decoded_eid.id == 9 - assert resource_type == VFSResourceType.RESULT + assert components.workflow_id.id == 42 + assert components.execution_id.id == 9 + assert components.resource_type == VFSResourceType.RESULT + decoded_gpi = components.global_port_id assert decoded_gpi is not None assert decoded_gpi.op_id.logical_op_id.id == "opA" assert decoded_gpi.op_id.layer_name == "main" @@ -98,25 +97,21 @@ class TestDecodeUriRoundTrip: wid, eid, gpi = WorkflowIdentity(id=5), ExecutionIdentity(id=6), _gpi() uri = VFSURIFactory.state_uri(VFSURIFactory.create_port_base_uri(wid, eid, gpi)) - decoded_wid, decoded_eid, decoded_gpi, resource_type = VFSURIFactory.decode_uri( - uri - ) + components = VFSURIFactory.decode_uri(uri) - assert decoded_wid.id == 5 - assert decoded_eid.id == 6 - assert resource_type == VFSResourceType.STATE - assert decoded_gpi.op_id.logical_op_id.id == "myOp" + assert components.workflow_id.id == 5 + assert components.execution_id.id == 6 + assert components.resource_type == VFSResourceType.STATE + assert components.global_port_id.op_id.logical_op_id.id == "myOp" def test_decode_returns_none_port_when_globalportid_absent(self): # A URI without a globalportid segment yields a None port identity, # exercising the optional branch in decode_uri. - wid, eid, port, resource_type = VFSURIFactory.decode_uri( - "vfs:///wid/11/eid/22/result" - ) - assert wid.id == 11 - assert eid.id == 22 - assert port is None - assert resource_type == VFSResourceType.RESULT + components = VFSURIFactory.decode_uri("vfs:///wid/11/eid/22/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 class TestDecodeUriErrorPaths: diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala index fdda487953..04104fb065 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala @@ -623,9 +623,12 @@ class WorkflowExecutionsResourceSpec assert(found.contains(targetUri)) // Sanity-check: the decoded URI is RESULT-typed and matches the target ids. - val (_, _, gpOpt, resType) = VFSURIFactory.decodeURI(found.get) - assert(resType == VFSResourceType.RESULT) - assert(gpOpt.exists(gp => gp.opId.logicalOpId == targetOpId && gp.portId == targetPortId)) + val components = VFSURIFactory.decodeURI(found.get) + assert(components.resourceType == VFSResourceType.RESULT) + assert( + components.globalPortId + .exists(gp => gp.opId.logicalOpId == targetOpId && gp.portId == targetPortId) + ) } it should "return None when no URI matches the requested op/port" in { 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 a26340e79c..910c1dce44 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 @@ -76,7 +76,7 @@ object DocumentFactory { def createDocument(uri: URI, schema: Schema): VirtualDocument[_] = { uri.getScheme match { case VFS_FILE_URI_SCHEME => - val (_, _, _, resourceType) = decodeURI(uri) + val resourceType = decodeURI(uri).resourceType val storageKey = sanitizeURIPath(uri) val namespace = resolveNamespace(resourceType) @@ -119,7 +119,7 @@ object DocumentFactory { def documentExists(uri: URI): Boolean = { uri.getScheme match { case VFS_FILE_URI_SCHEME => - val (_, _, _, resourceType) = decodeURI(uri) + val resourceType = decodeURI(uri).resourceType val storageKey = sanitizeURIPath(uri) val namespace = resolveNamespace(resourceType) IcebergCatalogInstance @@ -165,7 +165,7 @@ object DocumentFactory { uri.getScheme match { case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None) case VFS_FILE_URI_SCHEME => - val (_, _, _, resourceType) = decodeURI(uri) + val resourceType = decodeURI(uri).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 291c31896b..5124c745aa 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 @@ -37,6 +37,17 @@ object VFSResourceType extends Enumeration { val STATE: Value = Value("state") } +/** + * The named components encoded in a VFS URI, as returned by + * [[VFSURIFactory.decodeURI]]. + */ +case class VFSUriComponents( + workflowId: WorkflowIdentity, + executionId: ExecutionIdentity, + globalPortId: Option[GlobalPortIdentity], + resourceType: VFSResourceType.Value +) + object VFSURIFactory { val VFS_FILE_URI_SCHEME = "vfs" @@ -47,12 +58,7 @@ object VFSURIFactory { * @return A `VFSUriComponents` object with the extracted data. * @throws java.lang.IllegalArgumentException if the URI is malformed. */ - def decodeURI(uri: URI): ( - WorkflowIdentity, - ExecutionIdentity, - Option[GlobalPortIdentity], - VFSResourceType.Value - ) = { + def decodeURI(uri: URI): VFSUriComponents = { if (uri.getScheme != VFS_FILE_URI_SCHEME) { throw new IllegalArgumentException(s"Invalid URI scheme: ${uri.getScheme}") } @@ -80,7 +86,7 @@ object VFSURIFactory { .find(_.toString.toLowerCase == resourceTypeStr) .getOrElse(throw new IllegalArgumentException(s"Unknown resource type: $resourceTypeStr")) - (workflowId, executionId, globalPortIdOption, resourceType) + VFSUriComponents(workflowId, executionId, globalPortIdOption, resourceType) } /** 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 f08bd292d2..aa3eb0dfdb 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,12 +60,13 @@ class VFSURIFactorySpec extends AnyFlatSpec { assert(resultURI.getPath.endsWith("/result")) assert(stateURI.getPath.endsWith("/state")) - val (wid, eid, globalPortIdOpt, resourceType) = VFSURIFactory.decodeURI(resultURI) + val VFSUriComponents(wid, eid, globalPortIdOpt, resourceType) = + VFSURIFactory.decodeURI(resultURI) assert(wid == workflowId) assert(eid == executionId) assert(globalPortIdOpt.contains(portId)) assert(resourceType == VFSResourceType.RESULT) - assert(VFSURIFactory.decodeURI(stateURI)._4 == VFSResourceType.STATE) + assert(VFSURIFactory.decodeURI(stateURI).resourceType == VFSResourceType.STATE) } "VFSURIFactory.createRuntimeStatisticsURI" should "produce a runtimeStatistics URI without an opid segment" in { @@ -74,7 +75,7 @@ class VFSURIFactorySpec extends AnyFlatSpec { assert(path.endsWith("/runtimestatistics")) assert(!path.contains("/opid/")) - val (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) @@ -89,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 (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)
