.. changelog::
:version: 0.8.5
+ .. change::
+ :tags: postgresql, bug
+ :versions: 0.9.3
+
+ Support has been improved for Postgresql reflection behavior on very old
+ (pre 8.1) versions of Postgresql, and potentially other PG engines
+ such as Redshift (assuming Redshift reports the version as < 8.1).
+ The query for "indexes" as well as "primary keys" relies upon inspecting
+ a so-called "int2vector" datatype, which refuses to coerce to an array
+ prior to 8.1 causing failures regarding the "ANY()" operator used
+ in the query. Extensive googling has located the very hacky, but
+ recommended-by-PG-core-developer query to use when PG version < 8.1
+ is in use, so index and primary key constraint reflection now work
+ on these versions.
+
.. change::
:tags: feature, mysql
:versions: 0.9.3
table_oid = self.get_table_oid(connection, table_name, schema,
info_cache=kw.get('info_cache'))
- if self.server_version_info < (8, 0):
- # the shortcoming of this query is that it will
- # not detect a PK constraint that has been renamed.
- # This query was removed with #2291, however it was reported
- # that the newer queries do not work with PG 7 so here
- # it is restored when old PG versions are detected.
- PK_SQL = """
- SELECT attname FROM pg_attribute
- WHERE attrelid = (
- SELECT indexrelid FROM pg_index i
- WHERE i.indrelid = :table_oid
- AND i.indisprimary = 't')
- ORDER BY attnum
- """
- elif self.server_version_info < (8, 4):
- # unnest() and generate_subscripts() both introduced in
- # version 8.4
+ if self.server_version_info < (8, 4):
PK_SQL = """
SELECT a.attname
FROM
pg_class t
join pg_index ix on t.oid = ix.indrelid
join pg_attribute a
- on t.oid=a.attrelid and a.attnum=ANY(ix.indkey)
+ on t.oid=a.attrelid AND %s
WHERE
t.oid = :table_oid and ix.indisprimary = 't'
ORDER BY a.attnum
- """
+ """ % self._pg_index_any("a.attnum", "ix.indkey")
+
else:
+ # unnest() and generate_subscripts() both introduced in
+ # version 8.4
PK_SQL = """
SELECT a.attname
FROM pg_attribute a JOIN (
fkeys.append(fkey_d)
return fkeys
+ def _pg_index_any(self, col, compare_to):
+ if self.server_version_info < (8, 1):
+ # http://www.postgresql.org/message-id/10279.1124395722@sss.pgh.pa.us
+ # "In CVS tip you could replace this with "attnum = ANY (indkey)".
+ # Unfortunately, most array support doesn't work on int2vector in
+ # pre-8.1 releases, so I think you're kinda stuck with the above
+ # for now.
+ # regards, tom lane"
+ return "(%s)" % " OR ".join(
+ "%s[%d] = %s" % (compare_to, ind, col)
+ for ind in range(0, 10)
+ )
+ else:
+ return "%s = ANY(%s)" % (col, compare_to)
+
@reflection.cache
def get_indexes(self, connection, table_name, schema, **kw):
table_oid = self.get_table_oid(connection, table_name, schema,
SELECT
i.relname as relname,
ix.indisunique, ix.indexprs, ix.indpred,
- a.attname, a.attnum, ix.indkey::varchar
+ a.attname, a.attnum, ix.indkey%s
FROM
pg_class t
join pg_index ix on t.oid = ix.indrelid
join pg_class i on i.oid=ix.indexrelid
left outer join
pg_attribute a
- on t.oid=a.attrelid and a.attnum=ANY(ix.indkey)
+ on t.oid=a.attrelid and %s
WHERE
t.relkind = 'r'
and t.oid = :table_oid
ORDER BY
t.relname,
i.relname
- """
+ """ % (
+ "::varchar" if self.server_version_info >= (8, 1) else "",
+ self._pg_index_any("a.attnum", "ix.indkey")
+ )
t = sql.text(IDX_SQL, typemap={'attname': sqltypes.Unicode})
c = connection.execute(t, table_oid=table_oid)
: Numeric})
assert_raises(exc.InvalidRequestError, testing.db.execute, stmt)
+ @testing.only_if("postgresql >= 8.2", "requires standard_conforming_strings")
def test_serial_integer(self):
for type_, expected in [
(Integer, 'SERIAL'),
metadata.create_all()
self._assert_data_with_sequence(table, 'my_seq')
+ @testing.requires.returning
def test_sequence_returning_insert(self):
table = Table('testtable', metadata, Column('id', Integer,
Sequence('my_seq'), primary_key=True),
metadata.create_all()
self._assert_data_autoincrement(table)
+ @testing.requires.returning
def test_opt_sequence_returning_insert(self):
table = Table('testtable', metadata, Column('id', Integer,
Sequence('my_seq', optional=True),
metadata.create_all()
self._assert_data_autoincrement(table)
+ @testing.requires.returning
def test_autoincrement_returning_insert(self):
table = Table('testtable', metadata, Column('id', Integer,
primary_key=True), Column('data', String(30)))
class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
- __only_on__ = 'postgresql'
- __excluded_on__ = ('postgresql', '<', (8, 3, 0)),
+ __only_on__ = 'postgresql >= 8.3'
@classmethod
def setup_class(cls):
"""Test PostgreSQL domains"""
- __only_on__ = 'postgresql'
+ __only_on__ = 'postgresql > 8.2'
@classmethod
def setup_class(cls):
class ReflectionTest(fixtures.TestBase):
__only_on__ = 'postgresql'
- @testing.fails_if(('postgresql', '<', (8, 4)),
- "newer query is bypassed due to unsupported SQL functions")
+ @testing.fails_if("postgresql < 8.4",
+ "Better int2vector functions not available")
@testing.provide_metadata
def test_reflected_primary_key_order(self):
meta1 = self.metadata
eq_(
t.c.x.server_default.arg.text, "'%s'::character varying" % ("abcd" * 40)
)
+
+ @testing.fails_if("postgresql < 8.1", "schema name leaks in, not sure")
@testing.provide_metadata
def test_renamed_sequence_reflection(self):
metadata = self.metadata
class EnumTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
- __only_on__ = 'postgresql'
+ __only_on__ = 'postgresql > 8.3'
__dialect__ = postgresql.dialect()
def test_compile(self):
class UUIDTest(fixtures.TestBase):
"""Test the bind/return values of the UUID type."""
- __only_on__ = 'postgresql'
+ __only_on__ = 'postgresql >= 8.3'
@testing.requires.python25
@testing.fails_on('postgresql+zxjdbc',
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
