{"record":{"id":"77565346f9571032","repo":"apache/beam","slug":"set-watermark-expects-a-timestamp-as-input","errorCode":null,"errorMessage":"set_watermark expects a Timestamp as input","messagePattern":"set_watermark expects a Timestamp as input","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/bigquery_change_history.py","lineNumber":264,"sourceCode":"\n  State is checkpointed as (watermark_hold, last_end) so\n  both values survive SDF re-dispatch.\n  \"\"\"\n  def __init__(self, state: tuple[Timestamp, Timestamp]) -> None:\n    self._watermark_hold, self._last_end = state\n\n  def observe_timestamp(self, timestamp: Timestamp) -> None:\n    pass\n\n  def current_watermark(self) -> Timestamp:\n    return self._watermark_hold\n\n  def get_estimator_state(self) -> tuple[Timestamp, Timestamp]:\n    return (self._watermark_hold, self._last_end)\n\n  def set_watermark(self, timestamp: Timestamp) -> None:\n    if not isinstance(timestamp, Timestamp):\n      raise ValueError('set_watermark expects a Timestamp as input')\n    if self._watermark_hold and self._watermark_hold > timestamp:\n      raise ValueError(\n          'Watermark must be monotonically increasing. '\n          'Provided %s < current %s' % (timestamp, self._watermark_hold))\n    self._watermark_hold = timestamp\n\n  def advance_poll_cursor(self, end: Timestamp) -> None:\n    \"\"\"Record end so the next poll starts from here.\n\n    Only advances forward: if end is earlier than the current cursor\n    (e.g. BQ clock regression), the cursor stays put so the next poll\n    doesn't re-query an already-covered range.\n    \"\"\"\n    self._last_end = max(self._last_end, end)\n\n  def poll_cursor(self) -> Timestamp:\n    \"\"\"Return the start Timestamp for the next poll.\"\"\"\n    return self._last_end","sourceCodeStart":246,"sourceCodeEnd":282,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/bigquery_change_history.py#L246-L282","documentation":"The watermark holder in the BigQuery change-history source requires the new watermark to be an apache_beam Timestamp instance. Passing any other type (int epoch, datetime.datetime, string) is rejected with ValueError because downstream comparison and watermark propagation assume Timestamp semantics.","triggerScenarios":"Calling set_watermark(x) on the estimator where x is not a Timestamp instance, e.g. set_watermark(datetime.datetime.utcnow()) or set_watermark(time.time()) — typically from custom code or a custom trigger wired into _emit_query_ranges' watermark handling.","commonSituations":"Users converting datetime objects but forgetting Timestamp.from_utc_datetime; passing epoch floats/ints from external systems; type confusion between apache_beam.utils.timestamp.Timestamp and other timestamp types.","solutions":["Wrap the value: set_watermark(Timestamp.from_utc_datetime(dt)) for datetime inputs.","For epoch seconds, use Timestamp(seconds=epoch) before calling set_watermark.","If you already hold a Timestamp, check you are not shadowing the class with a different import."],"exampleFix":"// before\nstate.set_watermark(datetime.datetime.now(datetime.timezone.utc))\n// after\nfrom apache_beam.utils.timestamp import Timestamp\nstate.set_watermark(Timestamp.from_utc_datetime(datetime.datetime.now(datetime.timezone.utc)))","handlingStrategy":"type-guard","validationCode":"from apache_beam.utils.timestamp import Timestamp\nassert isinstance(ts, Timestamp), f\"expected Timestamp, got {type(ts).__name__}\"","typeGuard":"def is_beam_timestamp(value) -> bool:\n    from apache_beam.utils.timestamp import Timestamp\n    return isinstance(value, Timestamp)","tryCatchPattern":"try:\n    state.set_watermark(ts)\nexcept ValueError as e:\n    log.warning(\"watermark rejected: %s\", e)","preventionTips":["Convert datetimes with Timestamp.from_utc_datetime and epochs with Timestamp(seconds=...) before use","Avoid passing datetime.datetime or int where Beam Timestamp is expected","Import Timestamp from apache_beam.utils.timestamp only"],"tags":["python","apache-beam","bigquery","timestamp","type-mismatch"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}