[Python] Default PeriodicImpulse start_timestamp to creation time

CHANGES.md

@@ -105,6 +105,7 @@
 * (Python) `ReadFromText` with `escapechar` no longer skips a delimiter that directly follows an escaped delimiter, which merged two records into one ([#40459](https://github.com/apache/beam/issues/40459)).
 * (Python) DataFrame API: `agg` with a list that includes `'count'` or `'size'` now labels those rows `count`/`size` instead of `sum` ([#40504](https://github.com/apache/beam/issues/40504)).
 * (Python) DataFrame API: `Series.value_counts(bins=...)` no longer always fails with a `TypeError` ([#40503](https://github.com/apache/beam/issues/40503)).
+* (Python) `PeriodicImpulse` and `MatchContinuously` now default `start_timestamp` to the time the transform is created instead of the time the module was imported, which emitted a burst of backdated elements when the transform was created later ([#40564](https://github.com/apache/beam/issues/40564)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes

sdks/python/apache_beam/io/fileio.py

@@ -392,7 +392,7 @@ def __init__(
       file_pattern,
       interval=360.0,
       has_deduplication=True,
-      start_timestamp=Timestamp.now(),
+      start_timestamp=None,
       stop_timestamp=MAX_TIMESTAMP,
       match_updated_files=False,
       apply_windowing=False,
@@ -404,7 +404,8 @@ def __init__(
       file_pattern: The file path to read from.
       interval: Interval at which to check for files in seconds.
       has_deduplication: Whether files already read are discarded or not.
-      start_timestamp: Timestamp for start file checking.
+      start_timestamp: Timestamp for start file checking. Defaults to the time
+        the transform is created.
       stop_timestamp: Timestamp after which no more files will be checked.
       match_updated_files: (When has_deduplication is set to True) whether match
         file with timestamp changes.
@@ -421,7 +422,8 @@ def __init__(
     self.file_pattern = file_pattern
     self.interval = interval
     self.has_deduplication = has_deduplication
-    self.start_ts = start_timestamp
+    self.start_ts = (
+        start_timestamp if start_timestamp is not None else Timestamp.now())
     self.stop_ts = stop_timestamp
     self.match_upd = match_updated_files
     self.apply_windowing = apply_windowing

sdks/python/apache_beam/io/fileio_test.py

@@ -27,6 +27,7 @@
 import unittest
 import uuid
 import warnings
+from unittest import mock
 
 import pytest
 from hamcrest.library.text import stringmatches
@@ -387,6 +388,28 @@ def _create_extra_file(element):
 
       assert_that(match_continiously, equal_to(files))
 
+  def test_default_start_timestamp(self):
+    # The default start_timestamp is the time the transform is created, not
+    # the time the module was imported.
+    tempdir = '%s%s' % (self._new_tempdir(), os.sep)
+    file = self._create_temp_file(dir=tempdir)
+    interval = 0.2
+    start = Timestamp(seconds=1)
+    stop = start + interval + 0.1
+
+    with mock.patch.object(Timestamp, 'now', return_value=start):
+      match = fileio.MatchContinuously(
+          file_pattern=FileSystems.join(tempdir, '*'),
+          interval=interval,
+          has_deduplication=False,
+          stop_timestamp=stop)
+
+    with TestPipeline() as p:
+      match_continiously = (
+          p | match | beam.Map(lambda metadata: metadata.path))
+
+      assert_that(match_continiously, equal_to([file, file]))
+
   def test_match_updated_files(self):
     files = []
     tempdir = '%s%s' % (self._new_tempdir(), os.sep)

sdks/python/apache_beam/transforms/periodicsequence.py

@@ -292,14 +292,15 @@ def _validate_and_adjust_duration(self):
 
   def __init__(
       self,
-      start_timestamp: TimestampTypes = Timestamp.now(),
+      start_timestamp: Optional[TimestampTypes] = None,
       stop_timestamp: TimestampTypes = MAX_TIMESTAMP,
       fire_interval: float = 360.0,
       apply_windowing: bool = False,
       data: Optional[Sequence[Any]] = None,
       rebase: RebaseMode = RebaseMode.REBASE_NONE):
     '''
-    :param start_timestamp: Timestamp for first element.
+    :param start_timestamp: Timestamp for first element. Defaults to the time
+      the transform is created.
     :param stop_timestamp: Timestamp at or after which no elements will be
       output.
     :param fire_interval: Interval in seconds at which to output elements.
@@ -333,7 +334,8 @@ def __init__(
       execution time. See `RebaseMode` for more details. Defaults to
       `REBASE_NONE`.
     '''
-    self.start_ts = start_timestamp
+    self.start_ts = (
+        start_timestamp if start_timestamp is not None else Timestamp.now())
     self.stop_ts = stop_timestamp
     self.interval = fire_interval
     self.apply_windowing = apply_windowing

sdks/python/apache_beam/transforms/periodicsequence_test.py

@@ -19,11 +19,11 @@
 
 # pytype: skip-file
 
-import inspect
 import logging
 import random
 import time
 import unittest
+from unittest import mock
 
 from parameterized import parameterized
 
@@ -130,20 +130,18 @@ def test_windowing_on_si(self):
       assert_that(actual, equal_to(k))
 
   def test_default_start(self):
-    default_parameters = inspect.signature(PeriodicImpulse.__init__).parameters
-    it = default_parameters["start_timestamp"].default
+    # The default start_timestamp is the time the transform is created, not
+    # the time this module was imported.
+    it = Timestamp(seconds=1)
     duration = 1
     et = it + duration
     interval = 0.5
 
-    # Check default `stop_timestamp` is the same type `start_timestamp`
-    is_same_type = isinstance(
-        it, type(default_parameters["stop_timestamp"].default))
-    error = "'start_timestamp' and 'stop_timestamp' have different type"
-    assert is_same_type, error
+    with mock.patch.object(Timestamp, 'now', return_value=it):
+      impulse = PeriodicImpulse(stop_timestamp=et, fire_interval=interval)
 
     with TestPipeline() as p:
-      result = p | 'PeriodicImpulse' >> PeriodicImpulse(it, et, interval)
+      result = p | 'PeriodicImpulse' >> impulse
 
       k = [it + x * interval for x in range(0, int(duration / interval))]
       self.assertEqual(result.is_bounded, False)