fix: column order in get_multi_pk_constraint (#640)

* fix: column order in get_multi_pk_constraint

* chore: order by catalog and schema name

---------

Co-authored-by: Toby Woodwark <[email protected]>
Co-authored-by: Knut Olav Løite <[email protected]>

Author: 3 people

google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -1304,16 +1304,18 @@ def get_multi_pk_constraint(
         table_type_query = self._get_table_type_query(kind, True)
 
         sql = """
-            SELECT tc.table_schema, tc.table_name, ccu.COLUMN_NAME
+            SELECT tc.table_schema, tc.table_name, kcu.column_name
             FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
-            JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu
+            JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS kcu
                 USING (TABLE_CATALOG, TABLE_SCHEMA, CONSTRAINT_NAME)
             JOIN information_schema.tables AS t
                 ON  tc.TABLE_CATALOG = t.TABLE_CATALOG
                 AND tc.TABLE_SCHEMA = t.TABLE_SCHEMA
                 AND tc.TABLE_NAME = t.TABLE_NAME
             WHERE {table_filter_query} {table_type_query}
             {schema_filter_query} tc.CONSTRAINT_TYPE = "PRIMARY KEY"
+            ORDER BY tc.table_catalog ASC, tc.table_schema ASC,
+                     tc.table_name ASC, kcu.ordinal_position ASC
         """.format(
             table_filter_query=table_filter_query,
             table_type_query=table_type_query,

test/test_suite_13.py

@@ -1014,22 +1014,6 @@ def test_fk_column_order(self):
         eq_(set(fkey1.get("referred_columns")), {"name", "id", "attr"})
         eq_(set(fkey1.get("constrained_columns")), {"pname", "pid", "pattr"})
 
-    @testing.requires.primary_key_constraint_reflection
-    def test_pk_column_order(self, connection):
-        """
-        SPANNER OVERRIDE:
-        Emultor doesn't support returning pk sorted by ordinal value
-        of columns.
-        """
-        insp = inspect(connection)
-        primary_key = insp.get_pk_constraint(self.tables.tb1.name)
-        exp = (
-            ["id", "name", "attr"]
-            if bool(os.environ.get("SPANNER_EMULATOR_HOST"))
-            else ["name", "id", "attr"]
-        )
-        eq_(primary_key.get("constrained_columns"), exp)
-
 
 class RowFetchTest(_RowFetchTest):
     def test_row_w_scalar_select(self):

test/test_suite_14.py

@@ -858,22 +858,6 @@ def test_fk_column_order(self):
         eq_(set(fkey1.get("referred_columns")), {"name", "id", "attr"})
         eq_(set(fkey1.get("constrained_columns")), {"pname", "pid", "pattr"})
 
-    @testing.requires.primary_key_constraint_reflection
-    def test_pk_column_order(self, connection):
-        """
-        SPANNER OVERRIDE:
-        Emultor doesn't support returning pk sorted by ordinal value
-        of columns.
-        """
-        insp = inspect(connection)
-        primary_key = insp.get_pk_constraint(self.tables.tb1.name)
-        exp = (
-            ["id", "name", "attr"]
-            if bool(os.environ.get("SPANNER_EMULATOR_HOST"))
-            else ["name", "id", "attr"]
-        )
-        eq_(primary_key.get("constrained_columns"), exp)
-
 
 @pytest.mark.skip("Spanner doesn't support quotes in table names.")
 class QuotedNameArgumentTest(_QuotedNameArgumentTest):

test/test_suite_20.py

@@ -1539,22 +1539,6 @@ def test_fk_column_order(self, connection):
         eq_(set(fkey1.get("referred_columns")), {"name", "id", "attr"})
         eq_(set(fkey1.get("constrained_columns")), {"pname", "pid", "pattr"})
 
-    @testing.requires.primary_key_constraint_reflection
-    def test_pk_column_order(self, connection):
-        """
-        SPANNER OVERRIDE:
-        Emultor doesn't support returning pk sorted by ordinal value
-        of columns.
-        """
-        insp = inspect(connection)
-        primary_key = insp.get_pk_constraint(self.tables.tb1.name)
-        exp = (
-            ["id", "name", "attr"]
-            if bool(os.environ.get("SPANNER_EMULATOR_HOST"))
-            else ["name", "id", "attr"]
-        )
-        eq_(primary_key.get("constrained_columns"), exp)
-
 
 @pytest.mark.skip("Spanner doesn't support quotes in table names.")
 class QuotedNameArgumentTest(_QuotedNameArgumentTest):