[Python] Add UnboundedSource SDF wrapper (#19137)

Brings Java's ``UnboundedSource`` / ``UnboundedReader`` / ``CheckpointMark``
abstractions to the Python SDK as a Splittable-DoFn wrapper runnable on the
portable Fn API (DirectRunner / FnApiRunner). Wires the new source type into
``iobase.Read.expand()`` so ``p | beam.io.Read(my_unbounded_source)``
dispatches alongside the existing ``BoundedSource`` branch. Loosely inspired
by Java's ``Read.UnboundedSourceAsSDFWrapperFn``; the streaming-SDF template
followed for the process loop / watermark / defer plumbing is
``apache_beam.transforms.periodicsequence``.

addresses #19137

What's added
------------

``sdks/python/apache_beam/io/unbounded_source.py``

  Public ABCs (``CheckpointMark``, ``UnboundedReader``, ``UnboundedSource``,
  ``ReadFromUnboundedSource``) plus the SDF wrapper internals
  (``_UnboundedSourceRestriction``, ``_UnboundedSourceRestrictionCoder``,
  ``_UnboundedSourceRestrictionTracker``,
  ``_UnboundedSourceRestrictionProvider``).

``sdks/python/apache_beam/io/unbounded_source_test.py``

  42 deterministic tests covering ABC contracts, restriction coder
  round-trip, tracker state machine (claim / split / EOF / no-data /
  check_done / progress / is_bounded), finalize idempotency, fan-out
  via ``source.split``, source-watermark vs. record-event-time, finalize
  vs. resume channel separation, tracker-internal exception close on
  reader-method failures, DoFn generator close (unit + integration with
  downstream raising ``Map``), cloudpickle round-trip, circular import
  in three subprocess orderings, and an end-to-end DirectRunner pipeline.

What's changed in iobase.py
---------------------------

* ``Read.expand`` gains an ``UnboundedSource`` branch (function-local lazy
  import to break the iobase <-> unbounded_source cycle) that delegates to
  ``ReadFromUnboundedSource``.
* ``Read.to_runner_api_parameter`` widens the source ``isinstance`` to
  ``(BoundedSource, UnboundedSource)``, writing ``READ.urn`` +
  ``ReadPayload(is_bounded=UNBOUNDED)``. Decode rides the existing
  ``PICKLED_SOURCE`` URN on ``SourceBase``. Runner-side
  ``IsBounded.UNBOUNDED`` dispatch in
  ``bundle_processor.IMPULSE_READ_TRANSFORM`` remains W2 -- execution
  flows through the composite's expanded ``Impulse | Map | SDF-ParDo``.

Correctness covered
-------------------

* Data-path watermark uses ``reader.get_watermark()`` (Java
  ``Read.java:594`` parity), not the per-record event time. Holder is
  ``(value, record_ts, source_wm)``.
* Restriction has separate ``checkpoint_mark`` (resume) and
  ``finalization_checkpoint_mark`` (commit hook) channels; coder is a
  fixed 5-tuple.
* Reader is closed on every exit path -- tracker-internal close on EOF
  / split / reader-method exception; DoFn ``finally`` defense-in-depth
  for yield / downstream raise via the SDF wrapper's private chain
  (isinstance guard + warning log if the chain ever moves upstream).
* EOF advances the watermark estimator to ``MAX_TIMESTAMP`` so
  downstream event-time windows can close.
* ``UnboundedSource.split(desired_num_splits=20, options)`` is honoured;
  returned sub-sources are validated as ``UnboundedSource`` (raises
  ``TypeError`` outside the split-refusal ``except``); on
  split-exception the provider falls back to a single restriction with
  WARNING.
* ``default_output_coder`` reaches the output PCollection via
  ``coders.registry.register_coder`` + ``element_type``.
* ``ReadFromUnboundedSource`` validates ``poll_interval_seconds > 0``.

Out of scope (tracked under #19137)
-----------------------------------

Listed exhaustively in the module docstring at
``sdks/python/apache_beam/io/unbounded_source.py``:

* Record-id-based deduplication (Java's ``ValueWithRecordId``).
* Backlog-byte reporting (``restriction_size`` is constant 1;
  ``current_progress`` is binary 0.0 / 1.0).
* Dynamic split fractions / runner-initiated work stealing.
* Source-specific checkpoint coders threaded through the SDF
  restriction coder (checkpoint marks are pickled today regardless of
  the source's ``get_checkpoint_mark_coder``).
* Reader caching across bundles (Java uses a Guava cache).
* ``EmptyUnboundedSource`` terminal-state marker (this PoC uses an
  ``is_done`` flag).
* Runner-side ``IsBounded.UNBOUNDED`` dispatch in
  ``bundle_processor.IMPULSE_READ_TRANSFORM``.

Tests: 42/42 ``unbounded_source_test.py``, 16/16 ``iobase_test.py``,
yapf + isort clean.

Author: Eliaaazzz

sdks/python/apache_beam/io/iobase.py

@@ -919,7 +919,11 @@ def __init__(self, source: SourceBase) -> None:
     """Initializes a Read transform.
 
     Args:
-      source: Data source to read from.
+      source: Data source to read from. A ``BoundedSource`` is wrapped in the
+        bounded SDF reader; an ``UnboundedSource`` is dispatched through
+        :class:`apache_beam.io.unbounded_source.ReadFromUnboundedSource` with
+        the default poll interval (users wanting a custom poll cadence must
+        instantiate ``ReadFromUnboundedSource`` directly).
     """
     super().__init__()
     self.source = source
@@ -945,6 +949,16 @@ def expand(self, pbegin):
           | 'EmitSource' >>
           core.Map(lambda _: self.source).with_output_types(BoundedSource)
           | SDFBoundedSourceReader(display_data))
+    # Lazy import to break the iobase <-> unbounded_source cycle: the
+    # unbounded_source module imports iobase (UnboundedSource extends
+    # SourceBase). Pattern matches the _PubSubSource lazy import below.
+    from apache_beam.io.unbounded_source import ReadFromUnboundedSource
+    from apache_beam.io.unbounded_source import UnboundedSource
+    if isinstance(self.source, UnboundedSource):
+      # Delegate to the dedicated SDF PTransform; identical to the user
+      # writing `p | ReadFromUnboundedSource(self.source)` directly. Custom
+      # poll_interval_seconds requires using ReadFromUnboundedSource directly.
+      return pbegin | ReadFromUnboundedSource(self.source)
     elif isinstance(self.source, ptransform.PTransform):
       # The Read transform can also admit a full PTransform as an input
       # rather than an anctual source. If the input is a PTransform, then
@@ -986,7 +1000,15 @@ def to_runner_api_parameter(
               timestamp_attribute=self.source.timestamp_attribute,
               with_attributes=self.source.with_attributes,
               id_attribute=self.source.id_label))
-    if isinstance(self.source, BoundedSource):
+    # Lazy import to avoid the iobase <-> unbounded_source cycle.
+    from apache_beam.io.unbounded_source import UnboundedSource
+    if isinstance(self.source, (BoundedSource, UnboundedSource)):
+      # READ.urn covers both source flavours; the IsBounded enum distinguishes
+      # them. NB: today the bundle_processor.py IMPULSE_READ_TRANSFORM handler
+      # only consumes BOUNDED - the UNBOUNDED branch round-trips correctly
+      # through the protobuf graph but execution still flows through this
+      # composite's expanded sub-transforms (Impulse | Map | SDF-ParDo), not
+      # through the URN-handler. Runner-side UNBOUNDED dispatch is W2 work.
       return (
           common_urns.deprecated_primitives.READ.urn,
           beam_runner_api_pb2.ReadPayload(

sdks/python/apache_beam/io/iobase_test.py

@@ -220,5 +220,83 @@ def test_sdf_wrap_range_source(self):
     self._run_sdf_wrapper_pipeline(RangeSource(0, 4), [0, 1, 2, 3])
 
 
+class UseSdfUnboundedSourcesTests(unittest.TestCase):
+  """Mirrors UseSdfBoundedSourcesTests for the new UnboundedSource branch in
+  iobase.Read.expand(). Uses CountingSource from unbounded_source_test as the
+  fake finite UnboundedSource (avoids dragging the network in).
+  """
+
+  def test_read_dispatches_to_read_from_unbounded_source(self):
+    from apache_beam.io.unbounded_source_test import CountingSource
+    with mock.patch(
+        'apache_beam.io.unbounded_source.ReadFromUnboundedSource.expand'
+    ) as mock_expand:
+      mock_expand.side_effect = (
+          lambda pbegin: pbegin | beam.Impulse() | beam.Map(lambda _: 'fake'))
+      with beam.Pipeline() as p:
+        out = p | beam.io.Read(CountingSource(3))
+        assert_that(out, equal_to(['fake']))
+      mock_expand.assert_called_once()
+
+  def test_read_end_to_end_unbounded(self):
+    from apache_beam.io.unbounded_source_test import CountingSource
+    with beam.Pipeline() as p:
+      out = p | beam.io.Read(CountingSource(5))
+      assert_that(out, equal_to([0, 1, 2, 3, 4]))
+
+  def test_read_unbounded_pcollection_is_unbounded(self):
+    from apache_beam.io.unbounded_source_test import CountingSource
+    with beam.Pipeline() as p:
+      out = p | beam.io.Read(CountingSource(3))
+      self.assertFalse(out.is_bounded)
+
+  def test_to_runner_api_emits_unbounded_read_payload(self):
+    """``Read.to_runner_api_parameter`` must serialize an UnboundedSource as
+    ``READ.urn`` with ``IsBounded.UNBOUNDED``. The runner-side handler is W2
+    and ignores this enum today, but the wire format must round-trip
+    consistently for pipeline persistence / cross-runner submission.
+    """
+    from apache_beam.io.unbounded_source_test import CountingSource
+    from apache_beam.portability import common_urns
+    from apache_beam.portability.api import beam_runner_api_pb2
+    from apache_beam.runners.pipeline_context import PipelineContext
+
+    read = beam.io.Read(CountingSource(5))
+    urn, payload = read.to_runner_api_parameter(PipelineContext())
+
+    self.assertEqual(urn, common_urns.deprecated_primitives.READ.urn)
+    self.assertIsInstance(payload, beam_runner_api_pb2.ReadPayload)
+    self.assertEqual(
+        payload.is_bounded, beam_runner_api_pb2.IsBounded.UNBOUNDED)
+    # The source field must be populated -- a non-empty FunctionSpec proto.
+    self.assertTrue(payload.source.urn)
+
+  def test_read_unbounded_round_trips_through_runner_api(self):
+    """Encode then decode via the runner-API protobuf. The restored
+    transform must be a ``Read`` wrapping an equivalent UnboundedSource.
+    """
+    from apache_beam.io.unbounded_source import UnboundedSource
+    from apache_beam.io.unbounded_source_test import CountingSource
+    from apache_beam.portability import common_urns
+    from apache_beam.portability.api import beam_runner_api_pb2
+    from apache_beam.runners.pipeline_context import PipelineContext
+
+    original = beam.io.Read(CountingSource(7))
+    context = PipelineContext()
+    urn, payload = original.to_runner_api_parameter(context)
+
+    transform_proto = beam_runner_api_pb2.PTransform()
+    transform_proto.spec.urn = urn
+    restored = iobase.Read.from_runner_api_parameter(
+        transform_proto, payload, context)
+
+    self.assertIsInstance(restored, iobase.Read)
+    self.assertIsInstance(restored.source, UnboundedSource)
+    self.assertIsInstance(restored.source, CountingSource)
+    self.assertFalse(restored.source.is_bounded())
+    # Verify the source's internal state survived pickle round-trip.
+    self.assertEqual(restored.source._count, 7)
+
+
 if __name__ == '__main__':
   unittest.main()