[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``

  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,
  per-bundle read caps, 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)``. The ``BoundedSource`` branch keeps
  its ``is_bounded()`` conditional so a source reporting unbounded still
  serializes as UNBOUNDED. Decode rides the existing ``PICKLED_SOURCE`` URN
  on ``SourceBase``.

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

* Data-path watermark uses ``reader.get_watermark()`` (Java ``Read.java:594``
  parity), not the per-record event time. Each record is emitted before the
  watermark estimator advances, so a reader that reports ``MAX_TIMESTAMP`` on
  the same claim as its final record cannot push the output watermark past
  that record.
* Restriction has separate ``checkpoint_mark`` (resume) and
  ``finalization_checkpoint_mark`` (commit hook) channels; coder is a fixed
  5-tuple. The split residual gets its own clone of the checkpoint mark so a
  ``finalize_checkpoint()`` that mutates the primary's mark cannot corrupt the
  residual's resume position.
* A continuously busy reader cannot stay in ``process()`` forever: the
  has-data path is capped at ``max_records_per_bundle`` (10000) or
  ``max_read_time_seconds`` (10), then self-checkpoints via
  ``defer_remainder`` so the runner commits progress and resumes. Caps are
  validated at the transform boundary, and the read deadline is armed on the
  first emitted record so reader startup is excluded. ``UnboundedReader.advance``
  must not block, since the caps are enforced only between records.
* 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.
* ``finalize_checkpoint()`` failures are logged and swallowed to match the
  ``CheckpointMark`` best-effort contract; finalize is idempotent on retry.
* 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``; on
  split-exception the provider falls back to a single restriction.
* ``default_output_coder`` reaches the output PCollection via
  ``coders.registry.register_coder`` + ``element_type``.

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

Listed exhaustively in the module docstring:

* Record-id-based deduplication (Java's ``ValueWithRecordId``).
* Backlog-byte reporting (``restriction_size`` is constant 1).
* Dynamic split fractions / runner-initiated work stealing.
* Source-specific checkpoint coders threaded through the restriction coder.
* Reader caching across bundles.
* Runner-side ``IsBounded.UNBOUNDED`` dispatch in
  ``bundle_processor.IMPULSE_READ_TRANSFORM``.

Author: Eliaaazzz

sdks/python/apache_beam/io/iobase.py

@@ -918,7 +918,10 @@ def __init__(self, source: SourceBase) -> None:
     """Initializes a Read transform.
 
     Args:
-      source: Data source to read from.
+      source: the data source to read from. May be a ``BoundedSource``, an
+        ``UnboundedSource``, or a ``PTransform`` (which is applied directly).
+        For any other source ``Read`` is treated as a primitive and relayed to
+        the runner implementation.
     """
     super().__init__()
     self.source = source
@@ -944,6 +947,11 @@ def expand(self, pbegin):
           | 'EmitSource' >>
           core.Map(lambda _: self.source).with_output_types(BoundedSource)
           | SDFBoundedSourceReader(display_data))
+    # Local import to avoid a circular dependency.
+    from apache_beam.io.unbounded_source import ReadFromUnboundedSource
+    from apache_beam.io.unbounded_source import UnboundedSource
+    if isinstance(self.source, UnboundedSource):
+      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
@@ -993,6 +1001,10 @@ def to_runner_api_parameter(
               is_bounded=beam_runner_api_pb2.IsBounded.BOUNDED
               if self.source.is_bounded() else
               beam_runner_api_pb2.IsBounded.UNBOUNDED))
+    # Local import to avoid a circular dependency.
+    from apache_beam.io.unbounded_source import UnboundedSource
+    if isinstance(self.source, UnboundedSource):
+      return super().to_runner_api_parameter(context)
     elif isinstance(self.source, ptransform.PTransform):
       return self.source.to_runner_api_parameter(context)
     raise NotImplementedError(

sdks/python/apache_beam/io/iobase_test.py

@@ -21,15 +21,16 @@
 
 import unittest
 
-import mock
-
 import apache_beam as beam
-from apache_beam.io.concat_source import ConcatSource
-from apache_beam.io.concat_source_test import RangeSource
+import mock
 from apache_beam.io import iobase
 from apache_beam.io import range_trackers
+from apache_beam.io.concat_source import ConcatSource
+from apache_beam.io.concat_source_test import RangeSource
 from apache_beam.io.iobase import SourceBundle
 from apache_beam.options.pipeline_options import DebugOptions
+from apache_beam.portability import common_urns
+from apache_beam.portability import python_urns
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
 
@@ -220,5 +221,45 @@ def test_sdf_wrap_range_source(self):
     self._run_sdf_wrapper_pipeline(RangeSource(0, 4), [0, 1, 2, 3])
 
 
+class UseSdfUnboundedSourcesTests(unittest.TestCase):
+  """Covers the UnboundedSource branch in
+  ``iobase.Read.expand()``. Uses ``UnboundedCountingSource`` from
+  ``unbounded_source_test`` as a finite fake source (no network).
+  """
+  def test_read_end_to_end_unbounded(self):
+    from apache_beam.io.unbounded_source_test import UnboundedCountingSource
+    with beam.Pipeline() as p:
+      out = p | beam.io.Read(UnboundedCountingSource(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 UnboundedCountingSource
+    p = beam.Pipeline()
+    out = p | beam.io.Read(UnboundedCountingSource(3))
+    self.assertFalse(out.is_bounded)
+
+  def test_read_unbounded_serializes_as_expanded_composite(self):
+    from apache_beam.io.unbounded_source_test import UnboundedCountingSource
+    p = beam.Pipeline()
+    p | 'ReadIt' >> beam.io.Read(UnboundedCountingSource(3))
+
+    proto = p.to_runner_api(use_fake_coders=True)
+    transforms = proto.components.transforms.values()
+    deprecated_reads = [
+        transform.unique_name for transform in transforms
+        if transform.spec.urn == common_urns.deprecated_primitives.READ.urn
+    ]
+    read_transforms = [
+        transform for transform in proto.components.transforms.values()
+        if transform.unique_name == 'ReadIt'
+    ]
+
+    self.assertEqual([], deprecated_reads)
+    self.assertEqual(1, len(read_transforms))
+    self.assertEqual(
+        python_urns.GENERIC_COMPOSITE_TRANSFORM, read_transforms[0].spec.urn)
+    self.assertTrue(read_transforms[0].subtransforms)
+
+
 if __name__ == '__main__':
   unittest.main()