[Python] Default PeriodicImpulse start_timestamp to creation time

[raams95]

Fixes #40564

PeriodicImpulse and MatchContinuously declared start_timestamp=Timestamp.now() as a default argument. Python evaluates that once, when the module is imported, so the default start was the import time of apache_beam, not the time the transform was created. When the transform is created later (a long-running launcher process, a notebook kernel, slow pipeline setup), ImpulseSeqGenDoFn starts in the past and emits a burst of catch-up elements with timestamps from before the pipeline existed:

import time
import apache_beam as beam
from apache_beam.transforms.periodicsequence import PeriodicImpulse
from apache_beam.utils.timestamp import Timestamp

time.sleep(10)

built_at = Timestamp.now()
with beam.Pipeline() as p:
  _ = (
      p
      | PeriodicImpulse(stop_timestamp=built_at + 3, fire_interval=1)
      | beam.Map(print))

On master this prints 14 elements, starting 10 seconds before built_at. With this change it prints the 3 expected ones, built_at, built_at + 1 and built_at + 2.

The default is now None, and __init__ resolves it to Timestamp.now(). This matches what #35569 assumed when it added rebase: that a Timestamp.now() start is "the timestamp when the PTransform is expanded (i.e. the pipeline's construction time)". Explicit start_timestamp values and the rebase modes behave as before. MatchContinuously had the same default, and ReadAllFromTextContinuously / ReadAllFromAvroContinuously pass through to it, so it gets the same fix.

Tests:

• periodicsequence_test.test_default_start read the default from inspect.signature, so it tested the import-time value. It now creates the transform with Timestamp.now patched to a fixed time and checks the pipeline output starts there.
• fileio_test.MatchContinuouslyTest.test_default_start_timestamp does the same for MatchContinuously(has_deduplication=False).

Both fail before the change (the default start is the import time, which is later than the given stop, so initial_restriction raises AssertionError) and pass after it.

Testing:

• pytest apache_beam/transforms/periodicsequence_test.py apache_beam/io/fileio_test.py (71 passed, 3 skipped)
• pytest apache_beam/io/textio_test.py apache_beam/io/avroio_test.py -k Continuously
• yapf --diff and ruff check on the changed files

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123 or fixes #456).
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement. (Not needed, small change.)

See the Contributor Guide for more tips on how to make review process smoother.

[github-actions]

Assigning reviewers:

R: @jrmccluskey for label python.
Backup reviewer: tvalentyn (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

This pull request likely touches a core component ("core" label). Please review with scrutiny.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[shunping]

Hi @raams95, thank you for reporting this issue and submitting a fix.

After looking closer at the problem, I see that you are highlighting two distinct issues here:

• The burst of catch-up elements: As you noted, this behavior can already be addressed by explicitly using the rebase parameter.
• The inaccurate assumption in Add a flag in PeriodicImpulse to rebase timestamp. #35569: You are completely right about this. The description there incorrectly assumes the default is the transform creation time, when in reality it evaluates at the module import time. (I have already fixed the description in that PR.)

We should definitely fix the docstring to reflect reality, and we can do that in a separate PR. However, we shouldn't change the actual default behavior of PeriodicImpulse as proposed here, since it is widely used and changing the default would be a breaking change for existing pipelines.

Thanks again for bringing this up and catching the documentation error!

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 79.94%. Comparing base (76f6e6d) to head (1453f42).
⚠️ Report is 163 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #40565       +/-   ##
=============================================
+ Coverage     56.16%   79.94%   +23.77%     
=============================================
  Files          1124      488      -636     
  Lines        178336    82347    -95989     
  Branches       1489        0     -1489     
=============================================
- Hits         100170    65831    -34339     
+ Misses        75645    16516    -59129     
+ Partials       2521        0     -2521     

Flag	Coverage Δ
go	?
java	?
python	79.94% <100.00%> (+0.38%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[raams95]

Thanks, that makes sense. Opened #40599 to document the import-time default in the docstrings.