diff options
Diffstat (limited to 'test/dialect/oracle/test_compiler.py')
| -rw-r--r-- | test/dialect/oracle/test_compiler.py | 243 |
1 files changed, 214 insertions, 29 deletions
diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py index ecf43a2cf..45a83ed77 100644 --- a/test/dialect/oracle/test_compiler.py +++ b/test/dialect/oracle/test_compiler.py @@ -47,6 +47,12 @@ from sqlalchemy.testing.schema import Table class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = "oracle" + @testing.fixture + def legacy_oracle_limitoffset(self): + self.__dialect__ = oracle.OracleDialect(enable_offset_fetch=False) + yield + del self.__dialect__ + def test_true_false(self): self.assert_compile(sql.false(), "0") self.assert_compile(sql.true(), "1") @@ -159,7 +165,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "GROUP BY included_parts.sub_part", ) - def test_limit_one(self): + def test_limit_one_legacy(self, legacy_oracle_limitoffset): t = table("sometable", column("col1"), column("col2")) s = select(t) c = s.compile(dialect=oracle.OracleDialect()) @@ -182,6 +188,24 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): eq_(len(c._result_columns), 2) assert t.c.col1 in set(c._create_result_map()["col1"][1]) + def test_limit_one(self): + t = table("sometable", column("col1"), column("col2")) + s = select(t) + c = s.compile(dialect=oracle.OracleDialect()) + assert t.c.col1 in set(c._create_result_map()["col1"][1]) + s = select(t).limit(10).offset(20) + self.assert_compile( + s, + "SELECT sometable.col1, sometable.col2 FROM sometable " + "OFFSET __[POSTCOMPILE_param_1] ROWS " + "FETCH FIRST __[POSTCOMPILE_param_2] ROWS ONLY", + checkparams={"param_1": 20, "param_2": 10}, + ) + + c = s.compile(dialect=oracle.OracleDialect()) + eq_(len(c._result_columns), 2) + assert t.c.col1 in set(c._create_result_map()["col1"][1]) + def test_limit_one_literal_binds(self): """test for #6863. @@ -197,6 +221,25 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): eq_ignore_whitespace( str(c), + "SELECT sometable.col1, sometable.col2 FROM sometable " + "OFFSET 20 ROWS FETCH FIRST 10 ROWS ONLY", + ) + + def test_limit_one_literal_binds_legacy(self, legacy_oracle_limitoffset): + """test for #6863. + + the bug does not appear to have affected Oracle's case. + + """ + t = table("sometable", column("col1"), column("col2")) + s = select(t).limit(10).offset(20) + c = s.compile( + dialect=oracle.OracleDialect(enable_offset_fetch=False), + compile_kwargs={"literal_binds": True}, + ) + + eq_ignore_whitespace( + str(c), "SELECT anon_1.col1, anon_1.col2 FROM " "(SELECT anon_2.col1 AS col1, anon_2.col2 AS col2, " "ROWNUM AS ora_rn FROM (SELECT sometable.col1 AS col1, " @@ -204,7 +247,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "WHERE ROWNUM <= 10 + 20) anon_1 WHERE ora_rn > 20", ) - def test_limit_one_firstrows(self): + def test_limit_one_firstrows_legacy(self): t = table("sometable", column("col1"), column("col2")) s = select(t) s = select(t).limit(10).offset(20) @@ -220,7 +263,34 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "WHERE ora_rn > " "__[POSTCOMPILE_param_2]", checkparams={"param_1": 10, "param_2": 20}, - dialect=oracle.OracleDialect(optimize_limits=True), + dialect=oracle.OracleDialect( + optimize_limits=True, enable_offset_fetch=False + ), + ) + + def test_simple_fetch(self): + # as of #8221, all FETCH / ROWS ONLY is using postcompile params; + # this is in the spirit of the ROWNUM approach where users reported + # that bound parameters caused performance degradation + t = table("sometable", column("col1"), column("col2")) + s = select(t) + s = select(t).fetch(10) + self.assert_compile( + s, + "SELECT sometable.col1, sometable.col2 FROM sometable " + "FETCH FIRST __[POSTCOMPILE_param_1] ROWS ONLY", + checkparams={"param_1": 10}, + ) + + def test_simple_fetch_offset(self): + t = table("sometable", column("col1"), column("col2")) + s = select(t).fetch(10).offset(20) + self.assert_compile( + s, + "SELECT sometable.col1, sometable.col2 FROM sometable " + "OFFSET __[POSTCOMPILE_param_1] ROWS " + "FETCH FIRST __[POSTCOMPILE_param_2] ROWS ONLY", + checkparams={"param_1": 20, "param_2": 10}, ) def test_limit_two(self): @@ -230,6 +300,33 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): s2 = select(s.c.col1, s.c.col2) self.assert_compile( s2, + "SELECT anon_1.col1, anon_1.col2 FROM (SELECT sometable.col1 AS " + "col1, sometable.col2 AS col2 FROM sometable OFFSET " + "__[POSTCOMPILE_param_1] ROWS FETCH FIRST " + "__[POSTCOMPILE_param_2] ROWS ONLY) anon_1", + checkparams={"param_1": 20, "param_2": 10}, + ) + + self.assert_compile( + s2, + "SELECT anon_1.col1, anon_1.col2 FROM (SELECT sometable.col1 AS " + "col1, sometable.col2 AS col2 FROM sometable OFFSET 20 " + "ROWS FETCH FIRST 10 ROWS ONLY) anon_1", + render_postcompile=True, + ) + c = s2.compile(dialect=oracle.OracleDialect()) + eq_(len(c._result_columns), 2) + assert s.c.col1 in set(c._create_result_map()["col1"][1]) + + def test_limit_two_legacy(self): + t = table("sometable", column("col1"), column("col2")) + s = select(t).limit(10).offset(20).subquery() + + s2 = select(s.c.col1, s.c.col2) + + dialect = oracle.OracleDialect(enable_offset_fetch=False) + self.assert_compile( + s2, "SELECT anon_1.col1, anon_1.col2 FROM " "(SELECT anon_2.col1 AS col1, " "anon_2.col2 AS col2 " @@ -242,6 +339,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "anon_2 " "WHERE ora_rn > __[POSTCOMPILE_param_2]) anon_1", checkparams={"param_1": 10, "param_2": 20}, + dialect=dialect, ) self.assert_compile( @@ -257,8 +355,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "__[POSTCOMPILE_param_2]) " "anon_2 " "WHERE ora_rn > __[POSTCOMPILE_param_2]) anon_1", + dialect=dialect, ) - c = s2.compile(dialect=oracle.OracleDialect()) + c = s2.compile(dialect=dialect) eq_(len(c._result_columns), 2) assert s.c.col1 in set(c._create_result_map()["col1"][1]) @@ -268,6 +367,22 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): s = select(t).limit(10).offset(20).order_by(t.c.col2) self.assert_compile( s, + "SELECT sometable.col1, sometable.col2 FROM sometable " + "ORDER BY sometable.col2 OFFSET __[POSTCOMPILE_param_1] " + "ROWS FETCH FIRST __[POSTCOMPILE_param_2] ROWS ONLY", + checkparams={"param_1": 20, "param_2": 10}, + ) + c = s.compile(dialect=oracle.OracleDialect()) + eq_(len(c._result_columns), 2) + assert t.c.col1 in set(c._create_result_map()["col1"][1]) + + def test_limit_three_legacy(self): + t = table("sometable", column("col1"), column("col2")) + + s = select(t).limit(10).offset(20).order_by(t.c.col2) + dialect = oracle.OracleDialect(enable_offset_fetch=False) + self.assert_compile( + s, "SELECT anon_1.col1, anon_1.col2 FROM " "(SELECT anon_2.col1 AS col1, " "anon_2.col2 AS col2, ROWNUM AS ora_rn FROM (SELECT " @@ -277,12 +392,13 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "__[POSTCOMPILE_param_1] + __[POSTCOMPILE_param_2]) anon_1 " "WHERE ora_rn > __[POSTCOMPILE_param_2]", checkparams={"param_1": 10, "param_2": 20}, + dialect=dialect, ) - c = s.compile(dialect=oracle.OracleDialect()) + c = s.compile(dialect=dialect) eq_(len(c._result_columns), 2) assert t.c.col1 in set(c._create_result_map()["col1"][1]) - def test_limit_four(self): + def test_limit_four_legacy(self, legacy_oracle_limitoffset): t = table("sometable", column("col1"), column("col2")) s = select(t).with_for_update().limit(10).order_by(t.c.col2) @@ -296,7 +412,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"param_1": 10}, ) - def test_limit_four_firstrows(self): + def test_limit_four_firstrows_legacy(self): t = table("sometable", column("col1"), column("col2")) s = select(t).with_for_update().limit(10).order_by(t.c.col2) @@ -309,7 +425,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "sometable.col2) anon_1 WHERE ROWNUM <= __[POSTCOMPILE_param_1] " "FOR UPDATE", checkparams={"param_1": 10}, - dialect=oracle.OracleDialect(optimize_limits=True), + dialect=oracle.OracleDialect( + optimize_limits=True, enable_offset_fetch=False + ), ) def test_limit_five(self): @@ -318,6 +436,18 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): s = select(t).with_for_update().limit(10).offset(20).order_by(t.c.col2) self.assert_compile( s, + "SELECT sometable.col1, sometable.col2 FROM sometable " + "ORDER BY sometable.col2 OFFSET __[POSTCOMPILE_param_1] ROWS " + "FETCH FIRST __[POSTCOMPILE_param_2] ROWS ONLY FOR UPDATE", + checkparams={"param_1": 20, "param_2": 10}, + ) + + def test_limit_five_legacy(self, legacy_oracle_limitoffset): + t = table("sometable", column("col1"), column("col2")) + + s = select(t).with_for_update().limit(10).offset(20).order_by(t.c.col2) + self.assert_compile( + s, "SELECT anon_1.col1, anon_1.col2 FROM " "(SELECT anon_2.col1 AS col1, " "anon_2.col2 AS col2, ROWNUM AS ora_rn FROM (SELECT " @@ -341,6 +471,23 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) self.assert_compile( s, + "SELECT sometable.col1, sometable.col2 FROM sometable " + "ORDER BY sometable.col2 OFFSET :param_1 + :param_2 " + "ROWS FETCH FIRST __[POSTCOMPILE_param_3] ROWS ONLY", + checkparams={"param_1": 10, "param_2": 20, "param_3": 10}, + ) + + def test_limit_six_legacy(self, legacy_oracle_limitoffset): + t = table("sometable", column("col1"), column("col2")) + + s = ( + select(t) + .limit(10) + .offset(literal(10) + literal(20)) + .order_by(t.c.col2) + ) + self.assert_compile( + s, "SELECT anon_1.col1, anon_1.col2 FROM (SELECT anon_2.col1 AS " "col1, anon_2.col2 AS col2, ROWNUM AS ora_rn FROM " "(SELECT sometable.col1 AS col1, sometable.col2 AS col2 " @@ -350,7 +497,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"param_1": 10, "param_2": 10, "param_3": 20}, ) - def test_limit_special_quoting(self): + def test_limit_special_quoting_legacy(self, legacy_oracle_limitoffset): """Oracle-specific test for #4730. Even though this issue is generic, test the originally reported Oracle @@ -510,7 +657,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF table1", ) - def test_for_update_of_w_limit_adaption_col_present(self): + def test_for_update_of_w_limit_col_present_legacy( + self, legacy_oracle_limitoffset + ): table1 = table("mytable", column("myid"), column("name")) self.assert_compile( @@ -526,7 +675,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"param_1": 10, "myid_1": 7}, ) - def test_for_update_of_w_limit_adaption_col_unpresent(self): + def test_for_update_of_w_limit_col_unpresent_legacy( + self, legacy_oracle_limitoffset + ): table1 = table("mytable", column("myid"), column("name")) self.assert_compile( @@ -541,7 +692,25 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "FOR UPDATE OF anon_1.name NOWAIT", ) - def test_for_update_of_w_limit_offset_adaption_col_present(self): + def test_for_update_of_w_limit_offset_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 mytable.myid, mytable.name FROM mytable " + "WHERE mytable.myid = :myid_1 OFFSET __[POSTCOMPILE_param_1] " + "ROWS FETCH FIRST __[POSTCOMPILE_param_2] ROWS ONLY " + "FOR UPDATE OF mytable.name NOWAIT", + checkparams={"param_1": 50, "param_2": 10, "myid_1": 7}, + ) + + def test_for_update_of_w_limit_offset_col_present_legacy( + self, legacy_oracle_limitoffset + ): table1 = table("mytable", column("myid"), column("name")) self.assert_compile( @@ -563,7 +732,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"param_1": 10, "param_2": 50, "myid_1": 7}, ) - def test_for_update_of_w_limit_offset_adaption_col_unpresent(self): + def test_for_update_of_w_limit_offset_col_unpresent_legacy( + self, legacy_oracle_limitoffset + ): table1 = table("mytable", column("myid"), column("name")) self.assert_compile( @@ -584,7 +755,9 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"param_1": 10, "param_2": 50, "myid_1": 7}, ) - def test_for_update_of_w_limit_offset_adaption_partial_col_unpresent(self): + def test_for_update_of_w_limit_offset_partial_col_unpresent_legacy( + self, legacy_oracle_limitoffset + ): table1 = table("mytable", column("myid"), column("foo"), column("bar")) self.assert_compile( @@ -607,23 +780,25 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"param_1": 10, "param_2": 50, "myid_1": 7}, ) - def test_limit_preserves_typing_information(self): + def test_limit_preserves_typing_information_legacy(self): class MyType(TypeDecorator): impl = Integer cache_ok = True stmt = select(type_coerce(column("x"), MyType).label("foo")).limit(1) - dialect = oracle.dialect() + dialect = oracle.dialect(enable_offset_fetch=False) compiled = stmt.compile(dialect=dialect) assert isinstance(compiled._create_result_map()["foo"][-2], MyType) - def test_use_binds_for_limits_disabled_one(self): + def test_use_binds_for_limits_disabled_one_legacy(self): t = table("sometable", column("col1"), column("col2")) with testing.expect_deprecated( "The ``use_binds_for_limits`` Oracle dialect parameter is " "deprecated." ): - dialect = oracle.OracleDialect(use_binds_for_limits=False) + dialect = oracle.OracleDialect( + use_binds_for_limits=False, enable_offset_fetch=False + ) self.assert_compile( select(t).limit(10), @@ -634,13 +809,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=dialect, ) - def test_use_binds_for_limits_disabled_two(self): + def test_use_binds_for_limits_disabled_two_legacy(self): t = table("sometable", column("col1"), column("col2")) with testing.expect_deprecated( "The ``use_binds_for_limits`` Oracle dialect parameter is " "deprecated." ): - dialect = oracle.OracleDialect(use_binds_for_limits=False) + dialect = oracle.OracleDialect( + use_binds_for_limits=False, enable_offset_fetch=False + ) self.assert_compile( select(t).offset(10), @@ -652,13 +829,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=dialect, ) - def test_use_binds_for_limits_disabled_three(self): + def test_use_binds_for_limits_disabled_three_legacy(self): t = table("sometable", column("col1"), column("col2")) with testing.expect_deprecated( "The ``use_binds_for_limits`` Oracle dialect parameter is " "deprecated." ): - dialect = oracle.OracleDialect(use_binds_for_limits=False) + dialect = oracle.OracleDialect( + use_binds_for_limits=False, enable_offset_fetch=False + ) self.assert_compile( select(t).limit(10).offset(10), @@ -672,13 +851,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=dialect, ) - def test_use_binds_for_limits_enabled_one(self): + def test_use_binds_for_limits_enabled_one_legacy(self): t = table("sometable", column("col1"), column("col2")) with testing.expect_deprecated( "The ``use_binds_for_limits`` Oracle dialect parameter is " "deprecated." ): - dialect = oracle.OracleDialect(use_binds_for_limits=True) + dialect = oracle.OracleDialect( + use_binds_for_limits=True, enable_offset_fetch=False + ) self.assert_compile( select(t).limit(10), @@ -689,13 +870,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=dialect, ) - def test_use_binds_for_limits_enabled_two(self): + def test_use_binds_for_limits_enabled_two_legacy(self): t = table("sometable", column("col1"), column("col2")) with testing.expect_deprecated( "The ``use_binds_for_limits`` Oracle dialect parameter is " "deprecated." ): - dialect = oracle.OracleDialect(use_binds_for_limits=True) + dialect = oracle.OracleDialect( + use_binds_for_limits=True, enable_offset_fetch=False + ) self.assert_compile( select(t).offset(10), @@ -708,13 +891,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=dialect, ) - def test_use_binds_for_limits_enabled_three(self): + def test_use_binds_for_limits_enabled_three_legacy(self): t = table("sometable", column("col1"), column("col2")) with testing.expect_deprecated( "The ``use_binds_for_limits`` Oracle dialect parameter is " "deprecated." ): - dialect = oracle.OracleDialect(use_binds_for_limits=True) + dialect = oracle.OracleDialect( + use_binds_for_limits=True, enable_offset_fetch=False + ) self.assert_compile( select(t).limit(10).offset(10), @@ -932,7 +1117,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "anon_1 " "WHERE ora_rn > __[POSTCOMPILE_param_2]", checkparams={"param_1": 10, "param_2": 5}, - dialect=oracle.dialect(use_ansi=False), + dialect=oracle.dialect(use_ansi=False, enable_offset_fetch=False), ) def test_outer_join_six(self): |