diff --git a/CHANGES.md b/CHANGES.md index d1910513da58..c1e41f80f60a 100644 --- a/CHANGES.md +++ b/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 diff --git a/sdks/python/apache_beam/io/fileio.py b/sdks/python/apache_beam/io/fileio.py index fcce83fa59ec..1aac9f817f55 100644 --- a/sdks/python/apache_beam/io/fileio.py +++ b/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 diff --git a/sdks/python/apache_beam/io/fileio_test.py b/sdks/python/apache_beam/io/fileio_test.py index 1c650653804c..82425587ae4d 100644 --- a/sdks/python/apache_beam/io/fileio_test.py +++ b/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) diff --git a/sdks/python/apache_beam/transforms/periodicsequence.py b/sdks/python/apache_beam/transforms/periodicsequence.py index 7b90fa3bb2fa..cdd7c93b0dfe 100644 --- a/sdks/python/apache_beam/transforms/periodicsequence.py +++ b/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 diff --git a/sdks/python/apache_beam/transforms/periodicsequence_test.py b/sdks/python/apache_beam/transforms/periodicsequence_test.py index 367d36d8a4c9..55ae8f694647 100644 --- a/sdks/python/apache_beam/transforms/periodicsequence_test.py +++ b/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)