Implement Oracle SERIALIZABLE + real read of isolation level

There's some significant awkwardness in that we can't
read the level unless a transaction is started, which normally
does not occur unless DML is emitted.  The implementation
uses the local_transaction_id function to start a transaction.
It is not known what the performance impact of this might
have, however by default the function is called only once
on first connect and later only if the get_isolation_level()
method is used.

Fixes: #5755
Change-Id: I0453a6b0a49420826707f660931002ba2338fbf0

3 files changed, 104 insertions, 33 deletions

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 223c1db98..cbaed445b 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -79,18 +79,14 @@ This step is also required when using table reflection, i.e. autoload=True::
    in a :class:`_schema.Column` to specify the option of an autoincrementing
    column.
 
+.. _oracle_isolation_level:
+
 Transaction Isolation Level / Autocommit
 ----------------------------------------
 
-The Oracle database supports "READ COMMITTED" and "SERIALIZABLE" modes
-of isolation, however the SQLAlchemy Oracle dialect currently only has
-explicit support for "READ COMMITTED".  It is possible to emit a
-"SET TRANSACTION" statement on a connection in order to use SERIALIZABLE
-isolation, however the SQLAlchemy dialect will remain unaware of this setting,
-such as if the :meth:`_engine.Connection.get_isolation_level` method is used;
-this method is hardcoded to return "READ COMMITTED" right now.
-
-The AUTOCOMMIT isolation level is also supported by the cx_Oracle dialect.
+The Oracle database supports "READ COMMITTED" and "SERIALIZABLE" modes of
+isolation. The AUTOCOMMIT isolation level is also supported by the cx_Oracle
+dialect.
 
 To set using per-connection execution options::
 
@@ -99,15 +95,27 @@ To set using per-connection execution options::
         isolation_level="AUTOCOMMIT"
     )
 
+For ``READ COMMITTED`` and ``SERIALIZABLE``, the Oracle dialect sets the
+level at the session level using ``ALTER SESSION``, which is reverted back
+to its default setting when the connection is returned to the connection
+pool.
+
 Valid values for ``isolation_level`` include:
 
 * ``READ COMMITTED``
 * ``AUTOCOMMIT``
+* ``SERIALIZABLE``
 
+.. note:: The implementation :meth:`_engine.Connection.get_isolation_level`
+   implemented by the Oracle dialect necessarily forces the start of
+   a transaction using the Oracle LOCAL_TRANSACTION_ID function; otherwise
+   no level is normally readable.
 
 .. versionadded:: 1.3.16 added support for AUTOCOMMIT to the cx_oracle dialect
-   as well as the notion of a default isolation level, currently hardcoded
-   to "READ COMMITTED".
+   as well as the notion of a default isolation level
+
+.. versionadded:: 1.3.21 Added support for SERIALIZABLE as well as live
+   reading of the isolation level.
 
 .. seealso::
 
@@ -1542,19 +1550,13 @@ class OracleDialect(default.DefaultDialect):
             connection, additional_tests
         )
 
-    _isolation_lookup = ["READ COMMITTED"]
+    _isolation_lookup = ["READ COMMITTED", "SERIALIZABLE"]
 
     def get_isolation_level(self, connection):
-        return "READ COMMITTED"
+        raise NotImplementedError("implemented by cx_Oracle dialect")
 
     def set_isolation_level(self, connection, level):
-        # prior to adding AUTOCOMMIT support for cx_Oracle, the Oracle dialect
-        # had no notion of setting the isolation level.  As Oracle
-        # does not have a straightforward way of getting the isolation level
-        # if a server-side transaction is not yet in progress, we currently
-        # hardcode to only support "READ COMMITTED" and "AUTOCOMMIT" at the
-        # cx_oracle level.  See #5200.
-        pass
+        raise NotImplementedError("implemented by cx_Oracle dialect")
 
     def has_table(self, connection, table_name, schema=None):
         if not schema:
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index 8eb9f8b3c..e7db9272f 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -881,6 +881,66 @@ class OracleDialect_cx_oracle(OracleDialect):
 
         self._detect_decimal_char(connection)
 
+    def get_isolation_level(self, connection):
+        # sources:
+
+        # general idea of transaction id, have to start one, etc.
+        # https://stackoverflow.com/questions/10711204/how-to-check-isoloation-level
+
+        # how to decode xid cols from v$transaction to match
+        # https://asktom.oracle.com/pls/apex/f?p=100:11:0::::P11_QUESTION_ID:9532779900346079444
+
+        # Oracle tuple comparison without using IN:
+        # https://www.sql-workbench.eu/comparison/tuple_comparison.html
+
+        with connection.cursor() as cursor:
+            # this is the only way to ensure a transaction is started without
+            # actually running DML.   There's no way to see the configured
+            # isolation level without getting it from v$transaction which
+            # means transaction has to be started.
+            outval = cursor.var(str)
+            cursor.execute(
+                """
+                begin
+                   :trans_id := dbms_transaction.local_transaction_id( TRUE );
+                end;
+                """,
+                {"trans_id": outval},
+            )
+            trans_id = outval.getvalue()
+            xidusn, xidslot, xidsqn = trans_id.split(".", 2)
+
+            cursor.execute(
+                "SELECT CASE BITAND(t.flag, POWER(2, 28)) "
+                "WHEN 0 THEN 'READ COMMITTED' "
+                "ELSE 'SERIALIZABLE' END AS isolation_level "
+                "FROM v$transaction t WHERE "
+                "(t.xidusn, t.xidslot, t.xidsqn) = "
+                "((:xidusn, :xidslot, :xidsqn))",
+                {"xidusn": xidusn, "xidslot": xidslot, "xidsqn": xidsqn},
+            )
+            row = cursor.fetchone()
+            if row is None:
+                raise exc.InvalidRequestError(
+                    "could not retrieve isolation level"
+                )
+            result = row[0]
+
+        return result
+
+    def set_isolation_level(self, connection, level):
+        if hasattr(connection, "connection"):
+            dbapi_connection = connection.connection
+        else:
+            dbapi_connection = connection
+        if level == "AUTOCOMMIT":
+            dbapi_connection.autocommit = True
+        else:
+            dbapi_connection.autocommit = False
+            connection.rollback()
+            with connection.cursor() as cursor:
+                cursor.execute("ALTER SESSION SET ISOLATION_LEVEL=%s" % level)
+
     def _detect_decimal_char(self, connection):
         # we have the option to change this setting upon connect,
         # or just look at what it is upon connect and convert.
@@ -1186,18 +1246,5 @@ class OracleDialect_cx_oracle(OracleDialect):
     def do_recover_twophase(self, connection):
         connection.info.pop("cx_oracle_prepared", None)
 
-    def set_isolation_level(self, connection, level):
-        if hasattr(connection, "connection"):
-            dbapi_connection = connection.connection
-        else:
-            dbapi_connection = connection
-        if level == "AUTOCOMMIT":
-            dbapi_connection.autocommit = True
-        else:
-            dbapi_connection.autocommit = False
-            super(OracleDialect_cx_oracle, self).set_isolation_level(
-                dbapi_connection, level
-            )
-
 
 dialect = OracleDialect_cx_oracle
diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index 7f697b915..f860a321b 100644
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -122,6 +122,28 @@ class IsolationLevelTest(fixtures.TestBase):
 
             eq_(conn.get_isolation_level(), existing)
 
+    def test_all_levels(self):
+        levels = requirements.get_isolation_levels(config)
+
+        all_levels = levels["supported"]
+
+        for level in set(all_levels).difference(["AUTOCOMMIT"]):
+            with config.db.connect() as conn:
+                conn.execution_options(isolation_level=level)
+
+                eq_(conn.get_isolation_level(), level)
+
+                trans = conn.begin()
+                trans.rollback()
+
+                eq_(conn.get_isolation_level(), level)
+
+            with config.db.connect() as conn:
+                eq_(
+                    conn.get_isolation_level(),
+                    levels["default"],
+                )
+
 
 class AutocommitTest(fixtures.TablesTest):