Don't rely on string col name in adapt_to_context

fixed an issue where even though the method claims to be
matching up columns positionally, it was failing on that by
looking in "keymap" based on string name.

Adds a new member to the _keymap recs MD_RESULT_MAP_INDEX
so that we can efficiently link from the generated keymap
back to the compiled._result_columns structure without
any ambiguity.

Fixes: #5559

Change-Id: Ie2fa9165c16625ef860ffac1190e00575e96761f

5 files changed, 96 insertions, 30 deletions

diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py
index a4a8cd99f..8bfaded8f 100644
--- a/test/dialect/oracle/test_compiler.py
+++ b/test/dialect/oracle/test_compiler.py
@@ -545,7 +545,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         stmt = select(type_coerce(column("x"), MyType).label("foo")).limit(1)
         dialect = oracle.dialect()
         compiled = stmt.compile(dialect=dialect)
-        assert isinstance(compiled._create_result_map()["foo"][-1], MyType)
+        assert isinstance(compiled._create_result_map()["foo"][-2], MyType)
 
     def test_use_binds_for_limits_disabled_one(self):
         t = table("sometable", column("col1"), column("col2"))
@@ -1061,8 +1061,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         eq_(
             compiled._create_result_map(),
             {
-                "c3": ("c3", (t1.c.c3, "c3", "c3"), t1.c.c3.type),
-                "lower": ("lower", (fn, "lower", None), fn.type),
+                "c3": ("c3", (t1.c.c3, "c3", "c3"), t1.c.c3.type, 1),
+                "lower": ("lower", (fn, "lower", None), fn.type, 0),
             },
         )
 
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 1084d30cb..b43d09045 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4716,6 +4716,7 @@ class SchemaTest(fixtures.TestBase, AssertsCompiledSQL):
                         "here_yetagain_anotherid",
                     ),
                     t1.c.anotherid.type,
+                    0,
                 )
             },
         )
@@ -5200,8 +5201,8 @@ class ResultMapTest(fixtures.TestBase):
         eq_(
             comp._create_result_map(),
             {
-                "a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type),
-                "b": ("b", (t.c.b, "b", "b", "t_b"), t.c.b.type),
+                "a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type, 0),
+                "b": ("b", (t.c.b, "b", "b", "t_b"), t.c.b.type, 1),
             },
         )
 
@@ -5212,7 +5213,7 @@ class ResultMapTest(fixtures.TestBase):
         comp = stmt.compile()
         eq_(
             comp._create_result_map(),
-            {"a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type)},
+            {"a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type, 0)},
         )
 
     def test_compound_only_top_populates(self):
@@ -5221,7 +5222,7 @@ class ResultMapTest(fixtures.TestBase):
         comp = stmt.compile()
         eq_(
             comp._create_result_map(),
-            {"a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type)},
+            {"a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type, 0)},
         )
 
     def test_label_plus_element(self):
@@ -5234,12 +5235,13 @@ class ResultMapTest(fixtures.TestBase):
         eq_(
             comp._create_result_map(),
             {
-                "a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type),
-                "bar": ("bar", (l1, "bar"), l1.type),
+                "a": ("a", (t.c.a, "a", "a", "t_a"), t.c.a.type, 0),
+                "bar": ("bar", (l1, "bar"), l1.type, 1),
                 "anon_1": (
                     tc.anon_label,
                     (tc_anon_label, "anon_1", tc),
                     tc.type,
+                    2,
                 ),
             },
         )
@@ -5279,7 +5281,7 @@ class ResultMapTest(fixtures.TestBase):
         comp = stmt.compile(dialect=postgresql.dialect())
         eq_(
             comp._create_result_map(),
-            {"a": ("a", (aint, "a", "a", "t2_a"), aint.type)},
+            {"a": ("a", (aint, "a", "a", "t2_a"), aint.type, 0)},
         )
 
     def test_insert_from_select(self):
@@ -5293,7 +5295,7 @@ class ResultMapTest(fixtures.TestBase):
         comp = stmt.compile(dialect=postgresql.dialect())
         eq_(
             comp._create_result_map(),
-            {"a": ("a", (aint, "a", "a", "t2_a"), aint.type)},
+            {"a": ("a", (aint, "a", "a", "t2_a"), aint.type, 0)},
         )
 
     def test_nested_api(self):
@@ -5339,6 +5341,7 @@ class ResultMapTest(fixtures.TestBase):
                         "myothertable_otherid",
                     ),
                     table2.c.otherid.type,
+                    0,
                 ),
                 "othername": (
                     "othername",
@@ -5349,8 +5352,9 @@ class ResultMapTest(fixtures.TestBase):
                         "myothertable_othername",
                     ),
                     table2.c.othername.type,
+                    1,
                 ),
-                "k1": ("k1", (1, 2, 3), int_),
+                "k1": ("k1", (1, 2, 3), int_, 2),
             },
         )
         eq_(
@@ -5360,12 +5364,14 @@ class ResultMapTest(fixtures.TestBase):
                     "myid",
                     (table1.c.myid, "myid", "myid", "mytable_myid"),
                     table1.c.myid.type,
+                    0,
                 ),
-                "k2": ("k2", (3, 4, 5), int_),
+                "k2": ("k2", (3, 4, 5), int_, 3),
                 "name": (
                     "name",
                     (table1.c.name, "name", "name", "mytable_name"),
                     table1.c.name.type,
+                    1,
                 ),
                 "description": (
                     "description",
@@ -5376,6 +5382,7 @@ class ResultMapTest(fixtures.TestBase):
                         "mytable_description",
                     ),
                     table1.c.description.type,
+                    2,
                 ),
             },
         )
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index c83c71ada..04fed9b6e 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -811,6 +811,7 @@ class TextualSelectTest(fixtures.TestBase, AssertsCompiledSQL):
                     "myid",
                     (table1.c.myid, "myid", "myid", "mytable_myid"),
                     table1.c.myid.type,
+                    0,
                 )
             },
         )
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index 44c1565e4..578e20e44 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -1475,6 +1475,16 @@ class KeyTargetingTest(fixtures.TablesTest):
                 schema=testing.config.test_schema,
             )
 
+        Table(
+            "users",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("team_id", metadata, ForeignKey("teams.id")),
+        )
+        Table(
+            "teams", metadata, Column("id", Integer, primary_key=True),
+        )
+
     @classmethod
     def insert_data(cls, connection):
         conn = connection
@@ -1484,6 +1494,9 @@ class KeyTargetingTest(fixtures.TablesTest):
         conn.execute(cls.tables.keyed4.insert(), dict(b="b4", q="q4"))
         conn.execute(cls.tables.content.insert(), dict(type="t1"))
 
+        conn.execute(cls.tables.teams.insert(), dict(id=1))
+        conn.execute(cls.tables.users.insert(), dict(id=1, team_id=1))
+
         if testing.requires.schemas.enabled:
             conn.execute(
                 cls.tables["%s.wschema" % testing.config.test_schema].insert(),
@@ -1815,7 +1828,7 @@ class KeyTargetingTest(fixtures.TablesTest):
 
     def _adapt_result_columns_fixture_two(self):
         return text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(
-            keyed2_a=CHAR, keyed2_b=CHAR
+            column("keyed2_a", CHAR), column("keyed2_b", CHAR)
         )
 
     def _adapt_result_columns_fixture_three(self):
@@ -1834,11 +1847,34 @@ class KeyTargetingTest(fixtures.TablesTest):
 
         return stmt2
 
+    def _adapt_result_columns_fixture_five(self):
+        users, teams = self.tables("users", "teams")
+        return select([users.c.id, teams.c.id]).select_from(
+            users.outerjoin(teams)
+        )
+
+    def _adapt_result_columns_fixture_six(self):
+        # this has _result_columns structure that is not ordered
+        # the same as the cursor.description.
+        return text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(
+            keyed2_b=CHAR, keyed2_a=CHAR,
+        )
+
+    def _adapt_result_columns_fixture_seven(self):
+        # this has _result_columns structure that is not ordered
+        # the same as the cursor.description.
+        return text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(
+            keyed2_b=CHAR, bogus_col=CHAR
+        )
+
     @testing.combinations(
         _adapt_result_columns_fixture_one,
         _adapt_result_columns_fixture_two,
         _adapt_result_columns_fixture_three,
         _adapt_result_columns_fixture_four,
+        _adapt_result_columns_fixture_five,
+        _adapt_result_columns_fixture_six,
+        _adapt_result_columns_fixture_seven,
         argnames="stmt_fn",
     )
     def test_adapt_result_columns(self, connection, stmt_fn):
@@ -1863,31 +1899,41 @@ class KeyTargetingTest(fixtures.TablesTest):
             zip(stmt1.selected_columns, stmt2.selected_columns)
         )
 
-        result = connection.execute(stmt1)
+        for i in range(2):
+            try:
+                result = connection.execute(stmt1)
 
-        mock_context = Mock(
-            compiled=result.context.compiled, invoked_statement=stmt2
-        )
-        existing_metadata = result._metadata
-        adapted_metadata = existing_metadata._adapt_to_context(mock_context)
+                mock_context = Mock(
+                    compiled=result.context.compiled, invoked_statement=stmt2
+                )
+                existing_metadata = result._metadata
+                adapted_metadata = existing_metadata._adapt_to_context(
+                    mock_context
+                )
 
-        eq_(existing_metadata.keys, adapted_metadata.keys)
+                eq_(existing_metadata.keys, adapted_metadata.keys)
 
-        for k in existing_metadata._keymap:
-            if isinstance(k, ColumnElement) and k in column_linkage:
-                other_k = column_linkage[k]
-            else:
-                other_k = k
+                for k in existing_metadata._keymap:
+                    if isinstance(k, ColumnElement) and k in column_linkage:
+                        other_k = column_linkage[k]
+                    else:
+                        other_k = k
 
-            is_(
-                existing_metadata._keymap[k], adapted_metadata._keymap[other_k]
-            )
+                    is_(
+                        existing_metadata._keymap[k],
+                        adapted_metadata._keymap[other_k],
+                    )
+            finally:
+                result.close()
 
     @testing.combinations(
         _adapt_result_columns_fixture_one,
         _adapt_result_columns_fixture_two,
         _adapt_result_columns_fixture_three,
         _adapt_result_columns_fixture_four,
+        _adapt_result_columns_fixture_five,
+        _adapt_result_columns_fixture_six,
+        _adapt_result_columns_fixture_seven,
         argnames="stmt_fn",
     )
     def test_adapt_result_columns_from_cache(self, connection, stmt_fn):
@@ -1909,7 +1955,10 @@ class KeyTargetingTest(fixtures.TablesTest):
 
         row = result.first()
         for col in stmt2.selected_columns:
-            assert col in row._mapping
+            if "bogus" in col.name:
+                assert col not in row._mapping
+            else:
+                assert col in row._mapping
 
 
 class PositionalTextTest(fixtures.TablesTest):
diff --git a/test/sql/test_text.py b/test/sql/test_text.py
index 1a7ee6f34..9d5ab65ed 100644
--- a/test/sql/test_text.py
+++ b/test/sql/test_text.py
@@ -435,6 +435,8 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
             column("id", Integer), column("name")
         )
 
+        col_pos = {col.name: idx for idx, col in enumerate(t.selected_columns)}
+
         compiled = t.compile()
         eq_(
             compiled._create_result_map(),
@@ -443,11 +445,13 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
                     "id",
                     (t.selected_columns.id, "id", "id", "id"),
                     t.selected_columns.id.type,
+                    col_pos["id"],
                 ),
                 "name": (
                     "name",
                     (t.selected_columns.name, "name", "name", "name"),
                     t.selected_columns.name.type,
+                    col_pos["name"],
                 ),
             },
         )
@@ -455,6 +459,8 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
     def test_basic_toplevel_resultmap(self):
         t = text("select id, name from user").columns(id=Integer, name=String)
 
+        col_pos = {col.name: idx for idx, col in enumerate(t.selected_columns)}
+
         compiled = t.compile()
         eq_(
             compiled._create_result_map(),
@@ -463,11 +469,13 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
                     "id",
                     (t.selected_columns.id, "id", "id", "id"),
                     t.selected_columns.id.type,
+                    col_pos["id"],
                 ),
                 "name": (
                     "name",
                     (t.selected_columns.name, "name", "name", "name"),
                     t.selected_columns.name.type,
+                    col_pos["name"],
                 ),
             },
         )
@@ -490,6 +498,7 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
                     "myid",
                     (table1.c.myid, "myid", "myid", "mytable_myid"),
                     table1.c.myid.type,
+                    0,
                 )
             },
         )