[jiangzzhu]

Summary

This PR fixes two PostgreSQL metadata query issues in the Spanner Change Streams connector (PartitionMetadataDao):

1. PartitionMetadataDao.tableExists() on PostgreSQL-dialect databases:

   • Previously, tableExists() hardcoded WHERE t.table_catalog = '' AND t.table_schema = ''. In PostgreSQL-dialect Spanner databases, table_catalog is the database name and user tables reside in table_schema = 'public', causing tableExists() to always return false.
   • When InitializeDoFn runs more than once (e.g., on retry or pipeline restart with an existing metadata table), createPartitionMetadataTable() (CREATE TABLE IF NOT EXISTS) succeeds as a no-op, and createFakeParentPartition() subsequently fails with ALREADY_EXISTS: Row [Parent0] in table Metadata_... already exists.
   • Updated tableExists() to check t.table_schema = 'public' when this.isPostgres() is true (consistent with findAllTableIndexes()).

2. PartitionMetadataDao.getPartitionsMatchingState() on PostgreSQL-dialect databases:

   • Updated the PostgreSQL branch of getPartitionsMatchingState() to use bound query parameters (WHERE "State" = $1 AND "PartitionToken" = ANY ($2)) instead of string concatenation, and to omit the "PartitionToken" filter when composedPartitionTokens is empty (matching the GoogleSQL branch behavior).

Testing

• Added unit tests in PartitionMetadataDaoTest covering both GoogleSQL and PostgreSQL dialects.

[github-actions]

📋 Review Summary

This PR resolves three important bugs in Spanner Change Streams (SpannerIO.readChangeStream()): fixing PartitionMetadataDao.tableExists() for PostgreSQL-dialect databases by checking table_schema = 'public', parameterizing getPartitionsMatchingState() and short-circuiting empty token updates in updateToScheduled(), and adding support for schema-qualified change stream and TVF names across SpannerIO and ChangeStreamDao. The changes are well-structured, adhere to existing SpannerIO patterns, and include comprehensive unit tests for both GoogleSQL and PostgreSQL dialects.

🔍 General Feedback

• Switching PartitionMetadataDao.getPartitionsMatchingState() in the PostgreSQL branch from string concatenation to bound query parameters ($1 and ANY ($2)) is a great improvement for both query plan caching and safety.
• The schema-qualification handling across fetchPartitionMode, checkTvfExistence, and ChangeStreamDao.changeStreamQuery cleanly accounts for the differences between GoogleSQL (<schema>.READ_<name>) and PostgreSQL ("spanner"."read_json_<name>" / "spanner"."read_proto_bytes_<name>").

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[chenxuesdu]

LGTM

[github-actions]

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

[github-actions]

Assigning reviewers:

R: @nielm for label spanner.
Backup reviewer: chamikaramj (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

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).