Fixed bug whereby joining a select() of a table "A" with multiple

foreign key paths to a table "B", to that table "B", would fail
to produce the "ambiguous join condition" error that would be
reported if you join table "A" directly to "B"; it would instead
produce a join condition with multiple criteria.
[ticket:2738]

Conflicts:
	doc/build/changelog/changelog_09.rst

diff --git a/doc/build/changelog/changelog_08.rst b/doc/build/changelog/changelog_08.rst
index ed633ec6768f4557241087a9ef17595ce5813226..ec0027e736c151a10a1935b35b50feb056fde882 100644 (file)
--- a/doc/build/changelog/changelog_08.rst
+++ b/doc/build/changelog/changelog_08.rst
@@ -6,6 +6,16 @@
 .. changelog::
     :version: 0.8.2
 
+    .. change::
+        :tags: bug, sql
+        :tickets: 2738
+
+        Fixed bug whereby joining a select() of a table "A" with multiple
+        foreign key paths to a table "B", to that table "B", would fail
+        to produce the "ambiguous join condition" error that would be
+        reported if you join table "A" directly to "B"; it would instead
+        produce a join condition with multiple criteria.
+
     .. change::
         :tags: bug, sql, reflection
         :tickets: 2728

diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py
index d0964c37ac895e7a2531ffb5c753559bce43f077..6bd0283ff113a858010c682483204f479cb105b3 100644 (file)
--- a/lib/sqlalchemy/schema.py
+++ b/lib/sqlalchemy/schema.py
@@ -1138,7 +1138,8 @@ class Column(SchemaItem, expression.ColumnClause):
         information is not transferred.
 
         """
-        fk = [ForeignKey(f.column) for f in self.foreign_keys]
+        fk = [ForeignKey(f.column, _constraint=f.constraint)
+                for f in self.foreign_keys]
         if name is None and self.name is None:
             raise exc.InvalidRequestError("Cannot initialize a sub-selectable"
                     " with this Column object until it's 'name' has "

diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index db2eaa4fa75be6a937adabad2f1905e2036fb73e..fbdd06565a172214d04c9a74cf3b8dbb62fd42c0 100644 (file)
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -1211,6 +1211,30 @@ class ConstraintTest(fixtures.TestBase):
             schema.CreateTable(t1).compile
         )
 
+    def test_constraint_copied_to_proxy_ok(self):
+        m = MetaData()
+        t1 = Table('t1', m, Column('id', Integer, primary_key=True))
+        t2 = Table('t2', m, Column('id', Integer, ForeignKey('t1.id'),
+                        primary_key=True))
+
+        s = tsa.select([t2])
+        t2fk = list(t2.c.id.foreign_keys)[0]
+        sfk = list(s.c.id.foreign_keys)[0]
+
+        # the two FKs share the ForeignKeyConstraint
+        is_(
+            t2fk.constraint,
+            sfk.constraint
+        )
+
+        # but the ForeignKeyConstraint isn't
+        # aware of the select's FK
+        eq_(
+            t2fk.constraint.elements,
+            [t2fk]
+        )
+
+
 class ColumnDefinitionTest(AssertsCompiledSQL, fixtures.TestBase):
     """Test Column() construction."""
 

diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 183b721110b9ae0a007e77eaab754fabd5316acb..840abecdfc13e3607d34accabde2ee17caccc507 100644 (file)
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -781,15 +781,21 @@ class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults):
     def test_join_condition(self):
         m = MetaData()
         t1 = Table('t1', m, Column('id', Integer))
-        t2 = Table('t2', m, Column('id', Integer), Column('t1id',
-                   ForeignKey('t1.id')))
-        t3 = Table('t3', m, Column('id', Integer), Column('t1id',
-                   ForeignKey('t1.id')), Column('t2id',
-                   ForeignKey('t2.id')))
-        t4 = Table('t4', m, Column('id', Integer), Column('t2id',
-                   ForeignKey('t2.id')))
+        t2 = Table('t2', m, Column('id', Integer),
+                    Column('t1id', ForeignKey('t1.id')))
+        t3 = Table('t3', m,
+                    Column('id', Integer),
+                    Column('t1id', ForeignKey('t1.id')),
+                    Column('t2id', ForeignKey('t2.id')))
+        t4 = Table('t4', m, Column('id', Integer),
+                    Column('t2id', ForeignKey('t2.id')))
+        t5 = Table('t5', m,
+                    Column('t1id1', ForeignKey('t1.id')),
+                    Column('t1id2', ForeignKey('t1.id')),
+            )
         t1t2 = t1.join(t2)
         t2t3 = t2.join(t3)
+
         for (left, right, a_subset, expected) in [
             (t1, t2, None, t1.c.id == t2.c.t1id),
             (t1t2, t3, t2, t1t2.c.t2_id == t3.c.t2id),
@@ -803,12 +809,15 @@ class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults):
             assert expected.compare(sql_util.join_condition(left,
                                     right, a_subset=a_subset))
 
+
         # these are ambiguous, or have no joins
         for left, right, a_subset in [
             (t1t2, t3, None),
             (t2t3, t1, None),
             (t1, t4, None),
             (t1t2, t2t3, None),
+            (t5, t1, None),
+            (t5.select(use_labels=True), t1, None)
         ]:
             assert_raises(
                 exc.ArgumentError,