apache/beam · error · ValueError

set_watermark expects a Timestamp as input

Error message

set_watermark expects a Timestamp as input

What it means

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.

Source

Thrown at sdks/python/apache_beam/io/gcp/bigquery_change_history.py:264

  State is checkpointed as (watermark_hold, last_end) so
  both values survive SDF re-dispatch.
  """
  def __init__(self, state: tuple[Timestamp, Timestamp]) -> None:
    self._watermark_hold, self._last_end = state

  def observe_timestamp(self, timestamp: Timestamp) -> None:
    pass

  def current_watermark(self) -> Timestamp:
    return self._watermark_hold

  def get_estimator_state(self) -> tuple[Timestamp, Timestamp]:
    return (self._watermark_hold, self._last_end)

  def set_watermark(self, timestamp: Timestamp) -> None:
    if not isinstance(timestamp, Timestamp):
      raise ValueError('set_watermark expects a Timestamp as input')
    if self._watermark_hold and self._watermark_hold > timestamp:
      raise ValueError(
          'Watermark must be monotonically increasing. '
          'Provided %s < current %s' % (timestamp, self._watermark_hold))
    self._watermark_hold = timestamp

  def advance_poll_cursor(self, end: Timestamp) -> None:
    """Record end so the next poll starts from here.

    Only advances forward: if end is earlier than the current cursor
    (e.g. BQ clock regression), the cursor stays put so the next poll
    doesn't re-query an already-covered range.
    """
    self._last_end = max(self._last_end, end)

  def poll_cursor(self) -> Timestamp:
    """Return the start Timestamp for the next poll."""
    return self._last_end

View on GitHub (pinned to 12126d8942)

Solutions

  1. Wrap the value: set_watermark(Timestamp.from_utc_datetime(dt)) for datetime inputs.
  2. For epoch seconds, use Timestamp(seconds=epoch) before calling set_watermark.
  3. If you already hold a Timestamp, check you are not shadowing the class with a different import.

Example fix

// before
state.set_watermark(datetime.datetime.now(datetime.timezone.utc))
// after
from apache_beam.utils.timestamp import Timestamp
state.set_watermark(Timestamp.from_utc_datetime(datetime.datetime.now(datetime.timezone.utc)))
Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.utils.timestamp import Timestamp
assert isinstance(ts, Timestamp), f"expected Timestamp, got {type(ts).__name__}"

Type guard

def is_beam_timestamp(value) -> bool:
    from apache_beam.utils.timestamp import Timestamp
    return isinstance(value, Timestamp)

Try / catch

try:
    state.set_watermark(ts)
except ValueError as e:
    log.warning("watermark rejected: %s", e)

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/77565346f9571032. Report an issue: GitHub.