Merge "graceful degrade for FKs not reflectable" into main

6 files changed, 292 insertions, 51 deletions

diff --git a/doc/build/changelog/unreleased_14/8100.rst b/doc/build/changelog/unreleased_14/8100.rst
new file mode 100644
index 000000000..7c5fc49aa
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8100.rst
@@ -0,0 +1,30 @@
+.. change::
+    :tags: bug, reflection
+    :tickets: 8100, 8101
+
+    Fixed bugs involving the :paramref:`.Table.include_columns` and the
+    :paramref:`.Table.resolve_fks` parameters on :class:`.Table`; these
+    little-used parameters were apparently not working for columns that refer
+    to foreign key constraints.
+
+    In the first case, not-included columns that refer to foreign keys would
+    still attempt to create a :class:`.ForeignKey` object, producing errors
+    when attempting to resolve the columns for the foreign key constraint
+    within reflection; foreign key constraints that refer to skipped columns
+    are now omitted from the table reflection process in the same way as
+    occurs for :class:`.Index` and :class:`.UniqueConstraint` objects with the
+    same conditions. No warning is produced however, as we likely want to
+    remove the include_columns warnings for all constraints in 2.0.
+
+    In the latter case, the production of table aliases or subqueries would
+    fail on an FK related table not found despite the presence of
+    ``resolve_fks=False``; the logic has been repaired so that if a related
+    table is not found, the :class:`.ForeignKey` object is still proxied to the
+    aliased table or subquery (these :class:`.ForeignKey` objects are normally
+    used in the production of join conditions), but it is sent with a flag that
+    it's not resolvable. The aliased table / subquery will then work normally,
+    with the exception that it cannot be used to generate a join condition
+    automatically, as the foreign key information is missing. This was already
+    the behavior for such foreign key constraints produced using non-reflection
+    methods, such as joining :class:`.Table` objects from different
+    :class:`.MetaData` collections.
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index b8ece2b1d..4fc57d5f4 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -779,6 +779,7 @@ class Inspector(inspection.Inspectable["Inspector"]):
             schema,
             table,
             cols_by_orig_name,
+            include_columns,
             exclude_columns,
             resolve_fks,
             _extend_on,
@@ -922,6 +923,7 @@ class Inspector(inspection.Inspectable["Inspector"]):
         schema,
         table,
         cols_by_orig_name,
+        include_columns,
         exclude_columns,
         resolve_fks,
         _extend_on,
@@ -938,10 +940,17 @@ class Inspector(inspection.Inspectable["Inspector"]):
                 cols_by_orig_name[c].key if c in cols_by_orig_name else c
                 for c in fkey_d["constrained_columns"]
             ]
-            if exclude_columns and set(constrained_columns).intersection(
+
+            if (
                 exclude_columns
+                and set(constrained_columns).intersection(exclude_columns)
+                or (
+                    include_columns
+                    and set(constrained_columns).difference(include_columns)
+                )
             ):
                 continue
+
             referred_schema = fkey_d["referred_schema"]
             referred_table = fkey_d["referred_table"]
             referred_columns = fkey_d["referred_columns"]
@@ -976,6 +985,7 @@ class Inspector(inspection.Inspectable["Inspector"]):
                 options = fkey_d["options"]
             else:
                 options = {}
+
             table.append_constraint(
                 sa_schema.ForeignKeyConstraint(
                     constrained_columns,
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 447e102ed..c37b60003 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -2271,10 +2271,19 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause[_T]):
         information is not transferred.
 
         """
+
         fk = [
-            ForeignKey(f.column, _constraint=f.constraint)
-            for f in self.foreign_keys
+            ForeignKey(
+                col if col is not None else f._colspec,
+                _unresolvable=col is None,
+                _constraint=f.constraint,
+            )
+            for f, col in [
+                (fk, fk._resolve_column(raiseerr=False))
+                for fk in self.foreign_keys
+            ]
         ]
+
         if name is None and self.name is None:
             raise exc.InvalidRequestError(
                 "Cannot initialize a sub-selectable"
@@ -2375,6 +2384,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
         link_to_name: bool = False,
         match: Optional[str] = None,
         info: Optional[_InfoType] = None,
+        _unresolvable: bool = False,
         **dialect_kw: Any,
     ):
         r"""
@@ -2448,6 +2458,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
         """
 
         self._colspec = coercions.expect(roles.DDLReferredColumnRole, column)
+        self._unresolvable = _unresolvable
 
         if isinstance(self._colspec, str):
             self._table_column = None
@@ -2658,6 +2669,11 @@ class ForeignKey(DialectKWArgs, SchemaItem):
 
         parenttable = self.parent.table
 
+        if self._unresolvable:
+            schema, tname, colname = self._column_tokens
+            tablekey = _get_table_key(tname, schema)
+            return parenttable, tablekey, colname
+
         # assertion
         # basically Column._make_proxy() sends the actual
         # target Column to the ForeignKey object, so the
@@ -2742,13 +2758,30 @@ class ForeignKey(DialectKWArgs, SchemaItem):
 
         """
 
+        return self._resolve_column()
+
+    @overload
+    def _resolve_column(self, *, raiseerr: Literal[True] = ...) -> Column[Any]:
+        ...
+
+    @overload
+    def _resolve_column(
+        self, *, raiseerr: bool = ...
+    ) -> Optional[Column[Any]]:
+        ...
+
+    def _resolve_column(
+        self, *, raiseerr: bool = True
+    ) -> Optional[Column[Any]]:
         _column: Column[Any]
 
         if isinstance(self._colspec, str):
 
             parenttable, tablekey, colname = self._resolve_col_tokens()
 
-            if tablekey not in parenttable.metadata:
+            if self._unresolvable or tablekey not in parenttable.metadata:
+                if not raiseerr:
+                    return None
                 raise exc.NoReferencedTableError(
                     "Foreign key associated with column '%s' could not find "
                     "table '%s' with which to generate a "
@@ -2757,6 +2790,8 @@ class ForeignKey(DialectKWArgs, SchemaItem):
                     tablekey,
                 )
             elif parenttable.key not in parenttable.metadata:
+                if not raiseerr:
+                    return None
                 raise exc.InvalidRequestError(
                     "Table %s is no longer associated with its "
                     "parent MetaData" % parenttable
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 390e23952..0400ab3fe 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -235,7 +235,7 @@ def find_left_clause_to_join_from(
                 if set(f.c).union(s.c).issuperset(cols_in_onclause):
                     idx.append(i)
                     break
-            elif Join._can_join(f, s) or onclause is not None:
+            elif onclause is not None or Join._can_join(f, s):
                 idx.append(i)
                 break
 
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index d2e3d5f40..76099e863 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -12,6 +12,7 @@ from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
 from sqlalchemy import schema
+from sqlalchemy import select
 from sqlalchemy import sql
 from sqlalchemy import String
 from sqlalchemy import testing
@@ -23,6 +24,7 @@ from sqlalchemy.testing import ComparesTables
 from sqlalchemy.testing import config
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import eq_regex
+from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import in_
@@ -252,44 +254,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         )
         assert "nonexistent" not in meta.tables
 
-    def test_include_columns(self, connection, metadata):
-        meta = metadata
-        foo = Table(
-            "foo",
-            meta,
-            *[
-                Column(n, sa.String(30))
-                for n in ["a", "b", "c", "d", "e", "f"]
-            ],
-        )
-        meta.create_all(connection)
-        meta2 = MetaData()
-        foo = Table(
-            "foo",
-            meta2,
-            autoload_with=connection,
-            include_columns=["b", "f", "e"],
-        )
-        # test that cols come back in original order
-        eq_([c.name for c in foo.c], ["b", "e", "f"])
-        for c in ("b", "f", "e"):
-            assert c in foo.c
-        for c in ("a", "c", "d"):
-            assert c not in foo.c
-
-        # test against a table which is already reflected
-        meta3 = MetaData()
-        foo = Table("foo", meta3, autoload_with=connection)
-
-        foo = Table(
-            "foo", meta3, include_columns=["b", "f", "e"], extend_existing=True
-        )
-        eq_([c.name for c in foo.c], ["b", "e", "f"])
-        for c in ("b", "f", "e"):
-            assert c in foo.c
-        for c in ("a", "c", "d"):
-            assert c not in foo.c
-
     def test_extend_existing(self, connection, metadata):
         meta = metadata
 
@@ -2237,3 +2201,159 @@ class IdentityColumnTest(fixtures.TablesTest):
         is_true(table.c.id1.identity is not None)
         eq_(table.c.id1.identity.start, 2)
         eq_(table.c.id1.identity.increment, 3)
+
+
+class IncludeColsFksTest(AssertsCompiledSQL, fixtures.TestBase):
+    __dialect__ = "default"
+
+    @testing.fixture
+    def tab_wo_fks(self, connection, metadata):
+        meta = metadata
+        foo = Table(
+            "foo",
+            meta,
+            *[
+                Column(n, sa.String(30))
+                for n in ["a", "b", "c", "d", "e", "f"]
+            ],
+        )
+        meta.create_all(connection)
+
+        return foo
+
+    @testing.fixture
+    def tab_w_fks(self, connection, metadata):
+        Table(
+            "a",
+            metadata,
+            Column("x", Integer, primary_key=True),
+            test_needs_fk=True,
+        )
+
+        b = Table(
+            "b",
+            metadata,
+            Column("x", Integer, primary_key=True),
+            Column("q", Integer),
+            Column("p", Integer),
+            Column("r", Integer, ForeignKey("a.x")),
+            Column("s", Integer),
+            Column("t", Integer),
+            test_needs_fk=True,
+        )
+
+        metadata.create_all(connection)
+
+        return b
+
+    def test_include_columns(self, connection, tab_wo_fks):
+        foo = tab_wo_fks
+        meta2 = MetaData()
+        foo = Table(
+            "foo",
+            meta2,
+            autoload_with=connection,
+            include_columns=["b", "f", "e"],
+        )
+        # test that cols come back in original order
+        eq_([c.name for c in foo.c], ["b", "e", "f"])
+        for c in ("b", "f", "e"):
+            assert c in foo.c
+        for c in ("a", "c", "d"):
+            assert c not in foo.c
+
+        # test against a table which is already reflected
+        meta3 = MetaData()
+        foo = Table("foo", meta3, autoload_with=connection)
+
+        foo = Table(
+            "foo", meta3, include_columns=["b", "f", "e"], extend_existing=True
+        )
+        eq_([c.name for c in foo.c], ["b", "e", "f"])
+        for c in ("b", "f", "e"):
+            assert c in foo.c
+        for c in ("a", "c", "d"):
+            assert c not in foo.c
+
+    @testing.emits_warning
+    @testing.combinations(True, False, argnames="resolve_fks")
+    def test_include_cols_skip_fk_col(
+        self, connection, tab_w_fks, resolve_fks
+    ):
+        """test #8100"""
+
+        m2 = MetaData()
+
+        b2 = Table(
+            "b",
+            m2,
+            autoload_with=connection,
+            resolve_fks=resolve_fks,
+            include_columns=["x", "q", "p"],
+        )
+
+        eq_([c.name for c in b2.c], ["x", "q", "p"])
+
+        # no FK, whether or not resolve_fks was called
+        eq_(b2.constraints, set((b2.primary_key,)))
+
+        b2a = b2.alias()
+        eq_([c.name for c in b2a.c], ["x", "q", "p"])
+
+        self.assert_compile(select(b2), "SELECT b.x, b.q, b.p FROM b")
+        self.assert_compile(
+            select(b2.alias()),
+            "SELECT b_1.x, b_1.q, b_1.p FROM b AS b_1",
+        )
+
+    def test_table_works_minus_fks(self, connection, tab_w_fks):
+        """test #8101"""
+
+        m2 = MetaData()
+
+        b2 = Table(
+            "b",
+            m2,
+            autoload_with=connection,
+            resolve_fks=False,
+        )
+
+        eq_([c.name for c in b2.c], ["x", "q", "p", "r", "s", "t"])
+
+        b2a = b2.alias()
+        eq_([c.name for c in b2a.c], ["x", "q", "p", "r", "s", "t"])
+
+        self.assert_compile(
+            select(b2), "SELECT b.x, b.q, b.p, b.r, b.s, b.t FROM b"
+        )
+        b2a_1 = b2.alias()
+        self.assert_compile(
+            select(b2a_1),
+            "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t FROM b AS b_1",
+        )
+
+        # reflecting the related table
+        a2 = Table("a", m2, autoload_with=connection)
+
+        # the existing alias doesn't know about it
+        with expect_raises_message(
+            sa.exc.InvalidRequestError,
+            "Foreign key associated with column 'anon_1.r' could not find "
+            "table 'a' with which to generate a foreign key to target "
+            "column 'x'",
+        ):
+            select(b2a_1).join(a2).compile()
+
+        # can still join manually (needed to fix inside of util for this...)
+        self.assert_compile(
+            select(b2a_1).join(a2, b2a_1.c.r == a2.c.x),
+            "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t "
+            "FROM b AS b_1 JOIN a ON b_1.r = a.x",
+        )
+
+        # a new alias does know about it however
+        self.assert_compile(
+            select(b2.alias()).join(a2),
+            "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t "
+            "FROM b AS b_1 JOIN a ON a.x = b_1.r",
+        )
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index d05d7ad8b..3ecbfca27 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1478,21 +1478,67 @@ class SelectableTest(
         assert j4.corresponding_column(j2.c.aid) is j4.c.aid
         assert j4.corresponding_column(a.c.id) is j4.c.id
 
-    def test_two_metadata_join_raises(self):
+    @testing.combinations(True, False)
+    def test_two_metadata_join_raises(self, include_a_joining_table):
+        """test case from 2008 enhanced as of #8101, more specific failure
+        modes for non-resolvable FKs
+
+        """
         m = MetaData()
         m2 = MetaData()
 
         t1 = Table("t1", m, Column("id", Integer), Column("id2", Integer))
-        t2 = Table("t2", m, Column("id", Integer, ForeignKey("t1.id")))
+
+        if include_a_joining_table:
+            t2 = Table("t2", m, Column("id", Integer, ForeignKey("t1.id")))
+
         t3 = Table("t3", m2, Column("id", Integer, ForeignKey("t1.id2")))
 
-        s = (
-            select(t2, t3)
-            .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
-            .subquery()
-        )
+        with expect_raises_message(
+            exc.NoReferencedTableError,
+            "Foreign key associated with column 't3.id'",
+        ):
+            t3.join(t1)
 
-        assert_raises(exc.NoReferencedTableError, s.join, t1)
+        if include_a_joining_table:
+            s = (
+                select(t2, t3)
+                .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
+                .subquery()
+            )
+        else:
+            s = (
+                select(t3)
+                .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
+                .subquery()
+            )
+
+        with expect_raises_message(
+            exc.NoReferencedTableError,
+            "Foreign key associated with column 'anon_1.t3_id' could not "
+            "find table 't1' with which to generate a foreign key to target "
+            "column 'id2'",
+        ):
+            select(s.join(t1)),
+
+        # manual join is OK.  using select().join() here is also exercising
+        # that join() does not need to resolve FKs if we provided the
+        # ON clause
+        if include_a_joining_table:
+            self.assert_compile(
+                select(s).join(
+                    t1, and_(s.c.t2_id == t1.c.id, s.c.t3_id == t1.c.id)
+                ),
+                "SELECT anon_1.t2_id, anon_1.t3_id FROM (SELECT "
+                "t2.id AS t2_id, t3.id AS t3_id FROM t2, t3) AS anon_1 "
+                "JOIN t1 ON anon_1.t2_id = t1.id AND anon_1.t3_id = t1.id",
+            )
+        else:
+            self.assert_compile(
+                select(s).join(t1, s.c.t3_id == t1.c.id),
+                "SELECT anon_1.t3_id FROM (SELECT t3.id AS t3_id FROM t3) "
+                "AS anon_1 JOIN t1 ON anon_1.t3_id = t1.id",
+            )
 
     def test_multi_label_chain_naming_col(self):
         # See [ticket:2167] for this one.