fix(sqlalchemy-spanner): isolate literal round trip test tables with unique naming and teardown

[chalmerlowe]

Fix Schema Collision in SQLAlchemy Spanner Literal Tests

Problem

In the Cloud Spanner SQLAlchemy compliance tests, tests using the literal_round_trip_spanner fixture were failing with 400 Duplicate name in schema: t_literal_round_trip_spanner.

This occurred because the fixture created a table with a static, hardcoded name (t_literal_round_trip_spanner) without dropping the table when each test completed. When subsequent test methods in the same class (such as StringTest and NumericTest) invoked the fixture, Cloud Spanner rejected the creation of an already existing table.

Solution

1. Unique Table Naming: Generated an isolated table name per test invocation using the column type and a random hexadecimal suffix (for example, t_lit_string_a3f2b1).
2. Explicit DDL Flushes: Added a helper to flush and commit buffered Data Definition Language (DDL) statements in the Spanner database API immediately after table creation and insertion.
3. Guaranteed Teardown: Wrapped test execution in a try...finally block that drops the temporary table, commits the drop DDL statement, and removes the table reference from SQLAlchemy metadata so class teardowns do not collide.

Notes to Reviewers

• This change is localized to packages/sqlalchemy-spanner/tests/conftest.py.
• It provides test isolation for literal rendering tests, eliminating flakes and unblocking automated dependency updates without modifying runtime dialect behavior.

[gemini-code-assist]

Code Review

This pull request enhances test isolation in the literal_round_trip_spanner fixture by generating unique table names using UUIDs, explicitly committing DDL operations, and adding a finally block to drop the table and remove it from metadata. The review feedback suggests three improvements: using dbapi_connection with a fallback to support SQLAlchemy 2.0 compatibility, moving the table creation inside the try block to ensure cleanup if creation fails, and wrapping the table drop in a nested try...finally block to guarantee that the table is always removed from metadata even if the drop operation itself raises an exception.

[gemini-code-assist]

In SQLAlchemy 2.0, Connection.connection is deprecated in favor of Connection.dbapi_connection. To ensure compatibility with both SQLAlchemy 1.4 and 2.0 without triggering deprecation warnings, retrieve dbapi_connection first, falling back to connection if it is not present.

Suggested change

	dbapi_conn = getattr(connection, "connection", connection)
	dbapi_conn = getattr(connection, "dbapi_connection", getattr(connection, "connection", connection))

[gemini-code-assist]

If t.create(connection) fails or raises an exception, the try...finally block is never entered. Since Table automatically registers itself in the metadata object upon instantiation, this leaves a stale table reference in metadata and can leak the table in the database if the creation partially succeeded. Wrapping the table creation inside the try block ensures that the cleanup in the finally block is always executed.

Suggested change

	t.create(connection)
	for value in input_:
	ins = t.insert().values(x=literal(value, type_))
	connection.execute(ins)
	if support_whereclause:
	if compare:
	stmt = t.select().where(
	t.c.x
	== literal(
	compare,
	type_,
	),
	t.c.x
	== literal(
	input_[0],
	type_,
	),
	)
	else:
	stmt = t.select().where(
	t.c.x
	== literal(
	compare if compare is not None else input_[0],
	type_,
	_commit_ddl(connection)
	try:
	try:
	t.create(connection)
	_commit_ddl(connection)

[gemini-code-assist]

If t.drop(connection) or _commit_ddl(connection) raises an exception (e.g., due to connection issues), metadata.remove(t) will be bypassed, leaving a stale table reference in metadata. Wrapping the drop in a nested try...finally guarantees that metadata.remove(t) is always executed.

Suggested change

	finally:
	t.drop(connection, checkfirst=True)
	_commit_ddl(connection)
	metadata.remove(t)
	finally:
	try:
	t.drop(connection, checkfirst=True)
	_commit_ddl(connection)
	finally:
	metadata.remove(t)