test(workflow-core): add unit test coverage for port-identity serde helpers (#4954)

### What changes were proposed in this PR?

Adds `PortIdentitySerdeSpec` covering three Jackson / VFS-URI serde
helpers in `org.apache.texera.amber.util.serde`, and tightens the
production contract on `GlobalPortIdentitySerde` after review.

Production change (added per review):

| Surface | Change |
| --- | --- |
| `GlobalPortIdentitySerde.serializeAsString` | Rejects `portId < 0`,
`logicalOpId` containing `_`, `layerName` containing `_` with
`IllegalArgumentException`. |
| `GlobalPortIdentitySerde.deserializeFromString` | Symmetrically
rejects `portId < 0`. |
| Python `serialize_global_port_identity` /
`deserialize_global_port_identity` | Mirrored: raises `ValueError` on
the same invariants. |
| Iceberg test fixtures (`IcebergTableStatsSpec`, `IcebergDocumentSpec`,
`test_iceberg_document.py`) | Op-id `test_table_<uuid>` →
`test-table-<uuid>` to honor the new contract.
`IcebergDocumentConsoleMessagesSpec` is unchanged —
`createConsoleMessagesURI` does not go through `serializeAsString`. |

Test coverage:

| Helper | Pinned by this spec |
| --- | --- |
| `GlobalPortIdentitySerde.serializeAsString` / `deserializeFromString`
| Default + per-field round-trip; exact format pin (default +
non-default values); special-character pass-through (dashes / dots);
negative `portId` rejected (serialize + deserialize); underscored
`logicalOpId` rejected; underscored `layerName` rejected;
format-character no-underscore invariant; seven negative paths —
completely malformed, missing field, wrong field order, trailing
content, empty body, non-numeric `portId` → `NumberFormatException`,
non-boolean flag → `IllegalArgumentException` |
| `PortIdentityKeySerializer.portIdToString` | Exact `id_internal`
format. |
| `PortIdentityKeySerializer` + `PortIdentityKeyDeserializer` (Jackson
wiring) | The production `JSONUtils.objectMapper`, round-tripping
`Map[PortIdentity, String]` and an empty map. |
| `PortIdentityKeyDeserializer.deserializeKey` | Four negative paths
(non-integer id, non-boolean flag, missing-separator with non-numeric
body, missing-separator with numeric-only body); current lenient
extra-segments behavior pinned in a characterization test, with a
`pendingUntilFixed` for the stricter contract. |

### Any related issues, documentation, discussions?

Closes #4953

### How was this PR tested?

```
sbt "WorkflowCore/Test/testOnly org.apache.texera.amber.util.serde.PortIdentitySerdeSpec"
# 24 tests, all pass (1 unchanged pendingUntilFixed for PortIdentityKeyDeserializer)

sbt "WorkflowCore/Test/testOnly org.apache.texera.amber.core.storage.VFSURIFactorySpec org.apache.texera.amber.storage.result.iceberg.IcebergDocumentSpec org.apache.texera.amber.storage.result.iceberg.IcebergTableStatsSpec org.apache.texera.amber.storage.result.iceberg.IcebergDocumentConsoleMessagesSpec"
# 30 tests, all pass

sbt "WorkflowExecutionService/Test/testOnly org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResourceSpec"
# 2 tests, all pass

pytest core/util/virtual_identity/
# 17 tests, all pass

sbt "WorkflowCore/scalafmtCheck; WorkflowCore/Test/scalafmtCheck"
# clean
```

### Was this PR authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 4.7)

Author: aglinxinyuan

amber/src/main/python/core/util/virtual_identity.py

@@ -42,12 +42,27 @@ def serialize_global_port_identity(obj: GlobalPortIdentity) -> str:
     Expected format:
     ``(logicalOpId=<logicalOpId>,layerName=<layerName>,
     portId=<portId.id>,isInternal=<portId.internal>,isInput=<input>)``
+
+    Raises ValueError if `logicalOpId` or `layerName` contains an underscore
+    (VFS URI parsing relies on the absence of '_'), or if `portId` is negative.
     """
     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}"
+        )
+    if port_id < 0:
+        raise ValueError(f"portId must be non-negative: {port_id}")
     return (
         f"(logicalOpId={logical_op_id},layerName={layer_name},portId={port_id},"
         f"isInternal={str(is_internal).lower()},isInput={str(is_input_port).lower()})"
@@ -72,6 +87,8 @@ def deserialize_global_port_identity(encoded_str: str) -> GlobalPortIdentity:
         match.groups()
     )
     port_id = int(port_id_str)
+    if port_id < 0:
+        raise ValueError(f"portId must be non-negative: {port_id}")
     is_internal = is_internal_str.lower() == "true"
     is_input_port = is_input_str.lower() == "true"
     op_id = PhysicalOpIdentity(

amber/src/test/python/core/storage/iceberg/test_iceberg_document.py

@@ -86,7 +86,7 @@ def iceberg_document(self, amber_schema):
             ExecutionIdentity(id=0),
             GlobalPortIdentity(
                 op_id=PhysicalOpIdentity(
-                    logical_op_id=OperatorIdentity(id=f"test_table_{operator_uuid}"),
+                    logical_op_id=OperatorIdentity(id=f"test-table-{operator_uuid}"),
                     layer_name="main",
                 ),
                 port_id=PortIdentity(id=0),

amber/src/test/python/core/util/test_virtual_identity.py

@@ -105,6 +105,21 @@ def test_round_trips_through_deserialize(self):
         assert recovered.port_id.internal is True
         assert recovered.input is False
 
+    def test_rejects_underscore_in_logical_op_id(self):
+        # VFS-compatibility contract: serialized output must be
+        # underscore-free. Fail fast at the boundary on underscored input.
+        with pytest.raises(ValueError, match="logicalOpId must not contain"):
+            serialize_global_port_identity(_gpi(op_id="__DummyOperator"))
+
+    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_negative_port_id(self):
+        # Port ids are array indices and must be non-negative.
+        with pytest.raises(ValueError, match="portId must be non-negative"):
+            serialize_global_port_identity(_gpi(port=-1))
+
 
 class TestDeserializeGlobalPortIdentity:
     def test_parses_canonical_encoded_string(self):
@@ -137,6 +152,14 @@ def test_raises_value_error_on_missing_field(self):
                 "(logicalOpId=op,layerName=l,portId=0,isInternal=true)"
             )
 
+    def test_raises_value_error_on_negative_port_id(self):
+        # Symmetric with the serializer: tampered URIs with a negative
+        # portId must be rejected on the way back in.
+        with pytest.raises(ValueError, match="portId must be non-negative"):
+            deserialize_global_port_identity(
+                "(logicalOpId=op,layerName=l,portId=-1,isInternal=false,isInput=true)"
+            )
+
 
 class TestGetFromActorIdForInputPortStorage:
     def test_prefixes_materialization_reader_to_uri_plus_actor_name(self):

common/workflow-core/src/main/scala/org/apache/texera/amber/util/serde/GlobalPortIdentitySerde.scala

@@ -23,17 +23,20 @@ import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity}
 
 /**
   * Serialize and deserializes a GlobalPortIdentity object to a string using a custom, human-readable format
-  * to ensure it works with both URI and file path and does not incldue underscore "_" so that it does not
-  * interfere with our own VFS URI parsing.
+  * to ensure it works with both URI and file path and does not include underscore "_" so that it does not
+  * interfere with our own VFS URI parsing. Underscores in `logicalOpId` / `layerName` and negative `portId`
+  * values are rejected with `IllegalArgumentException`.
   */
 object GlobalPortIdentitySerde {
   implicit class SerdeOps(globalPortId: GlobalPortIdentity) {
 
     /**
       * Serializes a GlobalPortIdentity object into a string using our custom, human-readable format
-      * that works with both URI and file path and does not incldue underscore "_" so that it does not
+      * that works with both URI and file path and does not include underscore "_" so that it does not
       * interfere with our own VFS URI parsing.
       *
+      * @throws java.lang.IllegalArgumentException if `logicalOpId` or `layerName` contains an underscore,
+      *                                            or if `portId.id` is negative.
       * @return A serialized string representation of globalPortId
       */
     def serializeAsString: String = {
@@ -42,6 +45,15 @@ object GlobalPortIdentitySerde {
       val portId = globalPortId.portId.id
       val isInternal = globalPortId.portId.internal
       val isInput = globalPortId.input
+      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)"
     }
   }
@@ -58,6 +70,7 @@ object GlobalPortIdentitySerde {
     serializedGlobalPortId match {
       case pattern(logicalOpId, layerName, portIdStr, isInternalStr, isInputStr) =>
         val portIdInt = portIdStr.toInt
+        require(portIdInt >= 0, s"portId must be non-negative: $portIdInt")
         val isInternal = isInternalStr.toBoolean
         val isInput = isInputStr.toBoolean
         val physicalOpId = PhysicalOpIdentity(

common/workflow-core/src/test/scala/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala

@@ -85,7 +85,7 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
       GlobalPortIdentity(
         PhysicalOpIdentity(
           logicalOpId =
-            OperatorIdentity(s"test_table_${UUID.randomUUID().toString.replace("-", "")}"),
+            OperatorIdentity(s"test-table-${UUID.randomUUID().toString.replace("-", "")}"),
           layerName = "main"
         ),
         PortIdentity()

common/workflow-core/src/test/scala/org/apache/texera/amber/storage/result/iceberg/IcebergTableStatsSpec.scala

@@ -56,7 +56,7 @@ class IcebergTableStatsSpec extends AnyFlatSpec with BeforeAndAfterAll with Suit
     GlobalPortIdentity(
       PhysicalOpIdentity(
         logicalOpId =
-          OperatorIdentity(s"test_table_${UUID.randomUUID().toString.replace("-", "")}"),
+          OperatorIdentity(s"test-table-${UUID.randomUUID().toString.replace("-", "")}"),
         layerName = "main"
       ),
       PortIdentity()