select = select.where(whereclause)
select._oracle_visit = True
- if select._limit is not None or select._offset is not None:
+ limit_clause = select._limit_clause
+ offset_clause = select._offset_clause
+ if limit_clause is not None or offset_clause is not None:
# See http://www.oracle.com/technology/oramag/oracle/06-sep/o56asktom.html
#
# Generalized form of an Oracle pagination query:
# Wrap the middle select and add the hint
limitselect = sql.select([c for c in select.c])
- limit = select._limit
- if limit and self.dialect.optimize_limits:
- limitselect = limitselect.prefix_with("/*+ FIRST_ROWS(%d) */" % limit)
+ if limit_clause is not None and \
+ self.dialect.optimize_limits and \
+ select._simple_int_limit:
+ limitselect = limitselect.prefix_with(
+ "/*+ FIRST_ROWS(%d) */" %
+ select._limit)
limitselect._oracle_visit = True
limitselect._is_wrapper = True
# If needed, add the limiting clause
- offset = select._offset
- if limit is not None:
- max_row = limit
- if offset is not None:
- max_row += offset
+ if limit_clause is not None:
if not self.dialect.use_binds_for_limits:
+ # use simple int limits, will raise an exception
+ # if the limit isn't specified this way
+ max_row = select._limit
+
+ if offset_clause is not None:
+ max_row += select._offset
max_row = sql.literal_column("%d" % max_row)
+ else:
+ max_row = limit_clause
+ if offset_clause is not None:
+ max_row = max_row + offset_clause
limitselect.append_whereclause(
sql.literal_column("ROWNUM") <= max_row)
# If needed, add the ora_rn, and wrap again with offset.
- if offset is None:
+ if offset_clause is None:
limitselect._for_update_arg = select._for_update_arg
select = limitselect
else:
limitselect = limitselect.column(
- sql.literal_column("ROWNUM").label("ora_rn"))
+ sql.literal_column("ROWNUM").label("ora_rn"))
limitselect._oracle_visit = True
limitselect._is_wrapper = True
offsetselect = sql.select(
- [c for c in limitselect.c if c.key != 'ora_rn'])
+ [c for c in limitselect.c if c.key != 'ora_rn'])
offsetselect._oracle_visit = True
offsetselect._is_wrapper = True
- offset_value = offset
if not self.dialect.use_binds_for_limits:
- offset_value = sql.literal_column("%d" % offset_value)
+ offset_clause = sql.literal_column(
+ "%d" % select._offset)
offsetselect.append_whereclause(
- sql.literal_column("ora_rn") > offset_value)
+ sql.literal_column("ora_rn") > offset_clause)
offsetselect._for_update_arg = select._for_update_arg
select = offsetselect
'col2, ROWNUM AS ora_rn FROM (SELECT '
'sometable.col1 AS col1, sometable.col2 AS '
'col2 FROM sometable) WHERE ROWNUM <= '
- ':ROWNUM_1) WHERE ora_rn > :ora_rn_1')
+ ':param_1 + :param_2) WHERE ora_rn > :param_2',
+ checkparams={'param_1': 10, 'param_2': 20})
c = s.compile(dialect=oracle.OracleDialect())
assert t.c.col1 in set(c.result_map['col1'][1])
'FROM (SELECT col1, col2, ROWNUM AS ora_rn '
'FROM (SELECT sometable.col1 AS col1, '
'sometable.col2 AS col2 FROM sometable) '
- 'WHERE ROWNUM <= :ROWNUM_1) WHERE ora_rn > '
- ':ora_rn_1)')
+ 'WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > '
+ ':param_2)',
+ checkparams={'param_1': 10, 'param_2': 20})
self.assert_compile(s,
'SELECT col1, col2 FROM (SELECT col1, col2 '
'FROM (SELECT col1, col2, ROWNUM AS ora_rn '
'FROM (SELECT sometable.col1 AS col1, '
'sometable.col2 AS col2 FROM sometable) '
- 'WHERE ROWNUM <= :ROWNUM_1) WHERE ora_rn > '
- ':ora_rn_1)')
+ 'WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > '
+ ':param_2)')
s = select([t]).limit(10).offset(20).order_by(t.c.col2)
self.assert_compile(s,
'sometable.col1 AS col1, sometable.col2 AS '
'col2 FROM sometable ORDER BY '
'sometable.col2) WHERE ROWNUM <= '
- ':ROWNUM_1) WHERE ora_rn > :ora_rn_1')
+ ':param_1 + :param_2) WHERE ora_rn > :param_2',
+ checkparams={'param_1': 10, 'param_2': 20}
+ )
+
s = select([t], for_update=True).limit(10).order_by(t.c.col2)
self.assert_compile(s,
'SELECT col1, col2 FROM (SELECT '
'sometable.col1 AS col1, sometable.col2 AS '
'col2 FROM sometable ORDER BY '
- 'sometable.col2) WHERE ROWNUM <= :ROWNUM_1 '
+ 'sometable.col2) WHERE ROWNUM <= :param_1 '
'FOR UPDATE')
s = select([t],
'sometable.col1 AS col1, sometable.col2 AS '
'col2 FROM sometable ORDER BY '
'sometable.col2) WHERE ROWNUM <= '
- ':ROWNUM_1) WHERE ora_rn > :ora_rn_1 FOR '
+ ':param_1 + :param_2) WHERE ora_rn > :param_2 FOR '
'UPDATE')
def test_for_update(self):
self.assert_compile(select([t]).limit(10),
"SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, "
"sometable.col2 AS col2 FROM sometable) WHERE ROWNUM "
- "<= :ROWNUM_1",
+ "<= :param_1",
dialect=dialect)
self.assert_compile(select([t]).offset(10),
"SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
- "FROM sometable)) WHERE ora_rn > :ora_rn_1",
+ "FROM sometable)) WHERE ora_rn > :param_1",
dialect=dialect)
self.assert_compile(select([t]).limit(10).offset(10),
"SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
- "FROM sometable) WHERE ROWNUM <= :ROWNUM_1) WHERE ora_rn > "
- ":ora_rn_1",
- dialect=dialect)
+ "FROM sometable) WHERE ROWNUM <= :param_1 + :param_2) "
+ "WHERE ora_rn > :param_2",
+ dialect=dialect,
+ checkparams={'param_1': 10, 'param_2': 10})
def test_long_labels(self):
dialect = default.DefaultDialect()
'thirdtable.userid(+) = '
'myothertable.otherid AND mytable.myid = '
'myothertable.otherid ORDER BY '
- 'mytable.name) WHERE ROWNUM <= :ROWNUM_1) '
- 'WHERE ora_rn > :ora_rn_1',
+ 'mytable.name) WHERE ROWNUM <= :param_1 + :param_2) '
+ 'WHERE ora_rn > :param_2',
+ checkparams={'param_1': 10, 'param_2': 5},
dialect=oracle.dialect(use_ansi=False))
subq = select([table1]).select_from(table1.outerjoin(table2,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
