Skip to content

[SPARK-58309][SS] Anchor now() and current_time() to the batch timestamp in streaming queries#57480

Open
PorridgeSwim wants to merge 1 commit into
apache:masterfrom
PorridgeSwim:spark-streaming-now-determinism
Open

[SPARK-58309][SS] Anchor now() and current_time() to the batch timestamp in streaming queries#57480
PorridgeSwim wants to merge 1 commit into
apache:masterfrom
PorridgeSwim:spark-streaming-now-determinism

Conversation

@PorridgeSwim

@PorridgeSwim PorridgeSwim commented Jul 23, 2026

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

In MicroBatchExecution, the per-batch rewrite that replaces CURRENT_LIKE temporal expressions with CurrentBatchTimestamp only matched CurrentTimestamp, CurrentDate and LocalTimestamp. Two expressions carrying the CURRENT_LIKE tree pattern were missed:

  • Now (now()) -- a sibling of CurrentTimestamp under the CurrentTimestampLike base class. The match on the leaf type CurrentTimestamp did not catch it.
  • CurrentTime (current_time() / localtime(), the TIME value type).

This PR:

  • Widens the CurrentTimestamp case to CurrentTimestampLike, covering both current_timestamp() and now().
  • Adds a CurrentTime case, and a TimeType branch to CurrentBatchTimestamp.toLiteral that derives the time-of-day (nanoseconds since midnight, in the session time zone) from the batch timestamp, truncated to the type's precision. (Because the batch timestamp is millisecond-resolution, sub-millisecond digits of current_time() are zero in streaming.)

Why are the changes needed?

Expressions left unrewritten survive to the per-batch optimizer, where ComputeCurrentTime folds them using the wall clock at plan-optimization time. That value is not reproducible when a batch is replayed after failure or on restart from a checkpoint, breaking the determinism guarantee that CurrentBatchTimestamp (anchored to the batch timestamp persisted in the offset log) exists to provide. now() and current_time() were therefore non-deterministic across batch replays.

Does this PR introduce any user-facing change?

Yes. In a streaming query, now() and current_time() now return the batch timestamp (stable across replay/restart of the same batch), consistent with current_timestamp(), current_date() and localtimestamp(), instead of the wall clock at plan-optimization time.

How was this patch tested?

Added a parameterized test in StreamSuite covering current_timestamp(), now(), current_date() and current_time(). Using a manual clock it asserts the exact per-batch value, then purges the last batch's commit and restarts with the wall clock advanced far ahead, asserting the replayed batch reproduces its persisted timestamp rather than the advanced clock.

Was this patch authored or co-authored using generative AI tooling?

No

…amp in streaming queries

### What changes were proposed in this pull request?

In `MicroBatchExecution`, the per-batch rewrite that replaces `CURRENT_LIKE`
temporal expressions with `CurrentBatchTimestamp` only matched `CurrentTimestamp`,
`CurrentDate` and `LocalTimestamp`. Two expressions carrying the `CURRENT_LIKE`
tree pattern were missed:

- `Now` (`now()`) -- a sibling of `CurrentTimestamp` under the `CurrentTimestampLike`
  base class. The match on the leaf type `CurrentTimestamp` did not catch it.
- `CurrentTime` (`current_time()` / `localtime()`, a `TIME` value added in 4.x).

This PR:
- Widens the `CurrentTimestamp` case to `CurrentTimestampLike`, covering both
  `current_timestamp()` and `now()`.
- Adds a `CurrentTime` case, and a `TimeType` branch to
  `CurrentBatchTimestamp.toLiteral` that derives the time-of-day (nanoseconds since
  midnight, in the session time zone) from the batch timestamp, truncated to the
  type's precision.

### Why are the changes needed?

Expressions left unrewritten survive to the per-batch optimizer, where
`ComputeCurrentTime` folds them using the wall clock at plan-optimization time.
That value is not reproducible when a batch is replayed after failure or on
restart from a checkpoint, breaking the determinism guarantee that
`CurrentBatchTimestamp` (anchored to the batch timestamp persisted in the offset
log) exists to provide. `now()` and `current_time()` were therefore
non-deterministic across batch replays.

### Does this PR introduce _any_ user-facing change?

Yes. In a streaming query, `now()` and `current_time()` now return the batch
timestamp (stable across replay/restart of the same batch), consistent with
`current_timestamp()`, `current_date()` and `localtimestamp()`, instead of the
wall clock at plan-optimization time.

### How was this patch tested?

Added a parameterized test in `StreamSuite` covering `current_timestamp()`,
`now()`, `current_date()` and `current_time()`. Using a manual clock it asserts
the exact per-batch value, then purges the last batch's commit and restarts with
the wall clock advanced far ahead, asserting the replayed batch reproduces its
persisted timestamp rather than the advanced clock.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

Co-authored-by: Isaac
case _: TimestampNTZType =>
Literal(convertTz(timestampUs, ZoneOffset.UTC, zoneId), TimestampNTZType)
case _: DateType => Literal(microsToDays(timestampUs, zoneId), DateType)
case t: TimeType =>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please guard this behind a flag. also we should think through if this is a breaking change or behavior change.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants