--- /dev/null
+.. change::
+ :tags: bug, sql
+ :tickets: 4313
+
+ Fixed bug where the multi-table support for UPDATE and DELETE statements
+ did not consider the additional FROM elements as targets for correlation,
+ when a correlated SELECT were also combined with the statement. This
+ change now includes that a SELECT statement in the WHERE clause for such a
+ statement will try to auto-correlate back to these additional tables in the
+ parent UPDATE/DELETE or unconditionally correlate if
+ :meth:`.Select.correlate` is used. Note that auto-correlation raises an
+ error if the SELECT statement would have no FROM clauses as a result, which
+ can now occur if the parent UPDATE/DELETE specifies the same tables in its
+ additional set of tables; specify :meth:`.Select.correlate` explicitly to
+ resolve.
def visit_update(self, update_stmt, asfrom=False, **kw):
toplevel = not self.stack
- self.stack.append(
- {'correlate_froms': {update_stmt.table},
- "asfrom_froms": {update_stmt.table},
- "selectable": update_stmt})
-
extra_froms = update_stmt._extra_froms
is_multitable = bool(extra_froms)
render_extra_froms = [
f for f in extra_froms if f not in main_froms
]
+ correlate_froms = main_froms.union(extra_froms)
else:
render_extra_froms = []
+ correlate_froms = {update_stmt.table}
+
+ self.stack.append(
+ {'correlate_froms': correlate_froms,
+ "asfrom_froms": correlate_froms,
+ "selectable": update_stmt})
text = "UPDATE "
def visit_delete(self, delete_stmt, asfrom=False, **kw):
toplevel = not self.stack
- self.stack.append({'correlate_froms': {delete_stmt.table},
- "asfrom_froms": {delete_stmt.table},
- "selectable": delete_stmt})
-
crud._setup_crud_params(self, delete_stmt, crud.ISDELETE, **kw)
extra_froms = delete_stmt._extra_froms
+ correlate_froms = {delete_stmt.table}.union(extra_froms)
+ self.stack.append({'correlate_froms': correlate_froms,
+ "asfrom_froms": correlate_froms,
+ "selectable": delete_stmt})
+
text = "DELETE "
if delete_stmt._prefixes:
for t in extra_froms)
def delete_extra_from_clause(self, update_stmt,
- from_table, extra_froms,
- from_hints,
- **kw):
+ from_table, extra_froms,
+ from_hints,
+ **kw):
return ', ' + ', '.join(
t._compiler_dispatch(self, asfrom=True,
fromhints=from_hints, **kw)
#! coding:utf-8
from sqlalchemy import Integer, String, ForeignKey, delete, select, and_, \
- or_
+ or_, exists
from sqlalchemy.dialects import mysql
+from sqlalchemy.engine import default
from sqlalchemy import testing
-from sqlalchemy.testing import AssertsCompiledSQL, fixtures, eq_
+from sqlalchemy import exc
+from sqlalchemy.testing import AssertsCompiledSQL, fixtures, eq_, \
+ assert_raises_message
from sqlalchemy.testing.schema import Table, Column
')')
+class DeleteFromCompileTest(
+ _DeleteTestBase, fixtures.TablesTest, AssertsCompiledSQL):
+ # DELETE FROM is also tested by individual dialects since there is no
+ # consistent syntax. here we use the StrSQLcompiler which has a fake
+ # syntax.
+
+ __dialect__ = 'default_enhanced'
+
+ def test_delete_extra_froms(self):
+ table1, table2 = self.tables.mytable, self.tables.myothertable
+
+ stmt = table1.delete().where(table1.c.myid == table2.c.otherid)
+ self.assert_compile(
+ stmt,
+ "DELETE FROM mytable , myothertable "
+ "WHERE mytable.myid = myothertable.otherid",
+ )
+
+ def test_correlation_to_extra(self):
+ table1, table2 = self.tables.mytable, self.tables.myothertable
+
+ stmt = table1.delete().where(
+ table1.c.myid == table2.c.otherid).where(
+ ~exists().where(table2.c.otherid == table1.c.myid).
+ where(table2.c.othername == 'x').correlate(table2)
+ )
+
+ self.assert_compile(
+ stmt,
+ "DELETE FROM mytable , myothertable WHERE mytable.myid = "
+ "myothertable.otherid AND NOT (EXISTS "
+ "(SELECT * FROM mytable WHERE myothertable.otherid = "
+ "mytable.myid AND myothertable.othername = :othername_1))",
+ )
+
+ def test_dont_correlate_to_extra(self):
+ table1, table2 = self.tables.mytable, self.tables.myothertable
+
+ stmt = table1.delete().where(
+ table1.c.myid == table2.c.otherid).where(
+ ~exists().where(table2.c.otherid == table1.c.myid).
+ where(table2.c.othername == 'x').correlate()
+ )
+
+ self.assert_compile(
+ stmt,
+ "DELETE FROM mytable , myothertable WHERE mytable.myid = "
+ "myothertable.otherid AND NOT (EXISTS "
+ "(SELECT * FROM myothertable, mytable "
+ "WHERE myothertable.otherid = "
+ "mytable.myid AND myothertable.othername = :othername_1))",
+ )
+
+ def test_autocorrelate_error(self):
+ table1, table2 = self.tables.mytable, self.tables.myothertable
+
+ stmt = table1.delete().where(
+ table1.c.myid == table2.c.otherid).where(
+ ~exists().where(table2.c.otherid == table1.c.myid).
+ where(table2.c.othername == 'x')
+ )
+
+ assert_raises_message(
+ exc.InvalidRequestError,
+ ".*returned no FROM clauses due to auto-correlation.*",
+ stmt.compile, dialect=default.StrCompileDialect()
+ )
+
+
class DeleteFromRoundTripTest(fixtures.TablesTest):
__backend__ = True
from sqlalchemy import Integer, String, ForeignKey, and_, or_, func, \
- literal, update, table, bindparam, column, select, exc
+ literal, update, table, bindparam, column, select, exc, exists
from sqlalchemy import testing
from sqlalchemy.dialects import mysql
from sqlalchemy.engine import default
'AND anon_1.email_address = :email_address_1',
checkparams=checkparams)
+ def test_correlation_to_extra(self):
+ users, addresses = self.tables.users, self.tables.addresses
+
+ stmt = users.update().values(name="newname").where(
+ users.c.id == addresses.c.user_id
+ ).where(
+ ~exists().where(
+ addresses.c.user_id == users.c.id
+ ).where(addresses.c.email_address == 'foo').correlate(addresses)
+ )
+
+ self.assert_compile(
+ stmt,
+ "UPDATE users SET name=:name FROM addresses WHERE "
+ "users.id = addresses.user_id AND NOT "
+ "(EXISTS (SELECT * FROM users WHERE addresses.user_id = users.id "
+ "AND addresses.email_address = :email_address_1))"
+ )
+
+ def test_dont_correlate_to_extra(self):
+ users, addresses = self.tables.users, self.tables.addresses
+
+ stmt = users.update().values(name="newname").where(
+ users.c.id == addresses.c.user_id
+ ).where(
+ ~exists().where(
+ addresses.c.user_id == users.c.id
+ ).where(addresses.c.email_address == 'foo').correlate()
+ )
+
+ self.assert_compile(
+ stmt,
+ "UPDATE users SET name=:name FROM addresses WHERE "
+ "users.id = addresses.user_id AND NOT "
+ "(EXISTS (SELECT * FROM addresses, users "
+ "WHERE addresses.user_id = users.id "
+ "AND addresses.email_address = :email_address_1))"
+ )
+
+ def test_autocorrelate_error(self):
+ users, addresses = self.tables.users, self.tables.addresses
+
+ stmt = users.update().values(name="newname").where(
+ users.c.id == addresses.c.user_id
+ ).where(
+ ~exists().where(
+ addresses.c.user_id == users.c.id
+ ).where(addresses.c.email_address == 'foo')
+ )
+
+ assert_raises_message(
+ exc.InvalidRequestError,
+ ".*returned no FROM clauses due to auto-correlation.*",
+ stmt.compile, dialect=default.StrCompileDialect()
+ )
+
class UpdateFromRoundTripTest(_UpdateFromTestBase, fixtures.TablesTest):
__backend__ = True
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
