From f2ee514c757fc9ec33afaddc2a7b96d08b83a164 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Tue, 5 Jul 2016 12:48:41 -0400 Subject: Adapt "FOR UPDATE OF" with Oracle limit/offset This modifies the Oracle ROWNUM limit/offset approach to accommodate for the "OF" clause in a "FOR UPDATE" phrase. The column expressions must be added to the selected subquery if necessary and adapted on the outside. Change-Id: Ia71b5fc4df6d326e73863f8ae9f96e8f1a5acfc1 Fixes: #3741 --- test/dialect/test_oracle.py | 73 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) (limited to 'test') diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py index ed09141bb..1bdddb3bc 100644 --- a/test/dialect/test_oracle.py +++ b/test/dialect/test_oracle.py @@ -365,6 +365,79 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "mytable_1.myid, mytable_1.name" ) + def test_for_update_of_w_limit_adaption_col_present(self): + table1 = table('mytable', column('myid'), column('name')) + + self.assert_compile( + select([table1.c.myid, table1.c.name]). + where(table1.c.myid == 7). + with_for_update(nowait=True, of=table1.c.name). + limit(10), + "SELECT myid, name FROM " + "(SELECT mytable.myid AS myid, mytable.name AS name " + "FROM mytable WHERE mytable.myid = :myid_1) " + "WHERE ROWNUM <= :param_1 FOR UPDATE OF name NOWAIT", + ) + + def test_for_update_of_w_limit_adaption_col_unpresent(self): + table1 = table('mytable', column('myid'), column('name')) + + self.assert_compile( + select([table1.c.myid]). + where(table1.c.myid == 7). + with_for_update(nowait=True, of=table1.c.name). + limit(10), + "SELECT myid FROM " + "(SELECT mytable.myid AS myid, mytable.name AS name " + "FROM mytable WHERE mytable.myid = :myid_1) " + "WHERE ROWNUM <= :param_1 FOR UPDATE OF name NOWAIT", + ) + + def test_for_update_of_w_limit_offset_adaption_col_present(self): + table1 = table('mytable', column('myid'), column('name')) + + self.assert_compile( + select([table1.c.myid, table1.c.name]). + where(table1.c.myid == 7). + with_for_update(nowait=True, of=table1.c.name). + limit(10).offset(50), + "SELECT myid, name FROM (SELECT myid, name, ROWNUM AS ora_rn " + "FROM (SELECT mytable.myid AS myid, mytable.name AS name " + "FROM mytable WHERE mytable.myid = :myid_1) " + "WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > :param_2 " + "FOR UPDATE OF name NOWAIT", + ) + + def test_for_update_of_w_limit_offset_adaption_col_unpresent(self): + table1 = table('mytable', column('myid'), column('name')) + + self.assert_compile( + select([table1.c.myid]). + where(table1.c.myid == 7). + with_for_update(nowait=True, of=table1.c.name). + limit(10).offset(50), + "SELECT myid FROM (SELECT myid, ROWNUM AS ora_rn, name " + "FROM (SELECT mytable.myid AS myid, mytable.name AS name " + "FROM mytable WHERE mytable.myid = :myid_1) " + "WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > :param_2 " + "FOR UPDATE OF name NOWAIT", + ) + + def test_for_update_of_w_limit_offset_adaption_partial_col_unpresent(self): + table1 = table('mytable', column('myid'), column('foo'), column('bar')) + + self.assert_compile( + select([table1.c.myid, table1.c.bar]). + where(table1.c.myid == 7). + with_for_update(nowait=True, of=[table1.c.foo, table1.c.bar]). + limit(10).offset(50), + "SELECT myid, bar FROM (SELECT myid, bar, ROWNUM AS ora_rn, " + "foo FROM (SELECT mytable.myid AS myid, mytable.bar AS bar, " + "mytable.foo AS foo FROM mytable WHERE mytable.myid = :myid_1) " + "WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > :param_2 " + "FOR UPDATE OF foo, bar NOWAIT" + ) + def test_limit_preserves_typing_information(self): class MyType(TypeDecorator): impl = Integer -- cgit v1.2.1