Skip to content

[SpannerIO] Add low-latency configuration in Spanner Change Streams #37580

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
stankiewicz wants to merge 3 commits into from
Closed

[SpannerIO] Add low-latency configuration in Spanner Change Streams #37580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b714b84
feat: Make Spanner change stream query parameters configurable with `…
stankiewicz Feb 12, 2026
0d9b313
Make Spanner change stream query parameters configurable with and ; …
stankiewicz Feb 13, 2026
ee3f09d
remove low latency heartbeat setting
stankiewicz Feb 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 37 additions & 2 deletions .../io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,11 @@
import static java.util.stream.Collectors.toList;
import static org.apache.beam.sdk.io.gcp.spanner.MutationUtils.isPointDelete;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_CHANGE_STREAM_NAME;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_HEARTBEAT_MILLIS;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_INCLUSIVE_END_AT;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_INCLUSIVE_START_AT;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_LOW_LATENCY_REAL_TIME_CHECKPOINT_INTERVAL;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_REAL_TIME_CHECKPOINT_INTERVAL;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_RPC_PRIORITY;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_WATERMARK_REFRESH_RATE;
import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.MAX_INCLUSIVE_END_AT;
Expand Down Expand Up @@ -537,6 +540,8 @@ public static ReadChangeStream readChangeStream() {
.setRpcPriority(DEFAULT_RPC_PRIORITY)
.setInclusiveStartAt(DEFAULT_INCLUSIVE_START_AT)
.setInclusiveEndAt(DEFAULT_INCLUSIVE_END_AT)
.setRealTimeCheckpointInterval(DEFAULT_REAL_TIME_CHECKPOINT_INTERVAL)
.setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS)
.build();
}

Expand Down Expand Up @@ -1761,6 +1766,10 @@ public abstract static class ReadChangeStream

abstract @Nullable ValueProvider<Boolean> getPlainText();

abstract Duration getRealTimeCheckpointInterval();

abstract Integer getHeartbeatMillis();

abstract Builder toBuilder();

@AutoValue.Builder
Expand Down Expand Up @@ -1790,6 +1799,16 @@ abstract static class Builder {

abstract Builder setPlainText(ValueProvider<Boolean> plainText);

/**
* When caught up to real-time, checkpoint processing of change stream this often. This sets a
* bound on latency of processing if a steady trickle of elements prevents the heartbeat
* interval from triggering.
*/
abstract Builder setRealTimeCheckpointInterval(Duration realTimeCheckpointInterval);

/** Heartbeat interval for all change stream queries. */
abstract Builder setHeartbeatMillis(Integer heartbeatMillis);

abstract ReadChangeStream build();
}

Expand Down Expand Up @@ -1912,6 +1931,16 @@ public ReadChangeStream withUsingPlainTextChannel(boolean plainText) {
return withUsingPlainTextChannel(ValueProvider.StaticValueProvider.of(plainText));
}

/**
* Configures the change stream to checkpoint and flush output targeting low latency at the cost
* of higher rpc rate and cpu usage.
*/
public ReadChangeStream withLowLatency() {
Copy link

@tianz101 tianz101 Feb 26, 2026

Choose a reason for hiding this comment

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

where is this function gets called?

Copy link
Contributor Author

@stankiewicz stankiewicz Mar 10, 2026

Choose a reason for hiding this comment

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

it would be called by the user:

PCollection<DataChangeRecord> cdc = pipeline.apply(SpannerIO.readChangeStream().withSpannerConfig([..] ).withLowLatency()

Copy link
Contributor Author

@stankiewicz stankiewicz Mar 10, 2026

Choose a reason for hiding this comment

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

so default behaviour is not changed, only users who configure IO with withLowLatency will benefit from it at the cost of more frequent calls to spanner.

return toBuilder()
.setRealTimeCheckpointInterval(DEFAULT_LOW_LATENCY_REAL_TIME_CHECKPOINT_INTERVAL)
Copy link

@tianz101 tianz101 Feb 26, 2026

Choose a reason for hiding this comment

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

I understand you are lowering this to 1s to try to let the query return early so that you can do checkpoint and move forward the start ts for the next query.
But I have seen spanner patition query start delay (time spanner really start the query vs the startTS specified in the TVF query) is about 3s. So this will really come back around 1+3=4s. During this time, if there is any heartbeat event returned, you are able to checkpoint as well. So it seems to me it is more effective to lower the heartbeam return time (100ms for example).

Copy link
Contributor Author

@stankiewicz stankiewicz Mar 10, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

@tianz101 , should I update PR with 100ms heartbeat interval in other PR? Public docs state 1s is minimum. Can we update docs as well so there is no issue from support perspective.

Copy link
Contributor Author

@stankiewicz stankiewicz Mar 10, 2026

Choose a reason for hiding this comment

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

There is also issue with clock skew, see comments above. I feel heartbeat PR #37718 is better.

.build();
}

@Override
public PCollection<DataChangeRecord> expand(PBegin input) {
checkArgument(
Expand Down Expand Up @@ -2018,13 +2047,19 @@ && getInclusiveStartAt().toSqlTimestamp().after(getInclusiveEndAt().toSqlTimesta
MoreObjects.firstNonNull(getWatermarkRefreshRate(), DEFAULT_WATERMARK_REFRESH_RATE);
final CacheFactory cacheFactory = new CacheFactory(daoFactory, watermarkRefreshRate);

final long heartbeatMillis = getHeartbeatMillis().longValue();

final InitializeDoFn initializeDoFn =
new InitializeDoFn(daoFactory, mapperFactory, startTimestamp, endTimestamp);
new InitializeDoFn(
daoFactory, mapperFactory, startTimestamp, endTimestamp, heartbeatMillis);
final DetectNewPartitionsDoFn detectNewPartitionsDoFn =
new DetectNewPartitionsDoFn(
daoFactory, mapperFactory, actionFactory, cacheFactory, metrics);
final Duration realTimeCheckpointInterval = getRealTimeCheckpointInterval();

final ReadChangeStreamPartitionDoFn readChangeStreamPartitionDoFn =
new ReadChangeStreamPartitionDoFn(daoFactory, mapperFactory, actionFactory, metrics);
new ReadChangeStreamPartitionDoFn(
daoFactory, mapperFactory, actionFactory, metrics, realTimeCheckpointInterval);
final PostProcessingMetricsDoFn postProcessingMetricsDoFn =
new PostProcessingMetricsDoFn(metrics);

Expand Down
7 changes: 7 additions & 0 deletions ...rc/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/ChangeStreamsConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,13 @@ public class ChangeStreamsConstants {
*/
public static final Timestamp DEFAULT_INCLUSIVE_END_AT = MAX_INCLUSIVE_END_AT;

public static final Duration DEFAULT_REAL_TIME_CHECKPOINT_INTERVAL = Duration.standardMinutes(2);

public static final int DEFAULT_HEARTBEAT_MILLIS = 2000;

public static final Duration DEFAULT_LOW_LATENCY_REAL_TIME_CHECKPOINT_INTERVAL =
Copy link

@tianz101 tianz101 Feb 26, 2026

Choose a reason for hiding this comment

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

nit: format to 1 or 2 lines.

Duration.standardSeconds(1);

/** The default priority for a change stream query is {@link RpcPriority#HIGH}. */
public static final RpcPriority DEFAULT_RPC_PRIORITY = RpcPriority.HIGH;

Expand Down
7 changes: 5 additions & 2 deletions .../src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ActionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@ public synchronized PartitionEventRecordAction partitionEventRecordAction(
* @param partitionEventRecordAction action class to process {@link
* org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEventRecord}s
* @param metrics metrics gathering class
* @param realTimeCheckpointInterval the duration added to current time for the end timestamp
* @return single instance of the {@link QueryChangeStreamAction}
*/
public synchronized QueryChangeStreamAction queryChangeStreamAction(
Expand All @@ -188,7 +189,8 @@ public synchronized QueryChangeStreamAction queryChangeStreamAction(
PartitionEndRecordAction partitionEndRecordAction,
PartitionEventRecordAction partitionEventRecordAction,
ChangeStreamMetrics metrics,
boolean isMutableChangeStream) {
boolean isMutableChangeStream,
Duration realTimeCheckpointInterval) {
if (queryChangeStreamActionInstance == null) {
queryChangeStreamActionInstance =
new QueryChangeStreamAction(
Expand All @@ -203,7 +205,8 @@ public synchronized QueryChangeStreamAction queryChangeStreamAction(
partitionEndRecordAction,
partitionEventRecordAction,
metrics,
isMutableChangeStream);
isMutableChangeStream,
realTimeCheckpointInterval);
}
return queryChangeStreamActionInstance;
}
Expand Down
16 changes: 10 additions & 6 deletions ...java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@ public class QueryChangeStreamAction {
private final PartitionEventRecordAction partitionEventRecordAction;
private final ChangeStreamMetrics metrics;
private final boolean isMutableChangeStream;
private final Duration realTimeCheckpointInterval;

/**
* Constructs an action class for performing a change stream query for a given partition.
Expand All @@ -109,6 +110,7 @@ public class QueryChangeStreamAction {
* @param PartitionEventRecordAction action class to process {@link PartitionEventRecord}s
* @param metrics metrics gathering class
* @param isMutableChangeStream whether the change stream is mutable or not
* @param realTimeCheckpointInterval duration to add to current time
*/
QueryChangeStreamAction(
ChangeStreamDao changeStreamDao,
Expand All @@ -122,7 +124,8 @@ public class QueryChangeStreamAction {
PartitionEndRecordAction partitionEndRecordAction,
PartitionEventRecordAction partitionEventRecordAction,
ChangeStreamMetrics metrics,
boolean isMutableChangeStream) {
boolean isMutableChangeStream,
Duration realTimeCheckpointInterval) {
this.changeStreamDao = changeStreamDao;
this.partitionMetadataDao = partitionMetadataDao;
this.changeStreamRecordMapper = changeStreamRecordMapper;
Expand All @@ -135,6 +138,7 @@ public class QueryChangeStreamAction {
this.partitionEventRecordAction = partitionEventRecordAction;
this.metrics = metrics;
this.isMutableChangeStream = isMutableChangeStream;
this.realTimeCheckpointInterval = realTimeCheckpointInterval;
}

/**
Expand Down Expand Up @@ -387,12 +391,12 @@ private boolean isTimestampOutOfRange(SpannerException e) {
&& e.getMessage().contains(OUT_OF_RANGE_ERROR_MESSAGE);
}

// Return (now + 2 mins) as the end timestamp for reading change streams. This is only used if
// users want to run the connector forever. If the end timestamp is reached, we will resume
// processing from that timestamp on a subsequent DoFn execution.
// Return (now + config duration) as the end timestamp for reading change streams. This is only
// used if users want to run the connector forever. If the end timestamp is reached, we
// will resume processing from that timestamp on a subsequent DoFn execution.
private Timestamp getNextReadChangeStreamEndTimestamp() {
final Timestamp current = Timestamp.now();
return Timestamp.ofTimeSecondsAndNanos(current.getSeconds() + 2 * 60, current.getNanos());
return Timestamp.ofTimeMicroseconds(
Instant.now().plus(realTimeCheckpointInterval).getMillis() * 1000L);
Copy link
Contributor Author

@stankiewicz stankiewicz Feb 27, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

If end_time is calculated as now() + 2m it's very unlikely that clock skew will impact pulling data.
if end_time is now() + 1s and machine clock is skewed, e.g late few secs, we may not pull fresh data, even if it's available in spanner.

Copy link
Contributor

@scwhittle scwhittle Feb 27, 2026

Choose a reason for hiding this comment

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

I wonder if we could detect that by capturing the now() we used before starting the query and then comparing now() at the end of the query versus that start time.

If the duration is a lot less than our desired realtimecheckpointinterval, it means that our local clock is slow compared to the spanner true time. We could keep track of that and try to adjust our clock values for future calls.

Copy link
Contributor Author

@stankiewicz stankiewicz Feb 27, 2026

Choose a reason for hiding this comment

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

you mean something like:


acceptable_lag = 500ms
DRIFT = 0ms
start = partition.start e.g. 12:00:00
end = now() + 1s - DRIFT - e.g. 12:00:01 
loop_query()
true_end = now() - e.g. 12:00.00:100

if(end - true_end  < acceptable_lag) { // diff is 900ms 
// GCE clock is late by ~900ms as query finished (end - true_end) earlier than it should. we risk finishing query early, in worst case not pulling fresh data.

DRIFT = end - true_end // 900ms

} else {
 // opposite, `true_end` happened later than we predicted set with `end`, which may mean that GCE clock is early but it may also mean that we can have expensive processing downstream
}

I think I really prefer heartbeat PR as we don't have to include such heuristic :)

}

// For Mutable Change Stream bounded queries, update the query end timestamp to be within 2
Expand Down
12 changes: 5 additions & 7 deletions ...m/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/InitializeDoFn.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,11 +36,7 @@ public class InitializeDoFn extends DoFn<byte[], PartitionMetadata> implements S

private static final long serialVersionUID = -8921188388649003102L;

/** Heartbeat interval for all change stream queries will be of 2 seconds. */
// Be careful when changing this interval, as it needs to be less than the checkpointing interval
// in Dataflow. Otherwise, if there are no records within checkpoint intervals, the consuming of
// a change stream query might get stuck.
private static final long DEFAULT_HEARTBEAT_MILLIS = 2000;
private final long heartbeatMillis;

private final DaoFactory daoFactory;
private final MapperFactory mapperFactory;
Expand All @@ -53,11 +49,13 @@ public InitializeDoFn(
DaoFactory daoFactory,
MapperFactory mapperFactory,
com.google.cloud.Timestamp startTimestamp,
com.google.cloud.Timestamp endTimestamp) {
com.google.cloud.Timestamp endTimestamp,
long heartbeatMillis) {
this.daoFactory = daoFactory;
this.mapperFactory = mapperFactory;
this.startTimestamp = startTimestamp;
this.endTimestamp = endTimestamp;
this.heartbeatMillis = heartbeatMillis;
}

@ProcessElement
Expand Down Expand Up @@ -88,7 +86,7 @@ private void createFakeParentPartition() {
.setPartitionToken(InitialPartition.PARTITION_TOKEN)
.setStartTimestamp(startTimestamp)
.setEndTimestamp(endTimestamp)
.setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS)
.setHeartbeatMillis(heartbeatMillis)
.setState(State.CREATED)
.setWatermark(startTimestamp)
.build();
Expand Down
12 changes: 9 additions & 3 deletions .../org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,8 @@ public class ReadChangeStreamPartitionDoFn extends DoFn<PartitionMetadata, DataC
*/
private ThroughputEstimator<DataChangeRecord> throughputEstimator;

private final Duration realTimeCheckpointInterval;

private transient QueryChangeStreamAction queryChangeStreamAction;

/**
Expand All @@ -95,17 +97,20 @@ public class ReadChangeStreamPartitionDoFn extends DoFn<PartitionMetadata, DataC
* @param mapperFactory the {@link MapperFactory} to construct {@link ChangeStreamRecordMapper}s
* @param actionFactory the {@link ActionFactory} to construct actions
* @param metrics the {@link ChangeStreamMetrics} to emit partition related metrics
* @param realTimeCheckpointInterval duration to be used for the next end timestamp
*/
public ReadChangeStreamPartitionDoFn(
DaoFactory daoFactory,
MapperFactory mapperFactory,
ActionFactory actionFactory,
ChangeStreamMetrics metrics) {
ChangeStreamMetrics metrics,
Duration realTimeCheckpointInterval) {
this.daoFactory = daoFactory;
this.mapperFactory = mapperFactory;
this.actionFactory = actionFactory;
this.mapperFactory = mapperFactory;
this.metrics = metrics;
this.isMutableChangeStream = daoFactory.isMutableChangeStream();
this.realTimeCheckpointInterval = realTimeCheckpointInterval;
this.throughputEstimator = new NullThroughputEstimator<>();
}

Expand Down Expand Up @@ -218,7 +223,8 @@ public void setup() {
partitionEndRecordAction,
partitionEventRecordAction,
metrics,
isMutableChangeStream);
isMutableChangeStream,
realTimeCheckpointInterval);
}

/**
Expand Down
10 changes: 7 additions & 3 deletions .../org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamActionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
import org.apache.beam.sdk.transforms.splittabledofn.ManualWatermarkEstimator;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Test;
Expand Down Expand Up @@ -119,7 +120,8 @@ public void setUp() throws Exception {
partitionEndRecordAction,
partitionEventRecordAction,
metrics,
false);
false,
Duration.standardMinutes(2));
final Struct row = mock(Struct.class);
partition =
PartitionMetadata.newBuilder()
Expand Down Expand Up @@ -935,7 +937,8 @@ public void testQueryChangeStreamWithMutableChangeStreamCappedEndTimestamp() {
partitionEndRecordAction,
partitionEventRecordAction,
metrics,
true);
true,
Duration.standardMinutes(2));

// Set endTimestamp to 60 minutes in the future
Timestamp now = Timestamp.now();
Expand Down Expand Up @@ -983,7 +986,8 @@ public void testQueryChangeStreamWithMutableChangeStreamUncappedEndTimestamp() {
partitionEndRecordAction,
partitionEventRecordAction,
metrics,
true);
true,
Duration.standardMinutes(2));

// Set endTimestamp to only 10 seconds in the future
Timestamp now = Timestamp.now();
Expand Down
3 changes: 2 additions & 1 deletion ...c/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/InitializeDoFnTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,8 @@ public void setUp() {
daoFactory,
mapperFactory,
Timestamp.ofTimeMicroseconds(1L),
Timestamp.ofTimeMicroseconds(2L));
Timestamp.ofTimeMicroseconds(2L),
2000L);
}

@Test
Expand Down
8 changes: 6 additions & 2 deletions .../apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
import org.apache.beam.sdk.transforms.splittabledofn.ManualWatermarkEstimator;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Test;
Expand Down Expand Up @@ -103,7 +104,9 @@ public void setUp() {
partitionEventRecordAction = mock(PartitionEventRecordAction.class);
queryChangeStreamAction = mock(QueryChangeStreamAction.class);

doFn = new ReadChangeStreamPartitionDoFn(daoFactory, mapperFactory, actionFactory, metrics);
doFn =
new ReadChangeStreamPartitionDoFn(
daoFactory, mapperFactory, actionFactory, metrics, Duration.standardMinutes(2));
doFn.setThroughputEstimator(throughputEstimator);

partition =
Expand Down Expand Up @@ -152,7 +155,8 @@ public void setUp() {
eq(partitionEndRecordAction),
eq(partitionEventRecordAction),
eq(metrics),
anyBoolean()))
anyBoolean(),
eq(Duration.standardMinutes(2))))
.thenReturn(queryChangeStreamAction);

doFn.setup();
Expand Down
Loading