audition pymssql once more; retire sane_rowcount_returning

pymssql seems to be maintained again and seems to be working
completely, so let's try re-enabling it.

Fixed issue in the new :class:`.Uuid` datatype which prevented it from
working with the pymssql driver. As pymssql seems to be maintained again,
restored testing support for pymssql.

Tweaked the pymssql dialect to take better advantage of
RETURNING for INSERT statements in order to retrieve last inserted primary
key values, in the same way as occurs for the mssql+pyodbc dialect right
now.

Identified that the ``sqlite`` and ``mssql+pyodbc`` dialects are now
compatible with the SQLAlchemy ORM's "versioned rows" feature, since
SQLAlchemy now computes rowcount for a RETURNING statement in this specific
case by counting the rows returned, rather than relying upon
``cursor.rowcount``.  In particular, the ORM versioned rows use case
(documented at :ref:`mapper_version_counter`) should now be fully
supported with the SQL Server pyodbc dialect.

Change-Id: I38a0666587212327aecf8f98e86031ab25d1f14d
References: #5321
Fixes: #9414

9 files changed, 165 insertions, 59 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index a30c57c7f..b970f6c0a 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -885,29 +885,20 @@ The SQL Server drivers may have limited ability to return the number
 of rows updated from an UPDATE or DELETE statement.
 
 As of this writing, the PyODBC driver is not able to return a rowcount when
-OUTPUT INSERTED is used.  This impacts the SQLAlchemy ORM's versioning feature
-in many cases where server-side value generators are in use in that while the
-versioning operations can succeed, the ORM cannot always check that an UPDATE
-or DELETE statement matched the number of rows expected, which is how it
-verifies that the version identifier matched.   When this condition occurs, a
-warning will be emitted but the operation will proceed.
-
-The use of OUTPUT INSERTED can be disabled by setting the
-:paramref:`_schema.Table.implicit_returning` flag to ``False`` on a particular
-:class:`_schema.Table`, which in declarative looks like::
-
-    class MyTable(Base):
-        __tablename__ = 'mytable'
-        id = Column(Integer, primary_key=True)
-        stuff = Column(String(10))
-        timestamp = Column(TIMESTAMP(), default=text('DEFAULT'))
-        __mapper_args__ = {
-            'version_id_col': timestamp,
-            'version_id_generator': False,
-        }
-        __table_args__ = {
-            'implicit_returning': False
-        }
+OUTPUT INSERTED is used.    Previous versions of SQLAlchemy therefore had
+limitations for features such as the "ORM Versioning" feature that relies upon
+accurate rowcounts in order to match version numbers with matched rows.
+
+SQLAlchemy 2.0 now retrieves the "rowcount" manually for these particular use
+cases based on counting the rows that arrived back within RETURNING; so while
+the driver still has this limitation, the ORM Versioning feature is no longer
+impacted by it. As of SQLAlchemy 2.0.5, ORM versioning has been fully
+re-enabled for the pyodbc driver.
+
+.. versionchanged:: 2.0.5  ORM versioning support is restored for the pyodbc
+   driver.  Previously, a warning would be emitted during ORM flush that
+   versioning was not supported.
+
 
 Enabling Snapshot Isolation
 ---------------------------
@@ -2965,6 +2956,7 @@ class MSDialect(default.DefaultDialect):
     supports_statement_cache = True
     supports_default_values = True
     supports_empty_insert = False
+    favor_returning_over_lastrowid = True
 
     supports_comments = True
     supports_default_metavalue = False
diff --git a/lib/sqlalchemy/dialects/mssql/provision.py b/lib/sqlalchemy/dialects/mssql/provision.py
index a7ecf4aa3..336e10cd9 100644
--- a/lib/sqlalchemy/dialects/mssql/provision.py
+++ b/lib/sqlalchemy/dialects/mssql/provision.py
@@ -12,6 +12,7 @@ from ...schema import Table
 from ...testing.provision import create_db
 from ...testing.provision import drop_all_schema_objects_pre_tables
 from ...testing.provision import drop_db
+from ...testing.provision import generate_driver_url
 from ...testing.provision import get_temp_table_name
 from ...testing.provision import log
 from ...testing.provision import normalize_sequence
@@ -19,6 +20,27 @@ from ...testing.provision import run_reap_dbs
 from ...testing.provision import temp_table_keyword_args
 
 
+@generate_driver_url.for_db("mssql")
+def generate_driver_url(url, driver, query_str):
+
+    backend = url.get_backend_name()
+
+    new_url = url.set(drivername="%s+%s" % (backend, driver))
+
+    if driver != "pyodbc":
+        new_url = new_url.set(query="")
+
+    if query_str:
+        new_url = new_url.update_query_string(query_str)
+
+    try:
+        new_url.get_dialect()
+    except exc.NoSuchModuleError:
+        return None
+    else:
+        return new_url
+
+
 @create_db.for_db("mssql")
 def _mssql_create_db(cfg, eng, ident):
     with eng.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
diff --git a/lib/sqlalchemy/dialects/mssql/pymssql.py b/lib/sqlalchemy/dialects/mssql/pymssql.py
index 699f61714..3823db91b 100644
--- a/lib/sqlalchemy/dialects/mssql/pymssql.py
+++ b/lib/sqlalchemy/dialects/mssql/pymssql.py
@@ -16,10 +16,9 @@
 pymssql is a Python module that provides a Python DBAPI interface around
 `FreeTDS <https://www.freetds.org/>`_.
 
-.. note::
+.. versionchanged:: 2.0.5
 
-    pymssql is currently not included in SQLAlchemy's continuous integration
-    (CI) testing.
+    pymssql was restored to SQLAlchemy's continuous integration testing
 
 
 """  # noqa
@@ -51,6 +50,7 @@ class MSIdentifierPreparer_pymssql(MSIdentifierPreparer):
 class MSDialect_pymssql(MSDialect):
     supports_statement_cache = True
     supports_native_decimal = True
+    supports_native_uuid = True
     driver = "pymssql"
 
     preparer = MSIdentifierPreparer_pymssql
diff --git a/lib/sqlalchemy/dialects/mssql/pyodbc.py b/lib/sqlalchemy/dialects/mssql/pyodbc.py
index 4546cabcd..3fa752565 100644
--- a/lib/sqlalchemy/dialects/mssql/pyodbc.py
+++ b/lib/sqlalchemy/dialects/mssql/pyodbc.py
@@ -281,9 +281,9 @@ non-ODBC drivers such as pymssql where it works very well.
 Rowcount Support
 ----------------
 
-Pyodbc only has partial support for rowcount.  See the notes at
-:ref:`mssql_rowcount_versioning` for important notes when using ORM
-versioning.
+Previous limitations with the SQLAlchemy ORM's "versioned rows" feature with
+Pyodbc have been resolved as of SQLAlchemy 2.0.5. See the notes at
+:ref:`mssql_rowcount_versioning`.
 
 .. _mssql_pyodbc_fastexecutemany:
 
@@ -607,11 +607,10 @@ class MSExecutionContext_pyodbc(MSExecutionContext):
 class MSDialect_pyodbc(PyODBCConnector, MSDialect):
     supports_statement_cache = True
 
-    # mssql still has problems with this on Linux
+    # note this parameter is no longer used by the ORM or default dialect
+    # see #9414
     supports_sane_rowcount_returning = False
 
-    favor_returning_over_lastrowid = True
-
     execution_ctx_cls = MSExecutionContext_pyodbc
 
     colspecs = util.update_copy(
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index fcabf5bb8..20065d88e 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1910,7 +1910,10 @@ class SQLiteDialect(default.DefaultDialect):
     supports_default_values = True
     supports_default_metavalue = False
 
+    # sqlite issue:
     # https://github.com/python/cpython/issues/93421
+    # note this parameter is no longer used by the ORM or default dialect
+    # see #9414
     supports_sane_rowcount_returning = False
 
     supports_empty_insert = False
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index b8368001b..a331d4ed8 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -805,11 +805,7 @@ def _emit_update_statements(
             statement = statement.return_defaults(mapper.version_id_col)
             return_defaults = True
 
-        assert_singlerow = (
-            connection.dialect.supports_sane_rowcount
-            if not return_defaults
-            else connection.dialect.supports_sane_rowcount_returning
-        )
+        assert_singlerow = connection.dialect.supports_sane_rowcount
 
         assert_multirow = (
             assert_singlerow
@@ -1274,9 +1270,6 @@ def _emit_post_update_statements(
 
     if mapper._version_id_has_server_side_value:
         statement = statement.return_defaults(mapper.version_id_col)
-        return_defaults = True
-    else:
-        return_defaults = False
 
     # execute each UPDATE in the order according to the original
     # list of states to guarantee row access order, but
@@ -1291,11 +1284,7 @@ def _emit_post_update_statements(
         records = list(records)
         connection = key[0]
 
-        assert_singlerow = (
-            connection.dialect.supports_sane_rowcount
-            if not return_defaults
-            else connection.dialect.supports_sane_rowcount_returning
-        )
+        assert_singlerow = connection.dialect.supports_sane_rowcount
         assert_multirow = (
             assert_singlerow
             and connection.dialect.supports_sane_multi_rowcount
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 9bfc94e78..3332f7ce2 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -493,6 +493,13 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def arraysize(self):
+        """dialect includes the required pep-249 attribute
+        ``cursor.arraysize``"""
+
+        return exclusions.open()
+
+    @property
     def emulated_lastrowid(self):
         """target dialect retrieves cursor.lastrowid, or fetches
         from a database-side function after an insert() construct executes,
@@ -851,6 +858,24 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def date_implicit_bound(self):
+        """target dialect when given a date object will bind it such
+        that the database server knows the object is a date, and not
+        a plain string.
+
+        """
+        return exclusions.open()
+
+    @property
+    def time_implicit_bound(self):
+        """target dialect when given a time object will bind it such
+        that the database server knows the object is a time, and not
+        a plain string.
+
+        """
+        return exclusions.open()
+
+    @property
     def datetime_implicit_bound(self):
         """target dialect when given a datetime object will bind it such
         that the database server knows the object is a datetime, and not
diff --git a/lib/sqlalchemy/testing/suite/test_rowcount.py b/lib/sqlalchemy/testing/suite/test_rowcount.py
index 8e19a24a8..ba8e10437 100644
--- a/lib/sqlalchemy/testing/suite/test_rowcount.py
+++ b/lib/sqlalchemy/testing/suite/test_rowcount.py
@@ -3,6 +3,7 @@
 from sqlalchemy import bindparam
 from sqlalchemy import Column
 from sqlalchemy import Integer
+from sqlalchemy import MetaData
 from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import Table
@@ -88,23 +89,58 @@ class RowCountTest(fixtures.TablesTest):
         )
         eq_(r.rowcount, 3)
 
-    @testing.requires.update_returning
-    def test_update_rowcount_return_defaults(self, connection):
+    @testing.variation("implicit_returning", [True, False])
+    @testing.variation(
+        "dml",
+        [
+            ("update", testing.requires.update_returning),
+            ("delete", testing.requires.delete_returning),
+        ],
+    )
+    def test_update_delete_rowcount_return_defaults(
+        self, connection, implicit_returning, dml
+    ):
         """note this test should succeed for all RETURNING backends
         as of 2.0.  In
         Idf28379f8705e403a3c6a937f6a798a042ef2540 we changed rowcount to use
         len(rows) when we have implicit returning
 
         """
-        employees_table = self.tables.employees
+
+        if implicit_returning:
+            employees_table = self.tables.employees
+        else:
+            employees_table = Table(
+                "employees",
+                MetaData(),
+                Column(
+                    "employee_id",
+                    Integer,
+                    autoincrement=False,
+                    primary_key=True,
+                ),
+                Column("name", String(50)),
+                Column("department", String(1)),
+                implicit_returning=False,
+            )
 
         department = employees_table.c.department
-        stmt = (
-            employees_table.update()
-            .where(department == "C")
-            .values(name=employees_table.c.department + "Z")
-            .return_defaults()
-        )
+
+        if dml.update:
+            stmt = (
+                employees_table.update()
+                .where(department == "C")
+                .values(name=employees_table.c.department + "Z")
+                .return_defaults()
+            )
+        elif dml.delete:
+            stmt = (
+                employees_table.delete()
+                .where(department == "C")
+                .return_defaults()
+            )
+        else:
+            dml.fail()
 
         r = connection.execute(stmt)
         eq_(r.rowcount, 3)
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index d6a74b220..bc2885341 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -450,11 +450,6 @@ class _DateFixture(_LiteralRoundTripFixture, fixtures.TestBase):
             Column("decorated_date_data", Decorated),
         )
 
-    @testing.requires.datetime_implicit_bound
-    def test_select_direct(self, connection):
-        result = connection.scalar(select(literal(self.data)))
-        eq_(result, self.data)
-
     def test_round_trip(self, connection):
         date_table = self.tables.date_table
 
@@ -531,6 +526,11 @@ class DateTimeTest(_DateFixture, fixtures.TablesTest):
     datatype = DateTime
     data = datetime.datetime(2012, 10, 15, 12, 57, 18)
 
+    @testing.requires.datetime_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class DateTimeTZTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("datetime_timezone",)
@@ -540,6 +540,11 @@ class DateTimeTZTest(_DateFixture, fixtures.TablesTest):
         2012, 10, 15, 12, 57, 18, tzinfo=datetime.timezone.utc
     )
 
+    @testing.requires.datetime_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class DateTimeMicrosecondsTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("datetime_microseconds",)
@@ -566,6 +571,11 @@ class TimeTest(_DateFixture, fixtures.TablesTest):
     datatype = Time
     data = datetime.time(12, 57, 18)
 
+    @testing.requires.time_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class TimeTZTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("time_timezone",)
@@ -573,6 +583,11 @@ class TimeTZTest(_DateFixture, fixtures.TablesTest):
     datatype = Time(timezone=True)
     data = datetime.time(12, 57, 18, tzinfo=datetime.timezone.utc)
 
+    @testing.requires.time_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class TimeMicrosecondsTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("time_microseconds",)
@@ -580,6 +595,11 @@ class TimeMicrosecondsTest(_DateFixture, fixtures.TablesTest):
     datatype = Time
     data = datetime.time(12, 57, 18, 396)
 
+    @testing.requires.time_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class DateTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("date",)
@@ -587,6 +607,11 @@ class DateTest(_DateFixture, fixtures.TablesTest):
     datatype = Date
     data = datetime.date(2012, 10, 15)
 
+    @testing.requires.date_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class DateTimeCoercedToDateTimeTest(_DateFixture, fixtures.TablesTest):
     """this particular suite is testing that datetime parameters get
@@ -600,6 +625,11 @@ class DateTimeCoercedToDateTimeTest(_DateFixture, fixtures.TablesTest):
     data = datetime.datetime(2012, 10, 15, 12, 57, 18)
     compare = datetime.date(2012, 10, 15)
 
+    @testing.requires.datetime_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class DateTimeHistoricTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("datetime_historic",)
@@ -607,6 +637,11 @@ class DateTimeHistoricTest(_DateFixture, fixtures.TablesTest):
     datatype = DateTime
     data = datetime.datetime(1850, 11, 10, 11, 52, 35)
 
+    @testing.requires.date_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class DateHistoricTest(_DateFixture, fixtures.TablesTest):
     __requires__ = ("date_historic",)
@@ -614,6 +649,11 @@ class DateHistoricTest(_DateFixture, fixtures.TablesTest):
     datatype = Date
     data = datetime.date(1727, 4, 1)
 
+    @testing.requires.date_implicit_bound
+    def test_select_direct(self, connection):
+        result = connection.scalar(select(literal(self.data)))
+        eq_(result, self.data)
+
 
 class IntegerTest(_LiteralRoundTripFixture, fixtures.TestBase):
     __backend__ = True