fix(pyamber, v1.2): preserve channel marker ports

amber/src/main/python/core/architecture/handlers/control/end_channel_handler.py

@@ -25,8 +25,12 @@
 
 class EndChannelHandler(ControlHandler):
     async def end_channel(self, req: EmptyRequest) -> EmptyReturn:
+        port = self.context.input_manager.get_port_id(
+            self.context.current_input_channel_id
+        )
+        marker = EndChannel(port.id)
         self.context.input_manager.complete_current_port(
             self.context.current_input_channel_id
         )
-        self.context.tuple_processing_manager.current_internal_marker = EndChannel()
+        self.context.tuple_processing_manager.current_internal_marker = marker
         return EmptyReturn()

amber/src/main/python/core/architecture/handlers/control/start_channel_handler.py

@@ -25,5 +25,10 @@
 
 class StartChannelHandler(ControlHandler):
     async def start_channel(self, req: EmptyRequest) -> EmptyReturn:
-        self.context.tuple_processing_manager.current_internal_marker = StartChannel()
+        port = self.context.input_manager.get_port_id(
+            self.context.current_input_channel_id
+        )
+        self.context.tuple_processing_manager.current_internal_marker = StartChannel(
+            port.id
+        )
         return EmptyReturn()

amber/src/main/python/core/models/internal_marker.py

@@ -16,18 +16,36 @@
 # under the License.
 
 
+from dataclasses import dataclass
+
+
 class InternalMarker:
     """
-    A special Data Message, only being generated in un-packaging a batch into Tuples.
-    Markers retain the order information and served as a indicator of data state.
+    An internal event produced by batch unpacking or control handlers.
+    Markers preserve ordering and signal input lifecycle changes.
     """
 
     pass
 
 
-class StartChannel(InternalMarker):
-    pass
+@dataclass(frozen=True)
+class PortMarker(InternalMarker):
+    """A control marker bound to the input port that produced it."""
 
+    port_id: int
 
-class EndChannel(InternalMarker):
-    pass
+    def __post_init__(self) -> None:
+        if (
+            not isinstance(self.port_id, int)
+            or isinstance(self.port_id, bool)
+            or self.port_id < 0
+        ):
+            raise ValueError("marker port_id must be a nonnegative integer")
+
+
+class StartChannel(PortMarker):
+    """Start-of-channel marker with immutable input-port provenance."""
+
+
+class EndChannel(PortMarker):
+    """End-of-channel marker with immutable input-port provenance."""

amber/src/main/python/core/runnables/data_processor.py

@@ -24,8 +24,13 @@
 from typing import Iterator, Optional
 
 from core.architecture.managers import Context
+<<<<<<< HEAD
 from core.models import ExceptionInfo, State, TupleLike, InternalMarker
 from core.models.internal_marker import StartChannel, EndChannel
+=======
+from core.models import State, TupleLike
+from core.models.internal_marker import EndChannel, PortMarker, StartChannel
+>>>>>>> 471e53cfc (fix(pyamber): preserve channel marker ports (#8432))
 from core.models.table import all_output_to_tuple
 from core.util import Stoppable
 from core.util.console_message.replace_print import replace_print
@@ -74,8 +79,10 @@ def run(self) -> None:
             else:
                 self.process_tuple()
 
-    def process_internal_marker(self, internal_marker: InternalMarker) -> None:
-        with self._executor_session() as (executor, port_id):
+    def process_internal_marker(self, internal_marker: PortMarker) -> None:
+        if not isinstance(internal_marker, PortMarker):
+            raise TypeError("expected a PortMarker")
+        with self._executor_session(internal_marker.port_id) as (executor, port_id):
             if isinstance(internal_marker, StartChannel):
                 self._set_output_state(executor.produce_state_on_start(port_id))
             elif isinstance(internal_marker, EndChannel):
@@ -104,7 +111,7 @@ def process_tuple(self) -> None:
                 self._set_output_tuple(executor.process_tuple(tuple_, port_id))
 
     @contextmanager
-    def _executor_session(self):
+    def _executor_session(self, marker_port_id: int | None = None):
         """
         Open one executor invocation: hand back (executor, port_id) under a
         print-capture session, route any exception to the exception manager
@@ -116,7 +123,11 @@ def _executor_session(self):
         """
         try:
             executor = self._context.executor_manager.executor
-            port_id = self._context.tuple_processing_manager.get_input_port_id()
+            port_id = (
+                self._context.tuple_processing_manager.get_input_port_id()
+                if marker_port_id is None
+                else marker_port_id
+            )
             with replace_print(
                 self._context.worker_id,
                 self._context.console_message_manager.print_buf,

amber/src/test/python/core/architecture/handlers/control/test_channel_marker_handlers.py

@@ -0,0 +1,126 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from core.architecture.handlers.control.end_channel_handler import EndChannelHandler
+from core.architecture.handlers.control.start_channel_handler import (
+    StartChannelHandler,
+)
+from core.architecture.packaging.input_manager import InputManager
+from core.models import Schema
+from core.models.internal_marker import EndChannel, StartChannel
+from core.models.internal_queue import InternalQueue
+from proto.org.apache.texera.amber.core import (
+    ActorVirtualIdentity,
+    ChannelIdentity,
+    PortIdentity,
+)
+from proto.org.apache.texera.amber.engine.architecture.rpc import EmptyRequest
+
+WORKER_ID = "worker-1"
+UPSTREAM_ID = ActorVirtualIdentity("upstream-worker")
+
+
+def _channel(index: int) -> ChannelIdentity:
+    """Return a distinct data-channel identity."""
+    return ChannelIdentity(
+        ActorVirtualIdentity(f"{UPSTREAM_ID.name}-{index}"),
+        ActorVirtualIdentity(WORKER_ID),
+        False,
+    )
+
+
+def _build_context(*port_numbers: int):
+    """Build real input state and the minimal marker-handler context."""
+    manager = InputManager(WORKER_ID, InternalQueue())
+    channels = []
+    ports = []
+    for index, port_number in enumerate(port_numbers):
+        channel = _channel(index)
+        port = PortIdentity(port_number, False)
+        manager.add_input_port(port, Schema(), [], [])
+        manager.register_input(channel, port)
+        channels.append(channel)
+        ports.append(port)
+    context = SimpleNamespace(
+        input_manager=manager,
+        current_input_channel_id=channels[-1] if channels else _channel(99),
+        tuple_processing_manager=SimpleNamespace(current_internal_marker=None),
+    )
+    return context, channels, ports
+
+
+@pytest.mark.parametrize(
+    ("handler_type", "method_name", "marker_type"),
+    [
+        (StartChannelHandler, "start_channel", StartChannel),
+        (EndChannelHandler, "end_channel", EndChannel),
+    ],
+)
+def test_channel_marker_snapshots_the_current_ports_identity(
+    handler_type, method_name, marker_type
+):
+    """Each channel retains its own port."""
+    context, channels, ports = _build_context(3, 7)
+    handler = handler_type(context)
+
+    context.current_input_channel_id = channels[1]
+    asyncio.run(getattr(handler, method_name)(EmptyRequest()))
+
+    assert context.tuple_processing_manager.current_internal_marker == marker_type(
+        ports[1].id
+    )
+    assert context.input_manager.get_port(ports[0]).completed is False
+    assert context.input_manager.get_port(ports[1]).completed is (
+        handler_type is EndChannelHandler
+    )
+
+
+@pytest.mark.parametrize(
+    ("handler_type", "method_name"),
+    [
+        (StartChannelHandler, "start_channel"),
+        (EndChannelHandler, "end_channel"),
+    ],
+)
+def test_unknown_channel_leaves_marker_and_known_ports_untouched(
+    handler_type, method_name
+):
+    """A missing channel mapping must fail before producing any side effect."""
+    context, _, ports = _build_context(3)
+    context.current_input_channel_id = _channel(99)
+
+    with pytest.raises(KeyError):
+        asyncio.run(getattr(handler_type(context), method_name)(EmptyRequest()))
+
+    assert context.tuple_processing_manager.current_internal_marker is None
+    assert context.input_manager.get_port(ports[0]).completed is False
+
+
+def test_invalid_end_channel_port_fails_before_completing_the_port():
+    """An invalid marker must not complete its port."""
+    context, _, ports = _build_context(-1)
+
+    with pytest.raises(ValueError, match="nonnegative integer"):
+        asyncio.run(EndChannelHandler(context).end_channel(EmptyRequest()))
+
+    assert context.tuple_processing_manager.current_internal_marker is None
+    assert context.input_manager.get_port(ports[0]).completed is False