rearchitect reflection for batched performance

Rearchitected the schema reflection API to allow some dialects to make use
of high performing batch queries to reflect the schemas of many tables at
once using much fewer queries. The new performance features are targeted
first at the PostgreSQL and Oracle backends, and may be applied to any
dialect that makes use of SELECT queries against system catalog tables to
reflect tables (currently this omits the MySQL and SQLite dialects which
instead make use of parsing the "CREATE TABLE" statement, however these
dialects do not have a pre-existing performance issue with reflection. MS
SQL Server is still a TODO).

The new API is backwards compatible with the previous system, and should
require no changes to third party dialects to retain compatibility;
third party dialects can also opt into the new system by implementing
batched queries for schema reflection.

Along with this change is an updated reflection API that is fully
:pep:`484` typed, features many new methods and some changes.

Fixes: #4379
Change-Id: I897ec09843543aa7012bcdce758792ed3d415d08

1 files changed, 552 insertions, 2 deletions

diff --git a/test/dialect/oracle/test_reflection.py b/test/dialect/oracle/test_reflection.py
index bf76dca43..53eb94df3 100644
--- a/test/dialect/oracle/test_reflection.py
+++ b/test/dialect/oracle/test_reflection.py
@@ -27,14 +27,18 @@ from sqlalchemy.dialects.oracle.base import BINARY_FLOAT
 from sqlalchemy.dialects.oracle.base import DOUBLE_PRECISION
 from sqlalchemy.dialects.oracle.base import NUMBER
 from sqlalchemy.dialects.oracle.base import REAL
+from sqlalchemy.engine import ObjectKind
 from sqlalchemy.testing import assert_warns
 from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing.engines import testing_engine
 from sqlalchemy.testing.schema import Column
+from sqlalchemy.testing.schema import eq_compile_type
 from sqlalchemy.testing.schema import Table
 
 
@@ -384,6 +388,7 @@ class SystemTableTablenamesTest(fixtures.TestBase):
     __backend__ = True
 
     def setup_test(self):
+
         with testing.db.begin() as conn:
             conn.exec_driver_sql("create table my_table (id integer)")
             conn.exec_driver_sql(
@@ -417,6 +422,14 @@ class SystemTableTablenamesTest(fixtures.TestBase):
             set(["my_table", "foo_table"]),
         )
 
+    def test_reflect_system_table(self):
+        meta = MetaData()
+        t = Table("foo_table", meta, autoload_with=testing.db)
+        assert t.columns.keys() == ["id"]
+
+        t = Table("my_temp_table", meta, autoload_with=testing.db)
+        assert t.columns.keys() == ["id"]
+
 
 class DontReflectIOTTest(fixtures.TestBase):
     """test that index overflow tables aren't included in
@@ -509,6 +522,228 @@ class TableReflectionTest(fixtures.TestBase):
         tbl = Table("test_compress", m2, autoload_with=connection)
         assert tbl.dialect_options["oracle"]["compress"] == "OLTP"
 
+    def test_reflect_hidden_column(self):
+        with testing.db.begin() as conn:
+            conn.exec_driver_sql(
+                "CREATE TABLE my_table(id integer, hide integer INVISIBLE)"
+            )
+
+            try:
+                insp = inspect(conn)
+                cols = insp.get_columns("my_table")
+                assert len(cols) == 1
+                assert cols[0]["name"] == "id"
+            finally:
+                conn.exec_driver_sql("DROP TABLE my_table")
+
+
+class ViewReflectionTest(fixtures.TestBase):
+    __only_on__ = "oracle"
+    __backend__ = True
+
+    @classmethod
+    def setup_test_class(cls):
+        sql = """
+    CREATE TABLE tbl (
+        id INTEGER PRIMARY KEY,
+        data INTEGER
+    );
+
+    CREATE VIEW tbl_plain_v AS
+        SELECT id, data FROM tbl WHERE id > 100;
+
+    -- comments on plain views are created with "comment on table"
+    -- because why not..
+    COMMENT ON TABLE tbl_plain_v IS 'view comment';
+
+    CREATE MATERIALIZED VIEW tbl_v AS
+        SELECT id, data FROM tbl WHERE id > 42;
+
+    COMMENT ON MATERIALIZED VIEW tbl_v IS 'my mat view comment';
+
+    CREATE MATERIALIZED VIEW tbl_v2 AS
+        SELECT id, data FROM tbl WHERE id < 42;
+
+    COMMENT ON MATERIALIZED VIEW tbl_v2 IS 'my other mat view comment';
+
+    CREATE SYNONYM view_syn FOR tbl_plain_v;
+    CREATE SYNONYM %(test_schema)s.ts_v_s FOR tbl_plain_v;
+
+    CREATE VIEW %(test_schema)s.schema_view AS
+        SELECT 1 AS value FROM dual;
+
+    COMMENT ON TABLE %(test_schema)s.schema_view IS 'schema view comment';
+    CREATE SYNONYM syn_schema_view FOR %(test_schema)s.schema_view;
+    """
+        if testing.requires.oracle_test_dblink.enabled:
+            cls.dblink = config.file_config.get(
+                "sqla_testing", "oracle_db_link"
+            )
+            sql += """
+    CREATE SYNONYM syn_link FOR tbl_plain_v@%(link)s;
+    """ % {
+                "link": cls.dblink
+            }
+        with testing.db.begin() as conn:
+            for stmt in (
+                sql % {"test_schema": testing.config.test_schema}
+            ).split(";"):
+                if stmt.strip():
+                    conn.exec_driver_sql(stmt)
+
+    @classmethod
+    def teardown_test_class(cls):
+        sql = """
+    DROP MATERIALIZED VIEW tbl_v;
+    DROP MATERIALIZED VIEW tbl_v2;
+    DROP VIEW tbl_plain_v;
+    DROP TABLE tbl;
+    DROP VIEW %(test_schema)s.schema_view;
+    DROP SYNONYM view_syn;
+    DROP SYNONYM %(test_schema)s.ts_v_s;
+    DROP SYNONYM syn_schema_view;
+    """
+        if testing.requires.oracle_test_dblink.enabled:
+            sql += """
+    DROP SYNONYM syn_link;
+    """
+        with testing.db.begin() as conn:
+            for stmt in (
+                sql % {"test_schema": testing.config.test_schema}
+            ).split(";"):
+                if stmt.strip():
+                    conn.exec_driver_sql(stmt)
+
+    def test_get_names(self, connection):
+        insp = inspect(connection)
+        eq_(insp.get_table_names(), ["tbl"])
+        eq_(insp.get_view_names(), ["tbl_plain_v"])
+        eq_(insp.get_materialized_view_names(), ["tbl_v", "tbl_v2"])
+        eq_(
+            insp.get_view_names(schema=testing.config.test_schema),
+            ["schema_view"],
+        )
+
+    def test_get_table_comment_on_view(self, connection):
+        insp = inspect(connection)
+        eq_(insp.get_table_comment("tbl_v"), {"text": "my mat view comment"})
+        eq_(insp.get_table_comment("tbl_plain_v"), {"text": "view comment"})
+
+    def test_get_multi_view_comment(self, connection):
+        insp = inspect(connection)
+        plain = {(None, "tbl_plain_v"): {"text": "view comment"}}
+        mat = {
+            (None, "tbl_v"): {"text": "my mat view comment"},
+            (None, "tbl_v2"): {"text": "my other mat view comment"},
+        }
+        eq_(insp.get_multi_table_comment(kind=ObjectKind.VIEW), plain)
+        eq_(
+            insp.get_multi_table_comment(kind=ObjectKind.MATERIALIZED_VIEW),
+            mat,
+        )
+        eq_(
+            insp.get_multi_table_comment(kind=ObjectKind.ANY_VIEW),
+            {**plain, **mat},
+        )
+        ts = testing.config.test_schema
+        eq_(
+            insp.get_multi_table_comment(kind=ObjectKind.ANY_VIEW, schema=ts),
+            {(ts, "schema_view"): {"text": "schema view comment"}},
+        )
+        eq_(insp.get_multi_table_comment(), {(None, "tbl"): {"text": None}})
+
+    def test_get_table_comment_synonym(self, connection):
+        insp = inspect(connection)
+        eq_(
+            insp.get_table_comment("view_syn", oracle_resolve_synonyms=True),
+            {"text": "view comment"},
+        )
+        eq_(
+            insp.get_table_comment(
+                "syn_schema_view", oracle_resolve_synonyms=True
+            ),
+            {"text": "schema view comment"},
+        )
+        eq_(
+            insp.get_table_comment(
+                "ts_v_s",
+                oracle_resolve_synonyms=True,
+                schema=testing.config.test_schema,
+            ),
+            {"text": "view comment"},
+        )
+
+    def test_get_multi_view_comment_synonym(self, connection):
+        insp = inspect(connection)
+        exp = {
+            (None, "view_syn"): {"text": "view comment"},
+            (None, "syn_schema_view"): {"text": "schema view comment"},
+        }
+        if testing.requires.oracle_test_dblink.enabled:
+            exp[(None, "syn_link")] = {"text": "view comment"}
+        eq_(
+            insp.get_multi_table_comment(
+                oracle_resolve_synonyms=True, kind=ObjectKind.ANY_VIEW
+            ),
+            exp,
+        )
+        ts = testing.config.test_schema
+        eq_(
+            insp.get_multi_table_comment(
+                oracle_resolve_synonyms=True,
+                schema=ts,
+                kind=ObjectKind.ANY_VIEW,
+            ),
+            {(ts, "ts_v_s"): {"text": "view comment"}},
+        )
+
+    def test_get_view_definition(self, connection):
+        insp = inspect(connection)
+        eq_(
+            insp.get_view_definition("tbl_plain_v"),
+            "SELECT id, data FROM tbl WHERE id > 100",
+        )
+        eq_(
+            insp.get_view_definition("tbl_v"),
+            "SELECT id, data FROM tbl WHERE id > 42",
+        )
+        with expect_raises(exc.NoSuchTableError):
+            eq_(insp.get_view_definition("view_syn"), None)
+        eq_(
+            insp.get_view_definition("view_syn", oracle_resolve_synonyms=True),
+            "SELECT id, data FROM tbl WHERE id > 100",
+        )
+        eq_(
+            insp.get_view_definition(
+                "syn_schema_view", oracle_resolve_synonyms=True
+            ),
+            "SELECT 1 AS value FROM dual",
+        )
+        eq_(
+            insp.get_view_definition(
+                "ts_v_s",
+                oracle_resolve_synonyms=True,
+                schema=testing.config.test_schema,
+            ),
+            "SELECT id, data FROM tbl WHERE id > 100",
+        )
+
+    @testing.requires.oracle_test_dblink
+    def test_get_view_definition_dblink(self, connection):
+        insp = inspect(connection)
+        eq_(
+            insp.get_view_definition("syn_link", oracle_resolve_synonyms=True),
+            "SELECT id, data FROM tbl WHERE id > 100",
+        )
+        eq_(
+            insp.get_view_definition("tbl_plain_v", dblink=self.dblink),
+            "SELECT id, data FROM tbl WHERE id > 100",
+        )
+        eq_(
+            insp.get_view_definition("tbl_v", dblink=self.dblink),
+            "SELECT id, data FROM tbl WHERE id > 42",
+        )
+
 
 class RoundTripIndexTest(fixtures.TestBase):
     __only_on__ = "oracle"
@@ -722,8 +957,6 @@ class DBLinkReflectionTest(fixtures.TestBase):
 
     @classmethod
     def setup_test_class(cls):
-        from sqlalchemy.testing import config
-
         cls.dblink = config.file_config.get("sqla_testing", "oracle_db_link")
 
         # note that the synonym here is still not totally functional
@@ -863,3 +1096,320 @@ class IdentityReflectionTest(fixtures.TablesTest):
                 exp = common.copy()
                 exp["order"] = True
                 eq_(col["identity"], exp)
+
+
+class AdditionalReflectionTests(fixtures.TestBase):
+    __only_on__ = "oracle"
+    __backend__ = True
+
+    @classmethod
+    def setup_test_class(cls):
+        # currently assuming full DBA privs for the user.
+        # don't really know how else to go here unless
+        # we connect as the other user.
+
+        sql = """
+CREATE TABLE %(schema)sparent(
+    id INTEGER,
+    data VARCHAR2(50),
+    CONSTRAINT parent_pk_%(schema_id)s PRIMARY KEY (id)
+);
+CREATE TABLE %(schema)smy_table(
+    id INTEGER,
+    name VARCHAR2(125),
+    related INTEGER,
+    data%(schema_id)s NUMBER NOT NULL,
+    CONSTRAINT my_table_pk_%(schema_id)s PRIMARY KEY (id),
+    CONSTRAINT my_table_fk_%(schema_id)s FOREIGN KEY(related)
+    REFERENCES %(schema)sparent(id),
+    CONSTRAINT my_table_check_%(schema_id)s CHECK (data%(schema_id)s > 42),
+    CONSTRAINT data_unique%(schema_id)s UNIQUE (data%(schema_id)s)
+);
+CREATE INDEX my_table_index_%(schema_id)s on %(schema)smy_table (id, name);
+COMMENT ON TABLE %(schema)smy_table IS 'my table comment %(schema_id)s';
+COMMENT ON COLUMN %(schema)smy_table.name IS
+'my table.name comment %(schema_id)s';
+"""
+
+        with testing.db.begin() as conn:
+            for schema in ("", testing.config.test_schema):
+                dd = {
+                    "schema": f"{schema}." if schema else "",
+                    "schema_id": "sch" if schema else "",
+                }
+                for stmt in (sql % dd).split(";"):
+                    if stmt.strip():
+                        conn.exec_driver_sql(stmt)
+
+    @classmethod
+    def teardown_test_class(cls):
+        sql = """
+drop table %(schema)smy_table;
+drop table %(schema)sparent;
+"""
+        with testing.db.begin() as conn:
+            for schema in ("", testing.config.test_schema):
+                dd = {"schema": f"{schema}." if schema else ""}
+                for stmt in (sql % dd).split(";"):
+                    if stmt.strip():
+                        try:
+                            conn.exec_driver_sql(stmt)
+                        except:
+                            pass
+
+    def setup_test(self):
+        self.dblink = config.file_config.get("sqla_testing", "oracle_db_link")
+        self.dblink2 = config.file_config.get(
+            "sqla_testing", "oracle_db_link2"
+        )
+        self.columns = {}
+        self.indexes = {}
+        self.primary_keys = {}
+        self.comments = {}
+        self.uniques = {}
+        self.checks = {}
+        self.foreign_keys = {}
+        self.options = {}
+        self.allDicts = [
+            self.columns,
+            self.indexes,
+            self.primary_keys,
+            self.comments,
+            self.uniques,
+            self.checks,
+            self.foreign_keys,
+            self.options,
+        ]
+        for schema in (None, testing.config.test_schema):
+            suffix = "sch" if schema else ""
+
+            self.columns[schema] = {
+                (schema, "my_table"): [
+                    {
+                        "name": "id",
+                        "nullable": False,
+                        "type": eq_compile_type("INTEGER"),
+                        "default": None,
+                        "comment": None,
+                    },
+                    {
+                        "name": "name",
+                        "nullable": True,
+                        "type": eq_compile_type("VARCHAR(125)"),
+                        "default": None,
+                        "comment": f"my table.name comment {suffix}",
+                    },
+                    {
+                        "name": "related",
+                        "nullable": True,
+                        "type": eq_compile_type("INTEGER"),
+                        "default": None,
+                        "comment": None,
+                    },
+                    {
+                        "name": f"data{suffix}",
+                        "nullable": False,
+                        "type": eq_compile_type("NUMBER"),
+                        "default": None,
+                        "comment": None,
+                    },
+                ],
+                (schema, "parent"): [
+                    {
+                        "name": "id",
+                        "nullable": False,
+                        "type": eq_compile_type("INTEGER"),
+                        "default": None,
+                        "comment": None,
+                    },
+                    {
+                        "name": "data",
+                        "nullable": True,
+                        "type": eq_compile_type("VARCHAR(50)"),
+                        "default": None,
+                        "comment": None,
+                    },
+                ],
+            }
+            self.indexes[schema] = {
+                (schema, "my_table"): [
+                    {
+                        "name": f"data_unique{suffix}",
+                        "column_names": [f"data{suffix}"],
+                        "dialect_options": {},
+                        "unique": True,
+                    },
+                    {
+                        "name": f"my_table_index_{suffix}",
+                        "column_names": ["id", "name"],
+                        "dialect_options": {},
+                        "unique": False,
+                    },
+                ],
+                (schema, "parent"): [],
+            }
+            self.primary_keys[schema] = {
+                (schema, "my_table"): {
+                    "name": f"my_table_pk_{suffix}",
+                    "constrained_columns": ["id"],
+                },
+                (schema, "parent"): {
+                    "name": f"parent_pk_{suffix}",
+                    "constrained_columns": ["id"],
+                },
+            }
+            self.comments[schema] = {
+                (schema, "my_table"): {"text": f"my table comment {suffix}"},
+                (schema, "parent"): {"text": None},
+            }
+            self.foreign_keys[schema] = {
+                (schema, "my_table"): [
+                    {
+                        "name": f"my_table_fk_{suffix}",
+                        "constrained_columns": ["related"],
+                        "referred_schema": schema,
+                        "referred_table": "parent",
+                        "referred_columns": ["id"],
+                        "options": {},
+                    }
+                ],
+                (schema, "parent"): [],
+            }
+            self.checks[schema] = {
+                (schema, "my_table"): [
+                    {
+                        "name": f"my_table_check_{suffix}",
+                        "sqltext": f"data{suffix} > 42",
+                    }
+                ],
+                (schema, "parent"): [],
+            }
+            self.uniques[schema] = {
+                (schema, "my_table"): [
+                    {
+                        "name": f"data_unique{suffix}",
+                        "column_names": [f"data{suffix}"],
+                        "duplicates_index": f"data_unique{suffix}",
+                    }
+                ],
+                (schema, "parent"): [],
+            }
+            self.options[schema] = {
+                (schema, "my_table"): {},
+                (schema, "parent"): {},
+            }
+
+    def test_tables(self, connection):
+        insp = inspect(connection)
+
+        eq_(sorted(insp.get_table_names()), ["my_table", "parent"])
+
+    def _check_reflection(self, conn, schema, res_schema=False, **kw):
+        if res_schema is False:
+            res_schema = schema
+        insp = inspect(conn)
+        eq_(
+            insp.get_multi_columns(schema=schema, **kw),
+            self.columns[res_schema],
+        )
+        eq_(
+            insp.get_multi_indexes(schema=schema, **kw),
+            self.indexes[res_schema],
+        )
+        eq_(
+            insp.get_multi_pk_constraint(schema=schema, **kw),
+            self.primary_keys[res_schema],
+        )
+        eq_(
+            insp.get_multi_table_comment(schema=schema, **kw),
+            self.comments[res_schema],
+        )
+        eq_(
+            insp.get_multi_foreign_keys(schema=schema, **kw),
+            self.foreign_keys[res_schema],
+        )
+        eq_(
+            insp.get_multi_check_constraints(schema=schema, **kw),
+            self.checks[res_schema],
+        )
+        eq_(
+            insp.get_multi_unique_constraints(schema=schema, **kw),
+            self.uniques[res_schema],
+        )
+        eq_(
+            insp.get_multi_table_options(schema=schema, **kw),
+            self.options[res_schema],
+        )
+
+    @testing.combinations(True, False, argnames="schema")
+    def test_schema_translate_map(self, connection, schema):
+        schema = testing.config.test_schema if schema else None
+        c = connection.execution_options(
+            schema_translate_map={
+                None: "foo",
+                testing.config.test_schema: "bar",
+            }
+        )
+        self._check_reflection(c, schema)
+
+    @testing.requires.oracle_test_dblink
+    def test_db_link(self, connection):
+        self._check_reflection(connection, schema=None, dblink=self.dblink)
+        self._check_reflection(
+            connection,
+            schema=testing.config.test_schema,
+            dblink=self.dblink,
+        )
+
+    def test_no_synonyms(self, connection):
+        # oracle_resolve_synonyms is ignored if there are no matching synonym
+        self._check_reflection(
+            connection, schema=None, oracle_resolve_synonyms=True
+        )
+        connection.exec_driver_sql("CREATE SYNONYM tmp FOR parent")
+        for dict_ in self.allDicts:
+            dict_["tmp"] = {(None, "parent"): dict_[None][(None, "parent")]}
+        try:
+            self._check_reflection(
+                connection,
+                schema=None,
+                res_schema="tmp",
+                oracle_resolve_synonyms=True,
+                filter_names=["parent"],
+            )
+        finally:
+            connection.exec_driver_sql("DROP SYNONYM tmp")
+
+    @testing.requires.oracle_test_dblink
+    @testing.requires.oracle_test_dblink2
+    def test_multi_dblink_synonyms(self, connection):
+        # oracle_resolve_synonyms handles multiple dblink at once
+        connection.exec_driver_sql(
+            f"CREATE SYNONYM s1 FOR my_table@{self.dblink}"
+        )
+        connection.exec_driver_sql(
+            f"CREATE SYNONYM s2 FOR {testing.config.test_schema}."
+            f"my_table@{self.dblink2}"
+        )
+        connection.exec_driver_sql("CREATE SYNONYM s3 FOR parent")
+        for dict_ in self.allDicts:
+            dict_["tmp"] = {
+                (None, "s1"): dict_[None][(None, "my_table")],
+                (None, "s2"): dict_[testing.config.test_schema][
+                    (testing.config.test_schema, "my_table")
+                ],
+                (None, "s3"): dict_[None][(None, "parent")],
+            }
+        fk = self.foreign_keys["tmp"][(None, "s1")][0]
+        fk["referred_table"] = "s3"
+        try:
+            self._check_reflection(
+                connection,
+                schema=None,
+                res_schema="tmp",
+                oracle_resolve_synonyms=True,
+            )
+        finally:
+            connection.exec_driver_sql("DROP SYNONYM s1")
+            connection.exec_driver_sql("DROP SYNONYM s2")
+            connection.exec_driver_sql("DROP SYNONYM s3")