You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/29 05:54:03 UTC
(texera) branch gh-readonly-queue/main/pr-7020-d7cb2183869800461d07a93539289fa684bee94d created (now a61702fd10)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7020-d7cb2183869800461d07a93539289fa684bee94d
in repository https://gitbox.apache.org/repos/asf/texera.git
at a61702fd10 refactor(storage): return named VFSUriComponents from decodeURI (#7020)
This branch includes the following new commits:
new a61702fd10 refactor(storage): return named VFSUriComponents from decodeURI (#7020)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: refactor(storage): return named VFSUriComponents from decodeURI (#7020)
Posted by gi...@apache.org.
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 <me...@uci.edu>
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)