[Spanner Change Streams] Parse proto bytes change stream record in ChangeStreamRecordMapper #37427

chenxuesdu · 2026-01-27T00:33:02Z

When parsing the change stream record in postgresql, if the output is proto bytes, map it to proto type, so we can reuse the existing logic to parse the change stream record.

gemini-code-assist · 2026-01-27T00:33:19Z

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 enhances the Spanner Change Streams connector to support parsing change stream records that are delivered as proto bytes, particularly in PostgreSQL environments. By introducing specific methods to identify and deserialize these proto bytes, the system can now uniformly process change stream data, whether it originates as JSONB or proto, leveraging existing proto-based mapping logic. This change improves compatibility and robustness for different Spanner database dialects and output formats.

Highlights

Spanner Change Streams Proto Bytes Parsing: Introduced functionality to parse Spanner change stream records when they are output as proto bytes, specifically for PostgreSQL dialect databases.
ChangeStreamResultSet Enhancements: Added new methods getProtoChangeStreamRecordFromBytes() and isProtoBytesChangeRecord() to ChangeStreamResultSet to handle the extraction and parsing of proto bytes from the result set.
ChangeStreamRecordMapper Update: Modified ChangeStreamRecordMapper to conditionally parse proto bytes records for PostgreSQL, allowing reuse of existing proto parsing logic.
New Test Cases: Added comprehensive unit tests in ChangeStreamResultSetTest and ChangeStreamRecordMapperTest to validate the correct parsing and mapping of proto bytes change stream records.

🧠 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. ↩

tianz101 · 2026-01-27T00:50:56Z

...rc/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java

+  public com.google.spanner.v1.ChangeStreamRecord getProtoChangeStreamRecordFromBytes() {
+    recordReadAt = Timestamp.now();
+    // Use getBytes(0) for the BYTES column returned by read_proto_bytes_ TVF
+    byte[] protoBytes = resultSet.getBytes(0).toByteArray();


Can we pass this directly to the function below to avoid an extra copy of the bytes?

Updated. Thanks.

tianz101 · 2026-01-27T00:52:39Z

...rc/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java

        && resultSet.getColumnType(0).getCode() == com.google.cloud.spanner.Type.Code.PROTO;
  }

+  public com.google.spanner.v1.ChangeStreamRecord getProtoChangeStreamRecordFromBytes() {


Can we add accurate comments for this function, following similar convention for other functions? Thanks.

Added comments. Please take a look.

tianz101 · 2026-01-27T01:04:49Z

...rc/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java

+    try {
+      return com.google.spanner.v1.ChangeStreamRecord.parseFrom(protoBytes);
+    } catch (InvalidProtocolBufferException e) {
+      throw new RuntimeException("Failed to parse ChangeStreamRecord proto", e);


Maybe "Failed to parse the proto bytes to ChangeStreamRecord proto".

Updated. Thanks.

github-actions · 2026-01-27T02:51:28Z

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

chenxuesdu · 2026-01-27T23:55:41Z

Run Java_GCP_IO_Direct PreCommit

chenxuesdu · 2026-01-28T04:13:00Z

Run Java_GCP_IO_Direct PreCommit

chenxuesdu · 2026-01-28T18:06:01Z

Run Java_GCP_IO_Direct PreCommit

github-actions · 2026-01-28T20:45:50Z

Assigning reviewers:

R: @Abacn for label java.
R: @nielm for label spanner.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

tianz101 · 2026-01-28T22:14:47Z

...rc/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/ChangeStreamResultSet.java

   * Returns the only change stream record proto at the current pointer of the result set. It also
   * updates the timestamp at which the record was read. This function enhances the getProtoMessage
   * function but only focus on the ChangeStreamRecord type.
   *


Can you add a similar comment here for GoogleSQl databases like you did for line 136? Thanks.

tianz101 · 2026-01-28T22:18:40Z

...n/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/mapper/ChangeStreamRecordMapper.java

    if (this.isPostgres()) {
-      // In PostgresQL, change stream records are returned as JsonB.
+      // For `MUTABLE_KEY_RANGE` option, change stream records are returned as protos.
+      if (resultSet.isProtoBytesChangeRecord()) {


Maybe we have the getBytes(0) reflected here to match up getPgJsonb(0) so that it is very clear
that we should only expect one column from the result set.

nielm

LGTM - pending existing comments

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

tianz101 reviewed Jan 27, 2026

View reviewed changes

chenxuesdu force-pushed the master branch 3 times, most recently from a3858ab to 5e5c9aa Compare January 27, 2026 19:42

Parse proto bytes change stream record in ChangeStreamRecordMapper

0a07033

chenxuesdu force-pushed the master branch from 5e5c9aa to 0a07033 Compare January 28, 2026 18:16

github-actions bot added the Next Action: Reviewers label Jan 28, 2026

tianz101 reviewed Jan 28, 2026

View reviewed changes

tianz101 approved these changes Jan 28, 2026

View reviewed changes

nielm approved these changes Jan 29, 2026

View reviewed changes

[Spanner Change Streams] Parse proto bytes change stream record in ChangeStreamRecordMapper #37427

Are you sure you want to change the base?

[Spanner Change Streams] Parse proto bytes change stream record in ChangeStreamRecordMapper #37427

Conversation

chenxuesdu commented Jan 27, 2026

Uh oh!

gemini-code-assist bot commented Jan 27, 2026

Summary of Changes

Highlights

Footnotes

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!

github-actions bot commented Jan 27, 2026

Uh oh!

chenxuesdu commented Jan 27, 2026

Uh oh!

chenxuesdu commented Jan 28, 2026

Uh oh!

chenxuesdu commented Jan 28, 2026

Uh oh!

github-actions bot commented Jan 28, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nielm left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants