--- /dev/null
+.. change::
+ :tags: oracle, reflection
+ :tickets: 9597
+
+ Added reflection support in the Oracle dialect to expression based indexes
+ and the ordering direction of index expressions.
+
+.. change::
+ :tags: performance, schema
+ :tickets: 9597
+
+ Improved how table columns are added, avoiding unnecessary allocations,
+ significantly speeding up the creation of many table, like when reflecting
+ entire schemas.
else:
return value
+ remove_size = re.compile(r"\(\d+\)")
+
for row_dict in result:
table_name = self.normalize_name(row_dict["table_name"])
orig_colname = row_dict["column_name"]
elif "WITH LOCAL TIME ZONE" in coltype:
coltype = TIMESTAMP(local_timezone=True)
else:
- coltype = re.sub(r"\(\d+\)", "", coltype)
+ coltype = re.sub(remove_size, "", coltype)
try:
coltype = self.ischema_names[coltype]
except KeyError:
dictionary.all_indexes.c.uniqueness,
dictionary.all_indexes.c.compression,
dictionary.all_indexes.c.prefix_length,
+ dictionary.all_ind_columns.c.descend,
+ dictionary.all_ind_expressions.c.column_expression,
)
.select_from(dictionary.all_ind_columns)
.join(
sql.and_(
dictionary.all_ind_columns.c.index_name
== dictionary.all_indexes.c.index_name,
- dictionary.all_ind_columns.c.table_owner
- == dictionary.all_indexes.c.table_owner,
- # NOTE: this condition on table_name is not required
- # but it improves the query performance noticeably
- dictionary.all_ind_columns.c.table_name
- == dictionary.all_indexes.c.table_name,
+ dictionary.all_ind_columns.c.index_owner
+ == dictionary.all_indexes.c.owner,
+ ),
+ )
+ .outerjoin(
+ # NOTE: this adds about 20% to the query time. Using a
+ # case expression with a scalar subquery only when needed
+ # with the assumption that most indexes are not expression
+ # would be faster but oracle does not like that with
+ # LONG datatype. It errors with:
+ # ORA-00997: illegal use of LONG datatype
+ dictionary.all_ind_expressions,
+ sql.and_(
+ dictionary.all_ind_expressions.c.index_name
+ == dictionary.all_ind_columns.c.index_name,
+ dictionary.all_ind_expressions.c.index_owner
+ == dictionary.all_ind_columns.c.index_owner,
+ dictionary.all_ind_expressions.c.column_position
+ == dictionary.all_ind_columns.c.column_position,
),
)
.where(
- dictionary.all_ind_columns.c.table_owner == owner,
- dictionary.all_ind_columns.c.table_name.in_(
+ dictionary.all_indexes.c.table_owner == owner,
+ dictionary.all_indexes.c.table_name.in_(
bindparam("all_objects")
),
)
if row_dict["constraint_type"] == "P"
}
+ # all_ind_expressions.column_expression is LONG
result = self._run_batches(
connection,
query,
dblink,
- returns_long=False,
+ returns_long=True,
mappings=True,
all_objects=all_objects,
)
enabled = {"DISABLED": False, "ENABLED": True}
is_bitmap = {"BITMAP", "FUNCTION-BASED BITMAP"}
- oracle_sys_col = re.compile(r"SYS_NC\d+\$", re.IGNORECASE)
-
indexes = defaultdict(dict)
for row_dict in self._get_indexes_rows(
else:
index_dict = table_indexes[index_name]
- # filter out Oracle SYS_NC names. could also do an outer join
- # to the all_tab_columns table and check for real col names
- # there.
- if not oracle_sys_col.match(row_dict["column_name"]):
- index_dict["column_names"].append(
- self.normalize_name(row_dict["column_name"])
- )
+ expr = row_dict["column_expression"]
+ if expr is not None:
+ index_dict["column_names"].append(None)
+ if "expressions" in index_dict:
+ index_dict["expressions"].append(expr)
+ else:
+ index_dict["expressions"] = index_dict["column_names"][:-1]
+ index_dict["expressions"].append(expr)
+
+ if row_dict["descend"].lower() != "asc":
+ assert row_dict["descend"].lower() == "desc"
+ cs = index_dict.setdefault("column_sorting", {})
+ cs[expr] = ("desc",)
+ else:
+ assert row_dict["descend"].lower() == "asc"
+ cn = self.normalize_name(row_dict["column_name"])
+ index_dict["column_names"].append(cn)
+ if "expressions" in index_dict:
+ index_dict["expressions"].append(cn)
default = ReflectionDefaults.indexes
Column("auto", VARCHAR2(3)),
).alias("a_indexes")
+all_ind_expressions = Table(
+ "all_ind_expressions" + DB_LINK_PLACEHOLDER,
+ dictionary_meta,
+ Column("index_owner", VARCHAR2(128), nullable=False),
+ Column("index_name", VARCHAR2(128), nullable=False),
+ Column("table_owner", VARCHAR2(128), nullable=False),
+ Column("table_name", VARCHAR2(128), nullable=False),
+ Column("column_expression", LONG),
+ Column("column_position", NUMBER, nullable=False),
+).alias("a_ind_expressions")
+
all_constraints = Table(
"all_constraints" + DB_LINK_PLACEHOLDER,
dictionary_meta,
"""
column_sorting: NotRequired[Dict[str, Tuple[str]]]
- """optional dict mapping column names to tuple of sort keywords,
- which may include ``asc``, ``desc``, ``nulls_first``, ``nulls_last``.
+ """optional dict mapping column names or expressions to tuple of sort
+ keywords, which may include ``asc``, ``desc``, ``nulls_first``,
+ ``nulls_last``.
.. versionadded:: 1.3.5
"""
# in a _make_proxy operation
util.memoized_property.reset(named_column, "proxy_set")
else:
- l = len(self._collection)
- self._collection.append(
- (key, named_column, _ColumnMetrics(self, named_column))
- )
- self._colset.add(named_column._deannotate())
- self._index[l] = (key, named_column)
- self._index[key] = (key, named_column)
+ self._append_new_column(key, named_column)
+
+ def _append_new_column(self, key: str, named_column: _NAMEDCOL) -> None:
+ l = len(self._collection)
+ self._collection.append(
+ (key, named_column, _ColumnMetrics(self, named_column))
+ )
+ self._colset.add(named_column._deannotate())
+ self._index[l] = (key, named_column)
+ self._index[key] = (key, named_column)
def _populate_separate_keys(
self, iter_: Iterable[Tuple[str, _NAMEDCOL]]
if column.key in self._index:
remove_col.add(self._index[column.key][1])
+ if not remove_col:
+ self._append_new_column(column.key, column)
+ return
new_cols: List[Tuple[str, _NAMEDCOL, _ColumnMetrics[_NAMEDCOL]]] = []
replaced = False
for k, col, metrics in self._collection:
ASC/DESC."""
return exclusions.open()
+ @property
+ def reflect_indexes_with_ascdesc_as_expression(self):
+ """target database supports reflecting INDEX with per-column
+ ASC/DESC but reflects them as expressions (like oracle)."""
+ return exclusions.closed()
+
@property
def indexes_with_expressions(self):
"""target database supports CREATE INDEX against SQL expressions."""
):
fk_req = testing.requires.foreign_keys_reflect_as_index
dup_req = testing.requires.unique_constraints_reflect_as_index
+ sorting_expression = (
+ testing.requires.reflect_indexes_with_ascdesc_as_expression
+ )
+
if (fk and not fk_req.enabled) or (
duplicates and not dup_req.enabled
):
"include_columns": [],
}
if column_sorting:
- res["column_sorting"] = {"q": ("desc",)}
+ res["column_sorting"] = column_sorting
+ if sorting_expression.enabled:
+ res["expressions"] = orig = res["column_names"]
+ res["column_names"] = [
+ None if c in column_sorting else c for c in orig
+ ]
+
if duplicates:
res["duplicates_constraint"] = name
return [res]
insp.clear_cache()
eq_(insp.get_multi_table_comment(**kw), exp)
+ def _check_expressions(self, result, exp, err_msg):
+ def _clean(text: str):
+ return re.sub(r"['\" ]", "", text).lower()
+
+ if isinstance(exp, dict):
+ eq_({_clean(e): v for e, v in result.items()}, exp, err_msg)
+ else:
+ eq_([_clean(e) for e in result], exp, err_msg)
+
def _check_list(self, result, exp, req_keys=None, msg=None):
if req_keys is None:
eq_(result, exp, msg)
for r, e in zip(result, exp):
for k in set(r) | set(e):
if k in req_keys or (k in r and k in e):
- eq_(r[k], e[k], f"{msg} - {k} - {r}")
+ err_msg = f"{msg} - {k} - {r}"
+ if k in ("expressions", "column_sorting"):
+ self._check_expressions(r[k], e[k], err_msg)
+ else:
+ eq_(r[k], e[k], err_msg)
def _check_table_dict(self, result, exp, req_keys=None, make_lists=False):
eq_(set(result.keys()), set(exp.keys()))
class lower_index_str(str):
def __eq__(self, other):
+ ol = other.lower()
# test that lower and x or y are in the string
- return "lower" in other and ("x" in other or "y" in other)
+ return "lower" in ol and ("x" in ol or "y" in ol)
class coalesce_index_str(str):
def __eq__(self, other):
from sqlalchemy import Float
from sqlalchemy import ForeignKey
from sqlalchemy import ForeignKeyConstraint
-from sqlalchemy import func
from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import inspect
)
def test_reflect_fn_index(self, metadata, connection):
- """test reflection of a functional index.
+ """test reflection of a functional index."""
- it appears this emitted a warning at some point but does not right now.
- the returned data is not exactly correct, but this is what it's
- likely been doing for many years.
+ Table(
+ "sometable",
+ metadata,
+ Column("group", Unicode(255)),
+ Column("col", Unicode(255)),
+ Column("other", Unicode(255), index=True),
+ )
+ metadata.create_all(connection)
+ connection.exec_driver_sql(
+ """create index idx3 on sometable(
+ lower("group"), other, upper(other))"""
+ )
+ connection.exec_driver_sql(
+ """create index idx1 on sometable
+ (("group" || col), col || other desc)"""
+ )
+ connection.exec_driver_sql(
+ """
+ create unique index idx2 on sometable
+ (col desc, lower(other), "group" asc)
+ """
+ )
- """
+ expected = [
+ {
+ "name": "idx1",
+ "column_names": [None, None],
+ "expressions": ['"group"||"COL"', '"COL"||"OTHER"'],
+ "unique": False,
+ "dialect_options": {},
+ "column_sorting": {'"COL"||"OTHER"': ("desc",)},
+ },
+ {
+ "name": "idx2",
+ "column_names": [None, None, "group"],
+ "expressions": ['"COL"', 'LOWER("OTHER")', "group"],
+ "unique": True,
+ "column_sorting": {'"COL"': ("desc",)},
+ "dialect_options": {},
+ },
+ {
+ "name": "idx3",
+ "column_names": [None, "other", None],
+ "expressions": [
+ 'LOWER("group")',
+ "other",
+ 'UPPER("OTHER")',
+ ],
+ "unique": False,
+ "dialect_options": {},
+ },
+ {
+ "name": "ix_sometable_other",
+ "column_names": ["other"],
+ "unique": False,
+ "dialect_options": {},
+ },
+ ]
+
+ eq_(inspect(connection).get_indexes("sometable"), expected)
+ def test_indexes_asc_desc(self, metadata, connection):
s_table = Table(
"sometable",
metadata,
- Column("group", Unicode(255), primary_key=True),
+ Column("a", Unicode(255), primary_key=True),
+ Column("b", Unicode(255)),
+ Column("group", Unicode(255)),
Column("col", Unicode(255)),
)
-
- Index("data_idx", func.upper(s_table.c.col))
+ Index("id1", s_table.c.b.asc())
+ Index("id2", s_table.c.col.desc())
+ Index("id3", s_table.c.b.asc(), s_table.c.group.desc())
metadata.create_all(connection)
- eq_(
- inspect(connection).get_indexes("sometable"),
- [
- {
- "column_names": [],
- "dialect_options": {},
- "name": "data_idx",
- "unique": False,
- }
- ],
- )
+ expected = [
+ {
+ "name": "id1",
+ "column_names": ["b"],
+ "unique": False,
+ "dialect_options": {},
+ },
+ {
+ "name": "id2",
+ "column_names": [None],
+ "expressions": ['"COL"'],
+ "unique": False,
+ "column_sorting": {'"COL"': ("desc",)},
+ "dialect_options": {},
+ },
+ {
+ "name": "id3",
+ "column_names": ["b", None],
+ "expressions": ["b", '"group"'],
+ "unique": False,
+ "column_sorting": {'"group"': ("desc",)},
+ "dialect_options": {},
+ },
+ ]
+ eq_(inspect(connection).get_indexes("sometable"), expected)
def test_basic(self, metadata, connection):
connection.exec_driver_sql(
"""
create index idx3 on party
- (lower(name::text), other, lower(aname::text))
+ (lower(name::text), other, lower(aname::text) desc)
"""
)
connection.exec_driver_sql(
"unique": False,
"include_columns": [],
"dialect_options": {"postgresql_include": []},
+ "column_sorting": {"lower(aname::text)": ("desc",)},
},
{
"name": "idx4",
cols = []
for i in range(col_number - (0 if is_mssql else add_identity)):
args = []
- if random.random() < 0.95 or table_num == 0:
+ if random.random() < 0.99 or table_num == 0:
if is_mssql and add_identity and i == 0:
args.append(sa.Integer)
args.append(identity)
else:
args.append(random.choice(types))
else:
- args.append(
- sa.ForeignKey(f"table_{table_num-1}.table_{table_num-1}_col_1")
- )
+ target = random.randint(0, table_num - 1)
+ args.append(sa.ForeignKey(f"table_{target}.table_{target}_col_1"))
cols.append(
sa.Column(
f"table_{table_num}_col_{i+1}",
comment=f"primary key of table_{table_num}"
if i == 0
else None,
- index=random.random() > 0.9 and i > 0,
- unique=random.random() > 0.95 and i > 0,
+ index=random.random() > 0.97 and i > 0,
+ unique=random.random() > 0.97 and i > 0,
)
)
if add_identity and not is_mssql:
meta.create_all(engine, tables[i : i + 500])
+def _drop_ddl(name, schema_name, dialect_name):
+ if dialect_name.startswith("postgres"):
+ suffix = "CASCADE"
+ elif dialect_name.startswith("oracle"):
+ suffix = "CASCADE CONSTRAINTS PURGE"
+ else:
+ suffix = ""
+ if schema_name:
+ return sa.schema.DDL(f"DROP TABLE {schema_name}.{name} {suffix}")
+ else:
+ return sa.schema.DDL(f"DROP TABLE {name} {suffix}")
+
+
@log
def drop_tables(engine, meta, schema_name, table_names: list):
tables = list(meta.tables.values())[::-1]
meta.drop_all(engine, tables[i : i + 500])
remaining = sa.inspect(engine).get_table_names(schema=schema_name)
- suffix = ""
- if engine.dialect.name.startswith("postgres"):
- suffix = "CASCADE"
-
remaining = sorted(
remaining, key=lambda tn: int(tn.partition("_")[2]), reverse=True
)
name = engine.dialect.denormalize_name(tn)
else:
name = tn
- if schema_name:
- conn.execute(
- sa.schema.DDL(
- f'DROP TABLE {schema_name}."{name}" {suffix}'
- )
- )
- else:
- conn.execute(sa.schema.DDL(f'DROP TABLE "{name}" {suffix}'))
+ conn.execute(_drop_ddl(name, schema_name, engine.dialect.name))
if i % 500 == 0:
conn.commit()
conn.commit()
else:
engine = sa.create_engine(db, echo=args.echo, future=True)
+ if args.drop_all:
+ return drop_all(engine, schema_name)
+
if engine.name == "oracle":
# clear out oracle caches so that we get the real-world time the
# queries would normally take for scripts that aren't run repeatedly
return track_time
+def drop_all(engine, schema_name):
+ with engine.connect() as conn:
+ table_names = engine.dialect.get_table_names(conn, schema=schema_name)
+ print(f"Dropping {len(table_names)} tables")
+ dn = engine.dialect.name
+ i = 0
+ while table_names:
+ name = table_names.pop()
+ try:
+ conn.execute(_drop_ddl(name, schema_name, dn))
+ conn.commit()
+ except Exception:
+ conn.rollback()
+ table_names.insert(0, name)
+ i += 1
+ if i % 25 == 0:
+ print(f"Still running. Tables left {len(table_names)}")
+
+
if __name__ == "__main__":
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
parser.add_argument(
parser.add_argument(
"--no-drop", help="Do not run drop tables", action="store_true"
)
- parser.add_argument("--reflect", help="Run reflect", action="store_true")
+ parser.add_argument(
+ "--reflect", help="Run metadata reflect", action="store_true"
+ )
parser.add_argument(
"--test",
help="Run these tests. 'all' runs all tests",
"using single reflections. Mainly for sqlite.",
)
parser.add_argument("--pool-class", help="The pool class to use")
+ parser.add_argument(
+ "--drop-all",
+ action="store_true",
+ help="Drop all tables, do nothing else",
+ )
args = parser.parse_args()
min_cols = args.min_cols
and not self._mariadb_105(config)
)
- @property
- def reflect_indexes_with_ascdesc(self):
- return fails_if(["oracle"])
-
@property
def table_ddl_if_exists(self):
"""target platform supports IF NOT EXISTS / IF EXISTS for tables."""
@property
def indexes_with_expressions(self):
- return only_on(["postgresql", "sqlite>=3.9.0"])
+ return only_on(["postgresql", "sqlite>=3.9.0", "oracle"])
@property
def reflect_indexes_with_expressions(self):
- return only_on(["postgresql"])
+ return only_on(["postgresql", "oracle"])
+
+ @property
+ def reflect_indexes_with_ascdesc_as_expression(self):
+ return only_on(["oracle"])
@property
def temp_table_names(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
