jiangzzhu opened a new pull request, #40755:
URL: https://github.com/apache/beam/pull/40755
## Summary
This PR fixes three issues in the Spanner Change Streams connector
(`SpannerIO.readChangeStream()`):
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()` and
`updateToScheduled()` on PostgreSQL-dialect databases**:
- Returning early from `updateToScheduled()` when
`composedPartitionTokens` is empty avoids unnecessary read/write queries.
- Updated the PostgreSQL branch of `getPartitionsMatchingState()` to use
bound query parameters (`WHERE "State" = $1 AND "PartitionToken" = ANY ($2)`)
instead of string concatenation and omitting the `PartitionToken` filter when
the list is empty.
3. **Schema-qualified change streams (`myschema.my_cs`)**:
- Updated `SpannerIO.fetchPartitionMode()` to split schema-qualified
change stream names (`<schema>.<name>`) and filter
`information_schema.change_stream_options` by both `change_stream_schema` and
`change_stream_name`.
- Updated `SpannerIO.checkTvfExistence()` to extract the unqualified
routine name when matching against `information_schema.routines.routine_name`.
- Updated `ChangeStreamDao.changeStreamQuery()` so that schema-qualified
change streams construct `SELECT * FROM <schema>.READ_<name>(...)` in GoogleSQL
(instead of `READ_<schema>.<name>`) and use the unqualified name for
`"spanner"."read_json_<name>"` / `"spanner"."read_proto_bytes_<name>"` in
PostgreSQL.
## Testing
- Added unit tests in `PartitionMetadataDaoTest`, `ChangeStreamDaoTest`, and
`SpannerIOReadChangeStreamTest` covering both GoogleSQL and PostgreSQL dialects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
