Repository navigation
[sqlalchemy-spanner]: get_unique_constraints() splits multi-column unique constraints into separate single-column entries #18836
Description
Determine this is the right repository
- I determined this is the correct repository in which to report this bug.
Summary of the issue
SpannerDialect.get_unique_constraints() returns UNIQUE constraint instead of a single constraint dictionary containing all
Environment details
- OS: Linux
- Python version: 3.13
sqlalchemy-spannerversion: 1.17.2 (also reproduced onmain/ 1.20.1 ateb8a76628adf10093959600071a18f50285c2ecf)sqlalchemyversion: 2.0.52
Problem description
In SpannerDialect.get_unique_constraints:
sql = """
SELECT ccu.CONSTRAINT_NAME, ccu.COLUMN_NAME
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu
USING (TABLE_CATALOG, TABLE_SCHEMA, CONSTRAINT_NAME)
WHERE
tc.TABLE_NAME="{table_name}"
AND tc.TABLE_SCHEMA="{table_schema}"
AND tc.CONSTRAINT_TYPE = "UNIQUE"
AND tc.CONSTRAINT_NAME IS NOT NULL
""".format(table_schema=schema or "", table_name=table_name)
cols = []
with connection.connection.database.snapshot() as snap:
rows = snap.execute_sql(sql)
for row in rows:
cols.append({"name": row[0], "column_names": [row[1]]})
return colsThere are two bugs here when a table has a multi-column UNIQUE constraint (for example, the backing UNIQUE constraint Spanner creates in INFORMATION_SCHEMA.TABLE_CONSTRAINTS for a composite FOREIGN KEY (a, b) REFERENCES parent (col_b, col_a)):
- No grouping by constraint name: Every row in
CONSTRAINT_COLUMN_USAGEappends a separate{"name": row[0], "column_names": [row[1]]}dict, so a 2-column unique constraint produces twoUniqueConstraintobjects with the same name and 1 column each instead of oneUniqueConstraintwith 2 columns. - Missing column ordinal ordering:
INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGEhas noORDINAL_POSITIONcolumn. Preserving the constraint's declared column order requires joiningINFORMATION_SCHEMA.KEY_COLUMN_USAGEand ordering byORDINAL_POSITION(asget_multi_pk_constraintandget_multi_foreign_keysalready do).
Suggested fix
Join INFORMATION_SCHEMA.KEY_COLUMN_USAGE and aggregate columns per constraint with ARRAY_AGG(KCU.COLUMN_NAME ORDER BY KCU.ORDINAL_POSITION) (or implement get_multi_unique_constraints and delegate get_unique_constraints to it, matching get_columns, get_pk_constraint, get_foreign_keys, and get_indexes):
SELECT
TC.TABLE_NAME,
TC.CONSTRAINT_NAME,
ARRAY_AGG(KCU.COLUMN_NAME ORDER BY KCU.ORDINAL_POSITION) AS COLUMN_NAMES
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS TC
INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS KCU
USING (TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, CONSTRAINT_NAME)
WHERE
TC.TABLE_SCHEMA = @schema
AND TC.CONSTRAINT_TYPE = 'UNIQUE'
AND TC.CONSTRAINT_NAME IS NOT NULL
GROUP BY TC.TABLE_NAME, TC.CONSTRAINT_NAMEReproduction steps: code
import unittest.mock as mock
from google.cloud.sqlalchemy_spanner.sqlalchemy_spanner import SpannerDialect
dialect = SpannerDialect()
conn = mock.MagicMock()
snap = conn.connection.database.snapshot.return_value.__enter__.return_value
# Simulate INFORMATION_SCHEMA rows for a 2-column unique constraint uq_ab on (col_b, col_a)
snap.execute_sql.return_value = [
("uq_ab", "col_b"),
("uq_ab", "col_a"),
]
actual = dialect.get_unique_constraints(conn, "my_table")
print("Actual: ", actual)
print("Expected:", [{"name": "uq_ab", "column_names": ["col_b", "col_a"]}])Actual output
Actual: [{'name': 'uq_ab', 'column_names': ['col_b']}, {'name': 'uq_ab', 'column_names': ['col_a']}]
Expected output
Expected: [{'name': 'uq_ab', 'column_names': ['col_b', 'col_a']}]
Activity