--- /dev/null
+.. change::
+ :tags: usecase, mssql, postgresql
+ :tickets: 4966
+
+ Added support for inspection / reflection of partial indexes / filtered
+ indexes, i.e. those which use the ``mssql_where`` or ``postgresql_where``
+ parameters, with :class:`_schema.Index`. The entry is both part of the
+ dictionary returned by :meth:`.Inspector.get_indexes` as well as part of a
+ reflected :class:`_schema.Index` construct that was reflected. Pull
+ request courtesy Ramon Williams.
from ...engine import cursor as _cursor
from ...engine import default
from ...engine import reflection
+from ...sql import coercions
from ...sql import compiler
from ...sql import elements
from ...sql import expression
from ...sql import func
from ...sql import quoted_name
+from ...sql import roles
from ...sql import util as sql_util
from ...sql.type_api import to_instance
from ...types import BIGINT
whereclause = index.dialect_options["mssql"]["where"]
if whereclause is not None:
+ whereclause = coercions.expect(
+ roles.DDLExpressionRole, whereclause
+ )
+
where_compiled = self.sql_compiler.process(
whereclause, include_table=False, literal_binds=True
)
rp = connection.execution_options(future_result=True).execute(
sql.text(
- "select ind.index_id, ind.is_unique, ind.name "
+ "select ind.index_id, ind.is_unique, ind.name, "
+ "ind.filter_definition "
"from sys.indexes as ind join sys.tables as tab on "
"ind.object_id=tab.object_id "
"join sys.schemas as sch on sch.schema_id=tab.schema_id "
"unique": row["is_unique"] == 1,
"column_names": [],
}
+
+ if row["filter_definition"] is not None:
+ indexes[row["index_id"]].setdefault("dialect_options", {})[
+ "mssql_where"
+ ] = row["filter_definition"]
+
rp = connection.execution_options(future_result=True).execute(
sql.text(
"select ind_col.index_id, ind_col.object_id, col.name "
whereclause = index.dialect_options["postgresql"]["where"]
if whereclause is not None:
+ whereclause = coercions.expect(
+ roles.DDLExpressionRole, whereclause
+ )
+
where_compiled = self.sql_compiler.process(
whereclause, include_table=False, literal_binds=True
)
IDX_SQL = """
SELECT
i.relname as relname,
- ix.indisunique, ix.indexprs, ix.indpred,
+ ix.indisunique, ix.indexprs,
a.attname, a.attnum, c.conrelid, ix.indkey::varchar,
ix.indoption::varchar, i.reloptions, am.amname,
+ pg_get_expr(ix.indpred, ix.indrelid),
%s as indnkeyatts
FROM
pg_class t
idx_name,
unique,
expr,
- prd,
col,
col_num,
conrelid,
idx_option,
options,
amname,
+ filter_definition,
indnkeyatts,
) = row
sv_idx_name = idx_name
continue
- if prd and not idx_name == sv_idx_name:
- util.warn(
- "Predicate of partial index %s ignored during reflection"
- % idx_name
- )
- sv_idx_name = idx_name
-
has_idx = idx_name in indexes
index = indexes[idx_name]
if col is not None:
if amname and amname != "btree":
index["amname"] = amname
+ if filter_definition:
+ index["postgresql_where"] = filter_definition
+
result = []
for name, idx in indexes.items():
entry = {
entry.setdefault("dialect_options", {})[
"postgresql_using"
] = idx["amname"]
+ if "postgresql_where" in idx:
+ entry.setdefault("dialect_options", {})[
+ "postgresql_where"
+ ] = idx["postgresql_where"]
result.append(entry)
return result
"CREATE INDEX test_idx_data_1 ON test (data) WHERE data > 1",
)
+ idx = Index("test_idx_data_1", tbl.c.data, mssql_where="data > 1")
+ self.assert_compile(
+ schema.CreateIndex(idx),
+ "CREATE INDEX test_idx_data_1 ON test (data) WHERE data > 1",
+ )
+
def test_index_ordering(self):
metadata = MetaData()
tbl = Table(
from sqlalchemy.dialects.mssql import base
from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode
from sqlalchemy.dialects.mssql.information_schema import tables
+from sqlalchemy.schema import CreateIndex
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import ComparesTables
from sqlalchemy.testing import eq_
eq_(set(list(t2.indexes)[0].columns), set([t2.c["x col"], t2.c.y]))
+ @testing.provide_metadata
+ def test_indexes_with_filtered(self, connection):
+ metadata = self.metadata
+
+ t1 = Table(
+ "t",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("x", types.String(20)),
+ Column("y", types.Integer),
+ )
+ Index("idx_x", t1.c.x, mssql_where=t1.c.x == "test")
+ Index("idx_y", t1.c.y, mssql_where=t1.c.y >= 5)
+ metadata.create_all(connection)
+ ind = testing.db.dialect.get_indexes(connection, "t", None)
+
+ filtered_indexes = []
+ for ix in ind:
+ if "dialect_options" in ix:
+ filtered_indexes.append(ix["dialect_options"]["mssql_where"])
+
+ eq_(sorted(filtered_indexes), ["([x]='test')", "([y]>=(5))"])
+
+ t2 = Table("t", MetaData(), autoload_with=connection)
+ idx = list(sorted(t2.indexes, key=lambda idx: idx.name))[0]
+
+ self.assert_compile(
+ CreateIndex(idx), "CREATE INDEX idx_x ON t (x) WHERE ([x]='test')"
+ )
+
@testing.provide_metadata
def test_max_ident_in_varchar_not_present(self):
"""test [ticket:3504].
dialect=postgresql.dialect(),
)
+ idx3 = Index(
+ "test_idx2",
+ tbl.c.data,
+ postgresql_where="data > 'a' AND data < 'b''s'",
+ )
+ self.assert_compile(
+ schema.CreateIndex(idx3),
+ "CREATE INDEX test_idx2 ON testtbl (data) "
+ "WHERE data > 'a' AND data < 'b''s'",
+ dialect=postgresql.dialect(),
+ )
+
def test_create_index_with_ops(self):
+
m = MetaData()
tbl = Table(
"testtbl",
from sqlalchemy.dialects.postgresql import INTEGER
from sqlalchemy.dialects.postgresql import INTERVAL
from sqlalchemy.dialects.postgresql import TSRANGE
+from sqlalchemy.schema import CreateIndex
from sqlalchemy.sql.schema import CheckConstraint
+from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import assert_raises
base.PGDialect.ischema_names = ischema_names
-class ReflectionTest(fixtures.TestBase):
+class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
__only_on__ = "postgresql"
__backend__ = True
@testing.provide_metadata
def test_index_reflection(self):
- """ Reflecting partial & expression-based indexes should warn
+ """ Reflecting expression-based indexes should warn
"""
metadata = self.metadata
[
"Skipped unsupported reflection of "
"expression-based index idx1",
- "Predicate of partial index idx2 ignored during " "reflection",
"Skipped unsupported reflection of "
"expression-based index idx3",
],
)
+ @testing.provide_metadata
+ def test_index_reflection_partial(self, connection):
+ """Reflect the filter defintion on partial indexes
+ """
+
+ metadata = self.metadata
+
+ t1 = Table(
+ "table1",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(20)),
+ Column("x", Integer),
+ )
+ Index("idx1", t1.c.id, postgresql_where=t1.c.name == "test")
+ Index("idx2", t1.c.id, postgresql_where=t1.c.x >= 5)
+
+ metadata.create_all(connection)
+
+ ind = testing.db.dialect.get_indexes(connection, t1, None)
+
+ partial_definitions = []
+ for ix in ind:
+ if "dialect_options" in ix:
+ partial_definitions.append(
+ ix["dialect_options"]["postgresql_where"]
+ )
+
+ eq_(
+ sorted(partial_definitions),
+ ["((name)::text = 'test'::text)", "(x >= 5)"],
+ )
+
+ t2 = Table("table1", MetaData(), autoload_with=connection)
+ idx = list(sorted(t2.indexes, key=lambda idx: idx.name))[0]
+
+ self.assert_compile(
+ CreateIndex(idx),
+ "CREATE INDEX idx1 ON table1 (id) "
+ "WHERE ((name)::text = 'test'::text)",
+ )
+
@testing.fails_if("postgresql < 8.3", "index ordering not supported")
@testing.provide_metadata
def test_index_reflection_with_sorting(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
