Add logic to handle end timestamp of mutable change stream bounded query by chenxuesdu · Pull Request #37459 · apache/beam

chenxuesdu · 2026-01-30T19:47:43Z

For mutable change stream bounded queries, Spanner only allow the max end timestamp to be 30 minutes in the future. So when user provided a end timestamp to read, we choose the min of max allow timestamp and the provided timestamp.

gemini-code-assist · 2026-01-30T19:48:03Z

Summary of Changes

Hello @chenxuesdu, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request addresses a specific constraint in Google Cloud Spanner regarding mutable change stream bounded queries, where the end timestamp cannot exceed 30 minutes into the future. The changes introduce a mechanism to detect if a change stream is mutable and, if so, automatically adjust the provided end timestamp to comply with this limitation. This ensures that queries remain valid and prevents potential errors when users specify end timestamps beyond the allowed future window.

Highlights

Mutable Change Stream Identification: A new isMutableChangeStream flag has been introduced and propagated through relevant classes (DaoFactory, ReadChangeStreamPartitionDoFn, QueryChangeStreamAction) to specifically identify and handle mutable change streams.
Bounded Query End Timestamp Enforcement: Logic has been added to QueryChangeStreamAction to enforce Spanner's limitation on mutable change stream bounded queries. The end timestamp for such queries is now capped to a maximum of 30 minutes in the future from the current time, choosing the minimum between the user-provided timestamp and this limit.
Code Structure and Dependency Injection: The isMutableChangeStream flag is passed as a constructor argument to QueryChangeStreamAction and initialized within ReadChangeStreamPartitionDoFn from DaoFactory, ensuring proper dependency injection and enabling conditional logic based on the change stream type.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

github-actions · 2026-01-30T20:10:53Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

jiangzzhu · 2026-02-03T17:51:53Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

+    if (this.isMutableChangeStream) {
+      final Timestamp current = Timestamp.now();
+      Timestamp maxTimestamp =
+          Timestamp.ofTimeSecondsAndNanos(current.getSeconds() + 30 * 60, current.getNanos());


Should we leave some margin for the end timestamp, e.g. 25min or 20min in the future, in case that the query is processed too fast and this future timestamp still too far in the future?

Updated to 2 minutes in the future. Please take a look.

jiangzzhu · 2026-02-03T17:55:53Z

/gemini review

gemini-code-assist

Code Review

This pull request correctly implements the logic to handle the end timestamp for mutable change stream bounded queries in Spanner, which limits the end timestamp to a maximum of 30 minutes in the future. The changes are well-implemented, and the new logic is propagated correctly through the different layers of the application. The added tests effectively verify the new behavior. My review includes a few suggestions to improve code readability and maintainability by replacing magic numbers with named constants.

gemini-code-assist · 2026-02-03T17:59:32Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

+  private Timestamp getBoundedQueryEndTimestamp(Timestamp endTimestamp) {
+    if (this.isMutableChangeStream) {
+      final Timestamp current = Timestamp.now();
+      Timestamp maxTimestamp =
+          Timestamp.ofTimeSecondsAndNanos(current.getSeconds() + 30 * 60, current.getNanos());
+      return maxTimestamp.compareTo(endTimestamp) < 0 ? maxTimestamp : endTimestamp;
+    }
+    return endTimestamp;
+  }


For better readability and maintainability, it's good practice to extract magic numbers into constants. Also, using Comparators.min can make the intent of the code clearer.

You would need to add the following import:
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Comparators;

private Timestamp getBoundedQueryEndTimestamp(Timestamp endTimestamp) { if (this.isMutableChangeStream) { final long maxSecondsInFuture = 30 * 60; final Timestamp current = Timestamp.now(); final Timestamp maxTimestamp = Timestamp.ofTimeSecondsAndNanos(current.getSeconds() + maxSecondsInFuture, current.getNanos()); return Comparators.min(maxTimestamp, endTimestamp); } return endTimestamp; }

Maybe define MAX_FUTURE_END_TIME_SCEONDS = 30 * 60 into constant value of the class, in case we need to tune it in the future.

Updated to 2 minutes in the future. Please take a look.

gemini-code-assist · 2026-02-03T17:59:32Z

...ava/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamActionTest.java

+  public void testQueryChangeStreamWithMutableChangeStreamCappedEndTimestamp() {
+    // Initialize action with isMutableChangeStream = true
+    action =
+        new QueryChangeStreamAction(
+            changeStreamDao,
+            partitionMetadataDao,
+            changeStreamRecordMapper,
+            partitionMetadataMapper,
+            dataChangeRecordAction,
+            heartbeatRecordAction,
+            childPartitionsRecordAction,
+            partitionStartRecordAction,
+            partitionEndRecordAction,
+            partitionEventRecordAction,
+            metrics,
+            true);
+
+    // Set endTimestamp to 60 minutes in the future
+    Timestamp now = Timestamp.now();
+    Timestamp endTimestamp =
+        Timestamp.ofTimeSecondsAndNanos(now.getSeconds() + 60 * 60, now.getNanos());
+
+    partition = partition.toBuilder().setEndTimestamp(endTimestamp).build();
+    when(restriction.getTo()).thenReturn(endTimestamp);
+    when(partitionMetadataMapper.from(any())).thenReturn(partition);
+
+    final ChangeStreamResultSet resultSet = mock(ChangeStreamResultSet.class);
+    final ArgumentCaptor<Timestamp> timestampCaptor = ArgumentCaptor.forClass(Timestamp.class);
+    when(changeStreamDao.changeStreamQuery(
+            eq(PARTITION_TOKEN), eq(PARTITION_START_TIMESTAMP),
+            timestampCaptor.capture(), eq(PARTITION_HEARTBEAT_MILLIS)))
+        .thenReturn(resultSet);
+    when(resultSet.next()).thenReturn(false); // Query finishes (reaches cap)
+    when(watermarkEstimator.currentWatermark()).thenReturn(WATERMARK);
+    when(restrictionTracker.tryClaim(any(Timestamp.class))).thenReturn(true);
+
+    final ProcessContinuation result =
+        action.run(
+            partition, restrictionTracker, outputReceiver, watermarkEstimator, bundleFinalizer);
+
+    // Verify query was capped at ~30 minutes
+    long diff = timestampCaptor.getValue().getSeconds() - now.getSeconds();
+    assertTrue("Query should be capped at approx 30 minutes (1800s)", Math.abs(diff - 1800) < 10);
+
+    // Crucial: Should RESUME to process the rest later
+    assertEquals(ProcessContinuation.resume(), result);
+  }


Using named constants for magic numbers like 60 * 60 and 1800 improves code readability and maintainability.

public void testQueryChangeStreamWithMutableChangeStreamCappedEndTimestamp() { final long sixtyMinutesInSeconds = 60 * 60; final long thirtyMinutesInSeconds = 30 * 60; // Initialize action with isMutableChangeStream = true action = new QueryChangeStreamAction( changeStreamDao, partitionMetadataDao, changeStreamRecordMapper, partitionMetadataMapper, dataChangeRecordAction, heartbeatRecordAction, childPartitionsRecordAction, partitionStartRecordAction, partitionEndRecordAction, partitionEventRecordAction, metrics, true); // Set endTimestamp to 60 minutes in the future Timestamp now = Timestamp.now(); Timestamp endTimestamp = Timestamp.ofTimeSecondsAndNanos(now.getSeconds() + sixtyMinutesInSeconds, now.getNanos()); partition = partition.toBuilder().setEndTimestamp(endTimestamp).build(); when(restriction.getTo()).thenReturn(endTimestamp); when(partitionMetadataMapper.from(any())).thenReturn(partition); final ChangeStreamResultSet resultSet = mock(ChangeStreamResultSet.class); final ArgumentCaptor<Timestamp> timestampCaptor = ArgumentCaptor.forClass(Timestamp.class); when(changeStreamDao.changeStreamQuery( eq(PARTITION_TOKEN), eq(PARTITION_START_TIMESTAMP), timestampCaptor.capture(), eq(PARTITION_HEARTBEAT_MILLIS))) .thenReturn(resultSet); when(resultSet.next()).thenReturn(false); // Query finishes (reaches cap) when(watermarkEstimator.currentWatermark()).thenReturn(WATERMARK); when(restrictionTracker.tryClaim(any(Timestamp.class))).thenReturn(true); final ProcessContinuation result = action.run( partition, restrictionTracker, outputReceiver, watermarkEstimator, bundleFinalizer); // Verify query was capped at ~30 minutes long diff = timestampCaptor.getValue().getSeconds() - now.getSeconds(); assertTrue( "Query should be capped at approx 30 minutes (1800s)", Math.abs(diff - thirtyMinutesInSeconds) < 10); // Crucial: Should RESUME to process the rest later assertEquals(ProcessContinuation.resume(), result); }

gemini-code-assist · 2026-02-03T17:59:32Z

...ava/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamActionTest.java

+  public void testQueryChangeStreamWithMutableChangeStreamUncappedEndTimestamp() {
+    action =
+        new QueryChangeStreamAction(
+            changeStreamDao,
+            partitionMetadataDao,
+            changeStreamRecordMapper,
+            partitionMetadataMapper,
+            dataChangeRecordAction,
+            heartbeatRecordAction,
+            childPartitionsRecordAction,
+            partitionStartRecordAction,
+            partitionEndRecordAction,
+            partitionEventRecordAction,
+            metrics,
+            true);
+
+    // Set endTimestamp to only 10 minutes in the future
+    Timestamp now = Timestamp.now();
+    Timestamp endTimestamp =
+        Timestamp.ofTimeSecondsAndNanos(now.getSeconds() + 10 * 60, now.getNanos());
+
+    partition = partition.toBuilder().setEndTimestamp(endTimestamp).build();
+    when(restriction.getTo()).thenReturn(endTimestamp);
+    when(partitionMetadataMapper.from(any())).thenReturn(partition);
+
+    final ChangeStreamResultSet resultSet = mock(ChangeStreamResultSet.class);
+    when(changeStreamDao.changeStreamQuery(
+            eq(PARTITION_TOKEN), eq(PARTITION_START_TIMESTAMP),
+            eq(endTimestamp), eq(PARTITION_HEARTBEAT_MILLIS)))
+        .thenReturn(resultSet);
+    when(resultSet.next()).thenReturn(false);
+    when(watermarkEstimator.currentWatermark()).thenReturn(WATERMARK);
+    when(restrictionTracker.tryClaim(endTimestamp)).thenReturn(true);
+
+    final ProcessContinuation result =
+        action.run(
+            partition, restrictionTracker, outputReceiver, watermarkEstimator, bundleFinalizer);
+
+    // Should STOP because we reached the actual requested endTimestamp
+    assertEquals(ProcessContinuation.stop(), result);
+  }


Using a named constant for the magic number 10 * 60 would improve readability.

public void testQueryChangeStreamWithMutableChangeStreamUncappedEndTimestamp() { final long tenMinutesInSeconds = 10 * 60; action = new QueryChangeStreamAction( changeStreamDao, partitionMetadataDao, changeStreamRecordMapper, partitionMetadataMapper, dataChangeRecordAction, heartbeatRecordAction, childPartitionsRecordAction, partitionStartRecordAction, partitionEndRecordAction, partitionEventRecordAction, metrics, true); // Set endTimestamp to only 10 minutes in the future Timestamp now = Timestamp.now(); Timestamp endTimestamp = Timestamp.ofTimeSecondsAndNanos(now.getSeconds() + tenMinutesInSeconds, now.getNanos()); partition = partition.toBuilder().setEndTimestamp(endTimestamp).build(); when(restriction.getTo()).thenReturn(endTimestamp); when(partitionMetadataMapper.from(any())).thenReturn(partition); final ChangeStreamResultSet resultSet = mock(ChangeStreamResultSet.class); when(changeStreamDao.changeStreamQuery( eq(PARTITION_TOKEN), eq(PARTITION_START_TIMESTAMP), eq(endTimestamp), eq(PARTITION_HEARTBEAT_MILLIS))) .thenReturn(resultSet); when(resultSet.next()).thenReturn(false); when(watermarkEstimator.currentWatermark()).thenReturn(WATERMARK); when(restrictionTracker.tryClaim(endTimestamp)).thenReturn(true); final ProcessContinuation result = action.run( partition, restrictionTracker, outputReceiver, watermarkEstimator, bundleFinalizer); // Should STOP because we reached the actual requested endTimestamp assertEquals(ProcessContinuation.stop(), result); }

tianz101 · 2026-02-03T18:39:57Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

    // exceptions about being out of timestamp range).
-    boolean stopAfterQuerySucceeds = isBoundedRestriction;
+    boolean stopAfterQuerySucceeds =
+        isBoundedRestriction && changeStreamQueryEndTimestamp.equals(endTimestamp);


For v1, this changeStreamQueryEndTimestamp.equals(endTimestamp) is valid.
For v2, this changeStreamQueryEndTimestamp.equals(endTimestamp) may never happen. Is that intended? Let's walk through v2 scenario a little bit more.

For v2 bounded query, the changeStreamQueryEndTimestamp.equals(endTimestamp) condition can meet, since we choose the min of the changeStreamQueryEndTimestamp and endTimestamp, so they will equal to each other sometime. Started one connector test it, it stopped after the endTimestamp given by user.

tianz101 · 2026-02-03T18:41:19Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

+  private Timestamp getBoundedQueryEndTimestamp(Timestamp endTimestamp) {
+    if (this.isMutableChangeStream) {
+      final Timestamp current = Timestamp.now();
+      Timestamp maxTimestamp =


For v1 we use endTs,
For v2 we should use getNextReadChangeStreamEndTimestamp() here so that for V2, bounded endTs or unbounded endTs have similar query patterns.

Updated to 2 minutes in the future. Please take a look.

…ery.

chenxuesdu

updated. Please take another look.

jiangzzhu · 2026-02-04T23:39:19Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

+            ? getBoundedQueryEndTimestamp(endTimestamp)
+            : getNextReadChangeStreamEndTimestamp();

    // Once the changeStreamQuery completes we may need to resume reading from the partition if we


Should we update the comment here?

tianz101 · 2026-02-05T18:49:57Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

+  // the future.
+  private Timestamp getBoundedQueryEndTimestamp(Timestamp endTimestamp) {
+    if (this.isMutableChangeStream) {
+      Timestamp maxTimestamp = getNextReadChangeStreamEndTimestamp();


Should we call it nextTimestamp instead of maxTimestamp?

Also in general, for code we modified, can you double check the comments are updated as well. Thanks.

Updated. Please take a look.

tianz101 · 2026-02-05T18:55:08Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

    // we didn't  encounter any indications that the partition is done (explicit end records or
    // exceptions about being out of timestamp range).
-    boolean stopAfterQuerySucceeds = isBoundedRestriction;
+    boolean stopAfterQuerySucceeds = false;


This line seems not needed.

this is the definition for the boolean.

Abacn · 2026-02-06T01:57:37Z

...org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java

            partitionEventRecordAction,
-            metrics))
+            metrics,
+            anyBoolean()))


testQueryChangeStreamMode (org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFnTest) failed

sdks/java/io/google-cloud-platform/build/test-results/test/TEST-org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFnTest.xml [took 0s]

org.mockito.exceptions.misusing.InvalidUseOfMatchersException: Invalid use of argument matchers! 12 matchers expected, 1 recorded: -> at org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFnTest.setUp(ReadChangeStreamPartitionDoFnTest.java:154) This exception may occur if matchers are combined with raw values: //incorrect: someMethod(any(), "raw String"); When using matchers, all arguments have to be provided by matchers. For example: //correct: someMethod(any(), eq("String by matcher")); For more info see javadoc for Matchers class. at app//org.apache.beam.sdk.io.gcp.spanner.changestreams.action.ActionFactory.queryChangeStreamAction(ActionFactory.java:192) at app//org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFnTest.setUp(ReadChangeStreamPartitionDoFnTest.java:142) at java.base@11.0.24/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

looks relevant test failure

Abacn

Besides that, there is also a Spanner ChangeStream Integration failure:

testOrderedWithinKey (org.apache.beam.sdk.io.gcp.spanner.changestreams.it.SpannerChangeStreamOrderedWithinKeyIT) failed

java.lang.AssertionError: ParDo(ToString)/ParMultiDo(ToString).output: 
Expected: iterable with items ["{\"SingerId\":\"0\"}\n{\"FirstName\":\"Inserting mutation 0\",\"LastName\":null,\"SingerInfo\":null};{};", "{\"SingerId\":\"1\"}\n{\"FirstName\":\"Inserting mutation 1\",\"LastName\":null,\"SingerInfo\":null};{\"FirstName\":\"Updating mutation 1\"};{};{\"FirstName\":\"Inserting mutation 1\",\"LastName\":null,\"SingerInfo\":null};{};", "{\"SingerId\":\"2\"}\n{\"FirstName\":\"Inserting mutation 2\",\"LastName\":null,\"SingerInfo\":null};{\"FirstName\":\"Updating mutation 2\"};{};", "{\"SingerId\":\"3\"}\n{\"FirstName\":\"Inserting mutation 3\",\"LastName\":null,\"SingerInfo\":null};{\"FirstName\":\"Updating mutation 3\"};{};", "{\"SingerId\":\"4\"}\n{\"FirstName\":\"Inserting mutation 4\",\"LastName\":null,\"SingerInfo\":null};{};", "{\"SingerId\":\"5\"}\n{\"FirstName\":\"Updating mutation 5\",\"LastName\":null,\"SingerInfo\":null};{\"FirstName\":\"Updating mutation 5\"};{};"] in any order
     but: not matched: "{\"SingerId\":\"0\"}\n{\"FirstName\":\"Inserting mutation 0\",\"LastName\":null,\"SingerInfo\":null};"
	at org.apache.beam.sdk.testing.PAssert$PAssertionSite.capture(PAssert.java:176)
	at org.apache.beam.sdk.testing.PAssert.that(PAssert.java:461)
	at org.apache.beam.sdk.testing.PAssert.that(PAssert.java:453)

github-actions bot added java io gcp spanner labels Jan 30, 2026

chenxuesdu force-pushed the boundedV2Query branch 4 times, most recently from 76aa3cd to 56ab9e1 Compare January 30, 2026 21:48

jiangzzhu approved these changes Feb 3, 2026

View reviewed changes

gemini-code-assist bot reviewed Feb 3, 2026

View reviewed changes

tianz101 reviewed Feb 3, 2026

View reviewed changes

Add logic to handle end timestamp of mutable change stream bounded qu…

78b731f

…ery.

chenxuesdu force-pushed the boundedV2Query branch from 56ab9e1 to 78b731f Compare February 4, 2026 23:27

chenxuesdu commented Feb 4, 2026

View reviewed changes

jiangzzhu reviewed Feb 4, 2026

View reviewed changes

jiangzzhu approved these changes Feb 4, 2026

View reviewed changes

tianz101 reviewed Feb 5, 2026

View reviewed changes

chenxuesdu force-pushed the boundedV2Query branch 2 times, most recently from 9412858 to 94b3d4f Compare February 5, 2026 20:03

Resolve comments

5ebada3

chenxuesdu force-pushed the boundedV2Query branch from 94b3d4f to 5ebada3 Compare February 5, 2026 21:42

tianz101 approved these changes Feb 5, 2026

View reviewed changes

Abacn reviewed Feb 6, 2026

View reviewed changes

Conversation

chenxuesdu commented Jan 30, 2026

Uh oh!

gemini-code-assist bot commented Jan 30, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

github-actions bot commented Jan 30, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jiangzzhu commented Feb 3, 2026

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist bot Feb 3, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Feb 3, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Feb 3, 2026

Choose a reason for hiding this comment

Uh oh!

tianz101 Feb 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chenxuesdu left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Abacn left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

tianz101 Feb 3, 2026 •

edited

Loading