[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.
   • Previously, when composedPartitionTokens was empty, the PostgreSQL branch omitted the "PartitionToken" filter (because Array[] without an explicit type cast is invalid syntax in string-concatenated PostgreSQL SQL), causing an unintended unfiltered scan and read lock on all rows matching State inside the read-write transaction even though updateToScheduled() updates 0 rows. Always applying "PartitionToken" = ANY ($2) with a typed Value.stringArray(...) parameter evaluates an empty list to false (0 rows), aligning PostgreSQL with the GoogleSQL branch (WHERE PartitionToken IN UNNEST(@composedPartitionTokens) AND State = @state).

Testing

• Added unit tests in PartitionMetadataDaoTest covering both GoogleSQL and PostgreSQL dialects (including 0, 1, and multiple partition tokens for updateToScheduled).

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

[scwhittle]

This seems to change the behavior for empty composedPartitionTokens array.
Previously it wouldn't check COLUMN_PARTITION_TOKEN but now I think that it will do ANY() which evaluates to false.

Maybe have a branch on if composedPartitionTokens is empty and have two statements you bind to?

[jiangzzhu]

Yeah, it seems change the behavior. However, this change may be safe

Here is why:

• Method semantics & all callers in PartitionMetadataDao.java: getPartitionsMatchingState(List composedPartitionTokens, State state) is private and only has two callers:

1. updateToScheduled(List composedPartitionTokens) (lines 535–557): queries which tokens among composedPartitionTokens are currently in State.CREATED, then loops for (String composedPartitionToken : composedPartitionTokens) to update only those tokens to SCHEDULED. When composedPartitionTokens is empty, that for loop executes 0 iterations and updates nothing. Under the old PostgreSQL code, omitting AND "PartitionToken" = ANY(...) ran SELECT * FROM "" WHERE "State" = 'CREATED' — performing an unnecessary full scan (and taking read locks on all CREATED rows in the table inside a read-write transaction) only to discard every row returned.
2. updateToRunning(String composedPartitionToken) (lines 564–582): always passes Collections.singletonList(composedPartitionToken) (never empty).

• Matches the GoogleSQL branch (lines 707–722):

java
"SELECT * FROM " + metadataTableName
+ " WHERE " + COLUMN_PARTITION_TOKEN + " IN UNNEST(@composedPartitionTokens) AND "
+ COLUMN_STATE + " = @State"

In GoogleSQL, passing an empty list binds an empty ARRAY to @composedPartitionTokens, and IN UNNEST([]) evaluates to false (0 rows).

The only reason the original PostgreSQL code had if (!composedPartitionTokens.isEmpty()) was because it used raw SQL string concatenation (= ANY (Array[...])), where an untyped empty literal Array[] (without ::text[]) is a PostgreSQL syntax error (cannot determine type of empty array).
With a typed bound Spanner parameter $2 (Value.stringArray(...)), Spanner PostgreSQL resolves "PartitionToken" = ANY ($2) to the exact same $in_array(STRING, ARRAY) AST as GoogleSQL's IN UNNEST(@composedPartitionTokens) (verified in parameterized_queries.test), cleanly returning 0 rows when $2 is empty.

[scwhittle]

add test coverage of 0 and multiple tokens as well

[jiangzzhu]

Done — added testInTransactionContextUpdateToScheduledPostgresEmptyTokens (0 tokens) and testInTransactionContextUpdateToScheduledPostgresMultipleTokens (multiple tokens) in 49ac029.

[scwhittle]

Thanks for clarifying!