feat(spanner): add send and ack mutations Cloud Spanner Queues - #9200
feat(spanner): add send and ack mutations Cloud Spanner Queues#9200finnzzf wants to merge 5 commits into
Conversation
Update the system test instance to use ENTERPRISE edition for access to queue features.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces support for queue operations in Cloud Spanner by adding queueSend and queueAck methods to Transaction, MutationSet, and MutationGroup, along with corresponding options interfaces and mutation builders. It also adds unit and system tests for these new features. A review comment points out an issue in the system tests where the queue support flags are not set to false when the emulator is enabled, which would cause the tests to run and fail on the emulator.
|
presubmits are also failing, please fix that as well |
The lint errors do not seem to be related to my changes. They are coming from existing codes. |
finnzzf
changed the title
could you please run these two commands and push the changes?
|
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces support for queue send and ack mutations in Cloud Spanner transactions, including the necessary interfaces, builder functions, and methods across Transaction, MutationSet, and MutationGroup. It also adds corresponding unit and system tests. Feedback is provided to improve the system tests by cleaning up created queues to ensure test idempotency and to avoid using magic numbers for the instance edition.
| before(async () => { | ||
| try { | ||
| const queueDdl = `CREATE QUEUE ${QUEUE_NAME} ( | ||
| Id INT64 NOT NULL, | ||
| Payload STRING(MAX) NOT NULL | ||
| ) PRIMARY KEY (Id)`; | ||
| const [operation] = await DATABASE.updateSchema(queueDdl); | ||
| await operation.promise(); | ||
| } catch (err: any) { | ||
| if ( | ||
| err.code === 9 || | ||
| err.code === 12 || | ||
| err.message.includes('UNIMPLEMENTED') | ||
| ) { | ||
| gsqlQueueSupported = false; | ||
| } else { | ||
| throw err; | ||
| } | ||
| } | ||
|
|
||
| try { | ||
| const pgQueueDdl = `CREATE QUEUE ${QUEUE_NAME} ( | ||
| id bigint NOT NULL, | ||
| "Payload" varchar NOT NULL, | ||
| PRIMARY KEY (id) | ||
| )`; | ||
| const [pgOperation] = await PG_DATABASE.updateSchema(pgQueueDdl); | ||
| await pgOperation.promise(); | ||
| } catch (err: any) { | ||
| if ( | ||
| err.code === 9 || | ||
| err.code === 12 || | ||
| err.message.includes('UNIMPLEMENTED') | ||
| ) { | ||
| pgQueueSupported = false; | ||
| } else { | ||
| throw err; | ||
| } | ||
| } | ||
| }); |
There was a problem hiding this comment.
The system tests create queues but do not clean them up afterward. If the tests are run repeatedly against a persistent database instance, the 'before' block will fail with 'ALREADY_EXISTS' (gRPC code 6) on subsequent runs, causing the entire test suite to fail.
To make the tests robust and idempotent, we should:
- Handle 'ALREADY_EXISTS' (code 6) in the 'before' block and treat it as the feature being supported.
- Add an 'after' block to drop the created queues.
Avoid implementing defensive checks or setup for unsupported dialects like GoogleSQL, as the driver is scoped to support PostgreSQL.
before(async () => {
try {
const pgQueueDdl = "CREATE QUEUE " + QUEUE_NAME + " (\n id bigint NOT NULL,\n \"Payload\" varchar NOT NULL,\n PRIMARY KEY (id)\n )";
const [pgOperation] = await PG_DATABASE.updateSchema(pgQueueDdl);
await pgOperation.promise();
} catch (err: any) {
if (
err.code === 9 ||
err.code === 12 ||
err.message.includes("UNIMPLEMENTED")
) {
pgQueueSupported = false;
} else if (err.code === 6) {
pgQueueSupported = true;
} else {
throw err;
}
}
});
after(async () => {
if (pgQueueSupported) {
try {
const [pgOperation] = await PG_DATABASE.updateSchema(
"DROP QUEUE " + QUEUE_NAME
);
await pgOperation.promise();
} catch (err) {
// Ignore cleanup errors
}
}
});References
- Do not implement features, types, or defensive checks for unsupported dialects (such as GoogleSQL) if the driver is currently scoped to support only a specific dialect (such as PostgreSQL).
| INSTANCE_CONFIG.config, | ||
| ), | ||
| nodeCount: 1, | ||
| edition: 2, |
There was a problem hiding this comment.
Avoid using magic numbers like '2' for the instance edition. Please add a descriptive comment or use the appropriate enum if available to improve readability and maintainability.
| edition: 2, | |
| edition: 2, // ENTERPRISE |
2 participants
Cloud Spanner Queue Mutation Support
Fixes #9199 🦕