Merge "Emit v2.0 deprecation warning for "implicit autocommit""

19 files changed, 525 insertions, 112 deletions

diff --git a/doc/build/changelog/migration_14.rst b/doc/build/changelog/migration_14.rst
index e25d1e8bf..feb6617a8 100644
--- a/doc/build/changelog/migration_14.rst
+++ b/doc/build/changelog/migration_14.rst
@@ -409,6 +409,124 @@ is established as the implementation.
 :ticket:`1390`
 
 
+.. _deprecation_20_mode:
+
+SQLAlchemy 2.0 Deprecations Mode
+---------------------------------
+
+One of the primary goals of the 1.4 release is to provide a "transitional"
+release so that applications may migrate to SQLAlchemy 2.0 gradually.   Towards
+this end, a primary feature in release 1.4 is "2.0 deprecations mode", which is
+a series of deprecation warnings that emit against every detectable API pattern
+which will work differently in version 2.0.   The warnings all make use of the
+:class:`_exc.RemovedIn20Warning` class. As these warnings affect foundational
+patterns including the :func:`_sql.select` and :class:`_engine.Engine` constructs, even
+simple applications can generate a lot of warnings until appropriate API
+changes are made.   The warning mode is therefore turned off by default until
+the developer enables the environment variable ``SQLALCHEMY_WARN_20=1``.
+
+Given the example program below::
+
+  from sqlalchemy import column
+  from sqlalchemy import create_engine
+  from sqlalchemy import select
+  from sqlalchemy import table
+
+
+  engine = create_engine("sqlite://")
+
+  engine.execute("CREATE TABLE foo (id integer)")
+  engine.execute("INSERT INTO foo (id) VALUES (1)")
+
+
+  foo = table("foo", column("id"))
+  result = engine.execute(select([foo.c.id]))
+
+  print(result.fetchall())
+
+The above program uses several patterns that many users will already identify
+as "legacy", namely the use of the :meth:`_engine.Engine.execute` method
+that's part of the :ref:`connectionlesss execution <dbengine_implicit>`
+system.  When we run the above program against 1.4, it returns a single line::
+
+  $ python test3.py
+  [(1,)]
+
+To enable "2.0 deprecations mode", we enable the ``SQLALCHEMY_WARN_20=1``
+variable::
+
+    SQLALCHEMY_WARN_20=1 python test3.py
+
+**IMPORTANT** - older versions of Python may not emit deprecation warnings
+by default.   To guarantee deprecation warnings, use a `warnings filter`_
+that ensures warnings are printed::
+
+    SQLALCHEMY_WARN_20=1 python -W always::DeprecationWarning test3.py
+
+.. _warnings filter: https://docs.python.org/3/library/warnings.html#the-warnings-filter
+
+With warnings turned on, our program now has a lot to say::
+
+  $ SQLALCHEMY_WARN_20=1 python2 -W always::DeprecationWarning test3.py
+  test3.py:9: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    engine.execute("CREATE TABLE foo (id integer)")
+  /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    return connection.execute(statement, *multiparams, **params)
+  /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    self._commit_impl(autocommit=True)
+  test3.py:10: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    engine.execute("INSERT INTO foo (id) VALUES (1)")
+  /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0.  Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    return connection.execute(statement, *multiparams, **params)
+  /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0.   Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    self._commit_impl(autocommit=True)
+  /home/classic/dev/sqlalchemy/lib/sqlalchemy/sql/selectable.py:4271: RemovedIn20Warning: The legacy calling style of select() is deprecated and will be removed in SQLAlchemy 2.0.  Please use the new calling style described at select(). (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    return cls.create_legacy_select(*args, **kw)
+  test3.py:14: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+    result = engine.execute(select([foo.c.id]))
+  [(1,)]
+
+With the above guidance, we can migrate our program to use 2.0 styles, and
+as a bonus our program is much clearer::
+
+  from sqlalchemy import column
+  from sqlalchemy import create_engine
+  from sqlalchemy import select
+  from sqlalchemy import table
+  from sqlalchemy import text
+
+
+  engine = create_engine("sqlite://")
+
+  # don't rely on autocommit for DML and DDL
+  with engine.begin() as connection:
+      # use connection.execute(), not engine.execute()
+      # use the text() construct to execute textual SQL
+      connection.execute(text("CREATE TABLE foo (id integer)"))
+      connection.execute(text("INSERT INTO foo (id) VALUES (1)"))
+
+
+  foo = table("foo", column("id"))
+
+  with engine.connect() as connection:
+      # use connection.execute(), not engine.execute()
+      # select() now accepts column / table expressions positionally
+      result = connection.execute(select(foo.c.id))
+
+  print(result.fetchall())
+
+
+The goal of "2.0 deprecations mode" is that a program which runs with no
+:class:`_exc.RemovedIn20Warning` warnings with "2.0 deprecations mode" turned
+on is then ready to run in SQLAlchemy 2.0.
+
+
+.. seealso::
+
+  :ref:`migration_20_toplevel`
+
+
+
 API and Behavioral Changes - Core
 ==================================
 
diff --git a/doc/build/changelog/migration_20.rst b/doc/build/changelog/migration_20.rst
index 7b3d23c8c..7bb35d422 100644
--- a/doc/build/changelog/migration_20.rst
+++ b/doc/build/changelog/migration_20.rst
@@ -77,15 +77,18 @@ The steps to achieve this are as follows:
   well as providing for the initial real-world adoption of the new
   architectures.
 
-* A new deprecation class :class:`.exc.RemovedIn20Warning` is added, which
-  subclasses :class:`.exc.SADeprecationWarning`.   Applications and their test
+* A new deprecation class :class:`_exc.RemovedIn20Warning` is added, which
+  subclasses :class:`_exc.SADeprecationWarning`.   Applications and their test
   suites can opt to enable or disable reporting of the
-  :class:`.exc.RemovedIn20Warning` warning as needed.   To some extent, the
-  :class:`.exc.RemovedIn20Warning` deprecation class is analogous to the ``-3``
+  :class:`_exc.RemovedIn20Warning` warning as needed, by setting the
+  environment variable ``SQLALCHEMY_WARN_20=1`` **before** the program
+  runs.   To some extent, the
+  :class:`_exc.RemovedIn20Warning` deprecation class is analogous to the ``-3``
   flag available on Python 2 which reports on future Python 3
-  incompatibilities.
+  incompatibilities.   See :ref:`deprecation_20_mode` for background
+  on turning this on.
 
-* APIs which emit :class:`.exc.RemovedIn20Warning` should always feature a new
+* APIs which emit :class:`_exc.RemovedIn20Warning` should always feature a new
   1.4-compatible usage pattern that applications can migrate towards.  This
   pattern will then be fully compatible with SQLAlchemy 2.0.   In this way,
   an application can gradually adjust all of its 1.4-style code to work fully
diff --git a/doc/build/changelog/unreleased_14/4846.rst b/doc/build/changelog/unreleased_14/4846.rst
new file mode 100644
index 000000000..b534f02d2
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/4846.rst
@@ -0,0 +1,19 @@
+.. change::
+    :tags: engine
+    :tickets: 4846
+
+    "Implicit autocommit", which is the COMMIT that occurs when a DML or DDL
+    statement is emitted on a connection, is deprecated and won't be part of
+    SQLAlchemy 2.0.   A 2.0-style warning is emitted when autocommit takes
+    effect, so that the calling code may be adjusted to use an explicit
+    transaction.
+
+    As part of this change, DDL methods such as
+    :meth:`_schema.MetaData.create_all` when used against an
+    :class:`_engine.Engine` will run the operation in a BEGIN block if one is
+    not started already.
+
+    .. seealso::
+
+        :ref:`deprecation_20_mode`
+
diff --git a/doc/build/changelog/unreleased_14/mysql_has_table.rst b/doc/build/changelog/unreleased_14/mysql_has_table.rst
new file mode 100644
index 000000000..09faa6332
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/mysql_has_table.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, mysql
+
+    The MySQL and MariaDB dialects now query from the information_schema.tables
+    system view in order to determine if a particular table exists or not.
+    Previously, the "DESCRIBE" command was used with an exception catch to
+    detect non-existent,  which would have the undesirable effect of emitting a
+    ROLLBACK on the connection. There appeared to be legacy encoding issues
+    which prevented the use of "SHOW TABLES", for this, but as MySQL support is
+    now at 5.0.2  or above due to :ticket:`4189`, the information_schema tables
+    are now available in all cases.
+
diff --git a/doc/build/core/tutorial.rst b/doc/build/core/tutorial.rst
index 8d27dd21d..738d4d74e 100644
--- a/doc/build/core/tutorial.rst
+++ b/doc/build/core/tutorial.rst
@@ -150,15 +150,17 @@ each table first before creating, so it's safe to call multiple times:
 .. sourcecode:: pycon+sql
 
     {sql}>>> metadata.create_all(engine)
-    PRAGMA...
+    BEGIN...
     CREATE TABLE users (
         id INTEGER NOT NULL,
         name VARCHAR,
         fullname VARCHAR,
         PRIMARY KEY (id)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
-    COMMIT
+    <BLANKLINE>
     CREATE TABLE addresses (
         id INTEGER NOT NULL,
         user_id INTEGER,
@@ -166,6 +168,8 @@ each table first before creating, so it's safe to call multiple times:
         PRIMARY KEY (id),
         FOREIGN KEY(user_id) REFERENCES users (id)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
     COMMIT
 
diff --git a/doc/build/errors.rst b/doc/build/errors.rst
index ab9e018d9..61f009fad 100644
--- a/doc/build/errors.rst
+++ b/doc/build/errors.rst
@@ -104,13 +104,8 @@ a comprehensive future compatibility system that is to be integrated into the
 unambiguous, and incremental upgrade path in order to migrate applications to
 being fully 2.0 compatible.   The :class:`.exc.RemovedIn20Warning` deprecation
 warning is at the base of this system to provide guidance on what behaviors in
-an existing codebase will need to be modified.
-
-For some occurrences of this warning, an additional recommendation to use an
-API in either the ``sqlalchemy.future`` or  ``sqlalchemy.future.orm`` packages
-may be present.  This refers to two  special future-compatibility packages that
-are part of SQLAlchemy 1.4 and  are there to help migrate an application to the
-2.0 version.
+an existing codebase will need to be modified.  An overview of how to enable
+this warning is at :ref:`deprecation_20_mode`.
 
 .. seealso::
 
@@ -118,6 +113,10 @@ are part of SQLAlchemy 1.4 and  are there to help migrate an application to the
     the 1.x series, as well as the current goals and progress of SQLAlchemy
     2.0.
 
+
+    :ref:`deprecation_20_mode` - specific guidelines on how to use
+    "2.0 deprecations mode" in SQLAlchemy 1.4.
+
 .. _error_c9bf:
 
 A bind was located via legacy bound metadata, but since future=True is set on this Session, this bind is ignored.
diff --git a/doc/build/orm/tutorial.rst b/doc/build/orm/tutorial.rst
index dbad10b6f..8c148ac32 100644
--- a/doc/build/orm/tutorial.rst
+++ b/doc/build/orm/tutorial.rst
@@ -209,16 +209,16 @@ the actual ``CREATE TABLE`` statement:
 .. sourcecode:: python+sql
 
     >>> Base.metadata.create_all(engine)
-    PRAGMA main.table_info("users")
-    [...] ()
-    PRAGMA temp.table_info("users")
-    [...] ()
+    BEGIN...
     CREATE TABLE users (
-        id INTEGER NOT NULL, name VARCHAR,
+        id INTEGER NOT NULL,
+        name VARCHAR,
         fullname VARCHAR,
         nickname VARCHAR,
         PRIMARY KEY (id)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
     COMMIT
 
@@ -1215,14 +1215,16 @@ already been created:
 .. sourcecode:: python+sql
 
     {sql}>>> Base.metadata.create_all(engine)
-    PRAGMA...
+    BEGIN...
     CREATE TABLE addresses (
         id INTEGER NOT NULL,
         email_address VARCHAR NOT NULL,
         user_id INTEGER,
         PRIMARY KEY (id),
-         FOREIGN KEY(user_id) REFERENCES users (id)
+        FOREIGN KEY(user_id) REFERENCES users (id)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
     COMMIT
 
@@ -2080,15 +2082,17 @@ Create new tables:
 .. sourcecode:: python+sql
 
     {sql}>>> Base.metadata.create_all(engine)
-    PRAGMA...
+    BEGIN...
     CREATE TABLE keywords (
         id INTEGER NOT NULL,
         keyword VARCHAR(50) NOT NULL,
         PRIMARY KEY (id),
         UNIQUE (keyword)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
-    COMMIT
+    <BLANKLINE>
     CREATE TABLE posts (
         id INTEGER NOT NULL,
         user_id INTEGER,
@@ -2097,8 +2101,10 @@ Create new tables:
         PRIMARY KEY (id),
         FOREIGN KEY(user_id) REFERENCES users (id)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
-    COMMIT
+    <BLANKLINE>
     CREATE TABLE post_keywords (
         post_id INTEGER NOT NULL,
         keyword_id INTEGER NOT NULL,
@@ -2106,6 +2112,8 @@ Create new tables:
         FOREIGN KEY(post_id) REFERENCES posts (id),
         FOREIGN KEY(keyword_id) REFERENCES keywords (id)
     )
+    <BLANKLINE>
+    <BLANKLINE>
     [...] ()
     COMMIT
 
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 0c9859e79..1003eeca6 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -887,6 +887,7 @@ from collections import defaultdict
 import re
 
 from sqlalchemy import literal_column
+from sqlalchemy import text
 from sqlalchemy.sql import visitors
 from . import reflection as _reflection
 from .enumerated import ENUM
@@ -938,6 +939,7 @@ from ...sql import compiler
 from ...sql import elements
 from ...sql import roles
 from ...sql import util as sql_util
+from ...sql.sqltypes import Unicode
 from ...types import BINARY
 from ...types import BLOB
 from ...types import BOOLEAN
@@ -2708,39 +2710,24 @@ class MySQLDialect(default.DefaultDialect):
         return connection.exec_driver_sql("SELECT DATABASE()").scalar()
 
     def has_table(self, connection, table_name, schema=None):
-        # SHOW TABLE STATUS LIKE and SHOW TABLES LIKE do not function properly
-        # on macosx (and maybe win?) with multibyte table names.
-        #
-        # TODO: if this is not a problem on win, make the strategy swappable
-        # based on platform.  DESCRIBE is slower.
-
-        # [ticket:726]
-        # full_name = self.identifier_preparer.format_table(table,
-        #                                                   use_schema=True)
+        if schema is None:
+            schema = self.default_schema_name
 
-        full_name = ".".join(
-            self.identifier_preparer._quote_free_identifiers(
-                schema, table_name
-            )
+        rs = connection.execute(
+            text(
+                "SELECT * FROM information_schema.tables WHERE "
+                "table_schema = :table_schema AND "
+                "table_name = :table_name"
+            ).bindparams(
+                sql.bindparam("table_schema", type_=Unicode),
+                sql.bindparam("table_name", type_=Unicode),
+            ),
+            {
+                "table_schema": util.text_type(schema),
+                "table_name": util.text_type(table_name),
+            },
         )
-
-        st = "DESCRIBE %s" % full_name
-        rs = None
-        try:
-            try:
-                rs = connection.execution_options(
-                    skip_user_error_events=True
-                ).exec_driver_sql(st)
-                have = rs.fetchone() is not None
-                rs.close()
-                return have
-            except exc.DBAPIError as e:
-                if self._extract_error_code(e.orig) == 1146:
-                    return False
-                raise
-        finally:
-            if rs:
-                rs.close()
+        return bool(rs.scalar())
 
     def has_sequence(self, connection, sequence_name, schema=None):
         if not self.supports_sequences:
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 91fff4549..afab8e7b4 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -829,6 +829,15 @@ class Connection(Connectable):
     def _commit_impl(self, autocommit=False):
         assert not self.__branch_from
 
+        if autocommit:
+            util.warn_deprecated_20(
+                "The current statement is being autocommitted using "
+                "implicit autocommit, which will be removed in "
+                "SQLAlchemy 2.0. "
+                "Use the .begin() method of Engine or Connection in order to "
+                "use an explicit transaction for DML and DDL statements."
+            )
+
         if self._has_events or self.engine._has_events:
             self.dispatch.commit(self)
 
@@ -2814,7 +2823,7 @@ class Engine(Connectable, log.Identified):
             return conn.run_callable(callable_, *args, **kwargs)
 
     def _run_ddl_visitor(self, visitorcallable, element, **kwargs):
-        with self.connect() as conn:
+        with self.begin() as conn:
             conn._run_ddl_visitor(visitorcallable, element, **kwargs)
 
     @util.deprecated_20(
diff --git a/lib/sqlalchemy/exc.py b/lib/sqlalchemy/exc.py
index 491dde7b2..a17bb5cec 100644
--- a/lib/sqlalchemy/exc.py
+++ b/lib/sqlalchemy/exc.py
@@ -636,6 +636,8 @@ class RemovedIn20Warning(SADeprecationWarning):
 
         :ref:`error_b8d9`.
 
+        :ref:`deprecation_20_mode`
+
     """
 
     deprecated_since = "1.4"
diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index 9b1164874..45cc7ea2a 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -55,6 +55,7 @@ from .exclusions import only_if  # noqa
 from .exclusions import only_on  # noqa
 from .exclusions import skip  # noqa
 from .exclusions import skip_if  # noqa
+from .schema import eq_type_affinity  # noqa
 from .util import adict  # noqa
 from .util import fail  # noqa
 from .util import flag_combinations  # noqa
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index f78ebf496..67ef38a25 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -32,7 +32,7 @@ from ..util import decorator
 def expect_warnings(*messages, **kw):
     """Context manager which expects one or more warnings.
 
-    With no arguments, squelches all SAWarnings emitted via
+    With no arguments, squelches all SAWarning and RemovedIn20Warning emitted via
     sqlalchemy.util.warn and sqlalchemy.util.warn_limited.   Otherwise
     pass string expressions that will match selected warnings via regex;
     all non-matching warnings are sent through.
@@ -41,8 +41,10 @@ def expect_warnings(*messages, **kw):
 
     Note that the test suite sets SAWarning warnings to raise exceptions.
 
-    """
-    return _expect_warnings(sa_exc.SAWarning, messages, **kw)
+    """  # noqa
+    return _expect_warnings(
+        (sa_exc.SAWarning, sa_exc.RemovedIn20Warning), messages, **kw
+    )
 
 
 @contextlib.contextmanager
diff --git a/lib/sqlalchemy/testing/schema.py b/lib/sqlalchemy/testing/schema.py
index f5bd1f7a2..8e26d2eaf 100644
--- a/lib/sqlalchemy/testing/schema.py
+++ b/lib/sqlalchemy/testing/schema.py
@@ -9,6 +9,7 @@ from . import config
 from . import exclusions
 from .. import event
 from .. import schema
+from .. import types as sqltypes
 
 
 __all__ = ["Table", "Column"]
@@ -115,6 +116,43 @@ def Column(*args, **kw):
     return col
 
 
+class eq_type_affinity(object):
+    """Helper to compare types inside of datastructures based on affinity.
+
+    E.g.::
+
+        eq_(
+            inspect(connection).get_columns("foo"),
+            [
+                {
+                    "name": "id",
+                    "type": testing.eq_type_affinity(sqltypes.INTEGER),
+                    "nullable": False,
+                    "default": None,
+                    "autoincrement": False,
+                },
+                {
+                    "name": "data",
+                    "type": testing.eq_type_affinity(sqltypes.NullType),
+                    "nullable": True,
+                    "default": None,
+                    "autoincrement": False,
+                },
+            ],
+        )
+
+    """
+
+    def __init__(self, target):
+        self.target = sqltypes.to_instance(target)
+
+    def __eq__(self, other):
+        return self.target._type_affinity is other._type_affinity
+
+    def __ne__(self, other):
+        return self.target._type_affinity is not other._type_affinity
+
+
 def _truncate_name(dialect, name):
     if len(name) > dialect.max_identifier_length:
         return (
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 9a2fdf95a..8c6543700 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -57,7 +57,7 @@ class _LiteralRoundTripFixture(object):
         t = Table("t", self.metadata, Column("x", type_))
         t.create()
 
-        with testing.db.connect() as conn:
+        with testing.db.begin() as conn:
             for value in input_:
                 ins = (
                     t.insert()
diff --git a/test/base/test_tutorials.py b/test/base/test_tutorials.py
index 69f9f7e90..4ac3fb981 100644
--- a/test/base/test_tutorials.py
+++ b/test/base/test_tutorials.py
@@ -89,7 +89,7 @@ class DocTest(fixtures.TestBase):
     def test_orm(self):
         self._run_doctest("orm/tutorial.rst")
 
-    @testing.emits_warning("SELECT statement has a cartesian")
+    @testing.emits_warning()
     def test_core(self):
         self._run_doctest("core/tutorial.rst")
 
diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 6e4038eb4..0bd8f7a5a 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -12,6 +12,7 @@ from sqlalchemy import schema
 from sqlalchemy import Table
 from sqlalchemy import testing
 from sqlalchemy import types
+from sqlalchemy import types as sqltypes
 from sqlalchemy import util
 from sqlalchemy.dialects import mssql
 from sqlalchemy.dialects.mssql import base
@@ -143,20 +144,38 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
         eq_(table2.c["col1"].dialect_options["mssql"]["identity_start"], 2)
         eq_(table2.c["col1"].dialect_options["mssql"]["identity_increment"], 3)
 
-    @testing.emits_warning("Did not recognize")
     @testing.provide_metadata
-    def test_skip_types(self):
-        metadata = self.metadata
-        with testing.db.connect() as c:
-            c.exec_driver_sql(
-                "create table foo (id integer primary key, data xml)"
-            )
+    def test_skip_types(self, connection):
+        connection.exec_driver_sql(
+            "create table foo (id integer primary key, data xml)"
+        )
         with mock.patch.object(
-            testing.db.dialect, "ischema_names", {"int": mssql.INTEGER}
+            connection.dialect, "ischema_names", {"int": mssql.INTEGER}
         ):
-            t1 = Table("foo", metadata, autoload=True)
-        assert isinstance(t1.c.id.type, Integer)
-        assert isinstance(t1.c.data.type, types.NullType)
+            with testing.expect_warnings(
+                "Did not recognize type 'xml' of column 'data'"
+            ):
+                eq_(
+                    inspect(connection).get_columns("foo"),
+                    [
+                        {
+                            "name": "id",
+                            "type": testing.eq_type_affinity(sqltypes.INTEGER),
+                            "nullable": False,
+                            "default": None,
+                            "autoincrement": False,
+                        },
+                        {
+                            "name": "data",
+                            "type": testing.eq_type_affinity(
+                                sqltypes.NullType
+                            ),
+                            "nullable": True,
+                            "default": None,
+                            "autoincrement": False,
+                        },
+                    ],
+                )
 
     @testing.provide_metadata
     def test_cross_schema_fk_pk_name_overlaps(self):
diff --git a/test/dialect/oracle/test_reflection.py b/test/dialect/oracle/test_reflection.py
index beb24923d..b9975f65e 100644
--- a/test/dialect/oracle/test_reflection.py
+++ b/test/dialect/oracle/test_reflection.py
@@ -5,6 +5,7 @@ from sqlalchemy import exc
 from sqlalchemy import FLOAT
 from sqlalchemy import ForeignKey
 from sqlalchemy import ForeignKeyConstraint
+from sqlalchemy import func
 from sqlalchemy import Index
 from sqlalchemy import inspect
 from sqlalchemy import INTEGER
@@ -13,7 +14,6 @@ from sqlalchemy import MetaData
 from sqlalchemy import Numeric
 from sqlalchemy import PrimaryKeyConstraint
 from sqlalchemy import select
-from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import text
 from sqlalchemy import Unicode
@@ -437,29 +437,6 @@ class DontReflectIOTTest(fixtures.TestBase):
         eq_(set(t.name for t in m.tables.values()), set(["admin_docindex"]))
 
 
-class UnsupportedIndexReflectTest(fixtures.TestBase):
-    __only_on__ = "oracle"
-    __backend__ = True
-
-    @testing.emits_warning("No column names")
-    @testing.provide_metadata
-    def test_reflect_functional_index(self):
-        metadata = self.metadata
-        Table(
-            "test_index_reflect",
-            metadata,
-            Column("data", String(20), primary_key=True),
-        )
-        metadata.create_all()
-
-        exec_sql(
-            testing.db,
-            "CREATE INDEX DATA_IDX ON " "TEST_INDEX_REFLECT (UPPER(DATA))",
-        )
-        m2 = MetaData(testing.db)
-        Table("test_index_reflect", m2, autoload=True)
-
-
 def all_tables_compression_missing():
     try:
         exec_sql(testing.db, "SELECT compression FROM all_tables")
@@ -610,6 +587,40 @@ class RoundTripIndexTest(fixtures.TestBase):
         )
 
     @testing.provide_metadata
+    def test_reflect_fn_index(self, connection):
+        """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.
+
+        """
+
+        metadata = self.metadata
+        s_table = Table(
+            "sometable",
+            metadata,
+            Column("group", Unicode(255), primary_key=True),
+            Column("col", Unicode(255)),
+        )
+
+        Index("data_idx", func.upper(s_table.c.col))
+
+        metadata.create_all(connection)
+
+        eq_(
+            inspect(connection).get_indexes("sometable"),
+            [
+                {
+                    "column_names": [],
+                    "dialect_options": {},
+                    "name": "data_idx",
+                    "unique": False,
+                }
+            ],
+        )
+
+    @testing.provide_metadata
     def test_basic(self):
         metadata = self.metadata
 
diff --git a/test/engine/test_ddlevents.py b/test/engine/test_ddlevents.py
index 8f5834cc9..9f661abf0 100644
--- a/test/engine/test_ddlevents.py
+++ b/test/engine/test_ddlevents.py
@@ -549,6 +549,163 @@ class DDLExecutionTest(fixtures.TestBase):
             )
 
 
+class DDLTransactionTest(fixtures.TestBase):
+    """test DDL transactional behavior as of SQLAlchemy 1.4."""
+
+    @testing.fixture
+    def metadata_fixture(self):
+        m = MetaData()
+        Table("t1", m, Column("q", Integer))
+        Table("t2", m, Column("q", Integer))
+
+        try:
+            yield m
+        finally:
+            m.drop_all(testing.db)
+
+    def _listening_engine_fixture(self, future=False):
+        eng = engines.testing_engine(future=future)
+
+        m1 = mock.Mock()
+
+        event.listen(eng, "begin", m1.begin)
+        event.listen(eng, "commit", m1.commit)
+        event.listen(eng, "rollback", m1.rollback)
+
+        @event.listens_for(eng, "before_cursor_execute")
+        def before_cursor_execute(
+            conn, cursor, statement, parameters, context, executemany
+        ):
+            if "CREATE TABLE" in statement:
+                m1.cursor_execute("CREATE TABLE ...")
+
+        eng.connect().close()
+
+        return eng, m1
+
+    @testing.fixture
+    def listening_engine_fixture(self):
+        return self._listening_engine_fixture(future=False)
+
+    @testing.fixture
+    def future_listening_engine_fixture(self):
+        return self._listening_engine_fixture(future=True)
+
+    def test_ddl_legacy_engine(
+        self, metadata_fixture, listening_engine_fixture
+    ):
+        eng, m1 = listening_engine_fixture
+
+        metadata_fixture.create_all(eng)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call.begin(mock.ANY),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+            ],
+        )
+
+    def test_ddl_future_engine(
+        self, metadata_fixture, future_listening_engine_fixture
+    ):
+        eng, m1 = future_listening_engine_fixture
+
+        metadata_fixture.create_all(eng)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call.begin(mock.ANY),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+            ],
+        )
+
+    def test_ddl_legacy_connection_no_transaction(
+        self, metadata_fixture, listening_engine_fixture
+    ):
+        eng, m1 = listening_engine_fixture
+
+        with eng.connect() as conn:
+            with testing.expect_deprecated(
+                "The current statement is being autocommitted using "
+                "implicit autocommit"
+            ):
+                metadata_fixture.create_all(conn)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+            ],
+        )
+
+    def test_ddl_legacy_connection_transaction(
+        self, metadata_fixture, listening_engine_fixture
+    ):
+        eng, m1 = listening_engine_fixture
+
+        with eng.connect() as conn:
+            with conn.begin():
+                metadata_fixture.create_all(conn)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call.begin(mock.ANY),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+            ],
+        )
+
+    def test_ddl_future_connection_autobegin_transaction(
+        self, metadata_fixture, future_listening_engine_fixture
+    ):
+        eng, m1 = future_listening_engine_fixture
+
+        with eng.connect() as conn:
+            metadata_fixture.create_all(conn)
+
+            conn.commit()
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call.begin(mock.ANY),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+            ],
+        )
+
+    def test_ddl_future_connection_explicit_begin_transaction(
+        self, metadata_fixture, future_listening_engine_fixture
+    ):
+        eng, m1 = future_listening_engine_fixture
+
+        with eng.connect() as conn:
+            with conn.begin():
+                metadata_fixture.create_all(conn)
+
+        eq_(
+            m1.mock_calls,
+            [
+                mock.call.begin(mock.ANY),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.cursor_execute("CREATE TABLE ..."),
+                mock.call.commit(mock.ANY),
+            ],
+        )
+
+
 class DDLTest(fixtures.TestBase, AssertsCompiledSQL):
     def mock_engine(self):
         def executor(*a, **kw):
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index 62bac312b..d733bd6a7 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -13,6 +13,7 @@ from sqlalchemy import pool
 from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import testing
+from sqlalchemy import text
 from sqlalchemy import VARCHAR
 from sqlalchemy.engine import reflection
 from sqlalchemy.engine.base import Connection
@@ -176,31 +177,22 @@ class CreateEngineTest(fixtures.TestBase):
             )
 
 
-class TransactionTest(fixtures.TestBase):
+class TransactionTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        metadata = MetaData()
-        cls.users = Table(
-            "query_users",
+    def define_tables(cls, metadata):
+        Table(
+            "users",
             metadata,
             Column("user_id", Integer, primary_key=True),
             Column("user_name", String(20)),
             test_needs_acid=True,
         )
-        cls.users.create(testing.db)
-
-    def teardown(self):
-        with testing.db.connect() as conn:
-            conn.execute(self.users.delete())
-
-    @classmethod
-    def teardown_class(cls):
-        cls.users.drop(testing.db)
+        Table("inserttable", metadata, Column("data", String(20)))
 
     def test_transaction_container(self):
-        users = self.users
+        users = self.tables.users
 
         def go(conn, table, data):
             for d in data:
@@ -231,6 +223,38 @@ class TransactionTest(fixtures.TestBase):
         with testing.db.connect() as conn:
             eq_(conn.execute(users.select()).fetchall(), [(1, "user1")])
 
+    def test_implicit_autocommit_compiled(self):
+        users = self.tables.users
+
+        with testing.db.connect() as conn:
+            with testing.expect_deprecated_20(
+                "The current statement is being autocommitted "
+                "using implicit autocommit."
+            ):
+                conn.execute(
+                    users.insert(), {"user_id": 1, "user_name": "user3"}
+                )
+
+    def test_implicit_autocommit_text(self):
+        with testing.db.connect() as conn:
+            with testing.expect_deprecated_20(
+                "The current statement is being autocommitted "
+                "using implicit autocommit."
+            ):
+                conn.execute(
+                    text("insert into inserttable (data) values ('thedata')")
+                )
+
+    def test_implicit_autocommit_driversql(self):
+        with testing.db.connect() as conn:
+            with testing.expect_deprecated_20(
+                "The current statement is being autocommitted "
+                "using implicit autocommit."
+            ):
+                conn.exec_driver_sql(
+                    "insert into inserttable (data) values ('thedata')"
+                )
+
 
 class HandleInvalidatedOnConnectTest(fixtures.TestBase):
     __requires__ = ("sqlite",)