Disallow AliasedReturnsRows from execution

Executing a :class:`_sql.Subquery` using :meth:`_engine.Connection.execute`
is deprecated and will emit a deprecation warning; this use case was an
oversight that should have been removed from 1.4. The operation will now
execute the underlying :class:`_sql.Select` object directly for backwards
compatibility. Similarly, the :class:`_sql.CTE` class is also not
appropriate for execution. In 1.3, attempting to execute a CTE would result
in an invalid "blank" SQL statement being executed; since this use case was
not working it now raises :class:`_exc.ObjectNotExecutableError`.
Previously, 1.4 was attempting to execute the CTE as a statement however it
was working only erratically.

The change also breaks out StatementRole from ReturnsRowsRole, as these
roles should not be in the same lineage (some statements don't return
rows, the whole class of ReturnsRows that are from clauses are
not statements).    Consolidate StatementRole and
CoerceTextStatementRole as there's no usage difference between
these.   Simplify some old tests that were trying to make
sure that "execution options" didn't transmit from a cte/subquery
out to a select; as cte/subuqery() aren't executable in any case
the options are removed.

Fixes: #6204
Change-Id: I62613b7ab418afdd22c409eae75659e3f52fb65f

15 files changed, 79 insertions, 66 deletions

diff --git a/doc/build/changelog/unreleased_14/6204.rst b/doc/build/changelog/unreleased_14/6204.rst
new file mode 100644
index 000000000..b6c518cb9
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6204.rst
@@ -0,0 +1,14 @@
+.. change::
+    :tags: bug, sql
+    :tickets: 6204
+
+    Executing a :class:`_sql.Subquery` using :meth:`_engine.Connection.execute`
+    is deprecated and will emit a deprecation warning; this use case was an
+    oversight that should have been removed from 1.4. The operation will now
+    execute the underlying :class:`_sql.Select` object directly for backwards
+    compatibility. Similarly, the :class:`_sql.CTE` class is also not
+    appropriate for execution. In 1.3, attempting to execute a CTE would result
+    in an invalid "blank" SQL statement being executed; since this use case was
+    not working it now raises :class:`_exc.ObjectNotExecutableError`.
+    Previously, 1.4 was attempting to execute the CTE as a statement however it
+    was working only erratically.
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index e2cc36999..6f77fd706 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -65,7 +65,7 @@ __all__ = (
 )
 
 
-class ORMStatementRole(roles.CoerceTextStatementRole):
+class ORMStatementRole(roles.StatementRole):
     _role_name = (
         "Executable SQL or text() construct, including ORM " "aware objects"
     )
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 7e2fde749..b94f9ac37 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -3234,8 +3234,6 @@ class FromStatement(GroupedElement, SelectBase, Executable):
 
     _compile_state_factory = ORMFromStatementCompileState.create_for_statement
 
-    _is_future = True
-
     _for_update_arg = None
 
     _traverse_internals = [
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 340017adf..0562569bf 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1584,7 +1584,7 @@ class Session(_SessionClassMethods):
 
 
         """
-        statement = coercions.expect(roles.CoerceTextStatementRole, statement)
+        statement = coercions.expect(roles.StatementRole, statement)
 
         if kw:
             util.warn_deprecated_20(
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index ac9d66970..81685dfe0 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -754,7 +754,7 @@ class ExecutableOption(HasCopyInternals, HasCacheKey):
         return c
 
 
-class Executable(roles.CoerceTextStatementRole, Generative):
+class Executable(roles.StatementRole, Generative):
     """Mark a :class:`_expression.ClauseElement` as supporting execution.
 
     :class:`.Executable` is a superclass for all "statement" types
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index 35ac1a5ba..c00262fd5 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -824,11 +824,7 @@ class ReturnsRowsImpl(RoleImpl):
     __slots__ = ()
 
 
-class StatementImpl(_NoTextCoercion, RoleImpl):
-    __slots__ = ()
-
-
-class CoerceTextStatementImpl(_CoerceLiterals, RoleImpl):
+class StatementImpl(_CoerceLiterals, RoleImpl):
     __slots__ = ()
 
     def _implicit_coercions(
@@ -837,7 +833,7 @@ class CoerceTextStatementImpl(_CoerceLiterals, RoleImpl):
         if resolved._is_lambda_element:
             return resolved
         else:
-            return super(CoerceTextStatementImpl, self)._implicit_coercions(
+            return super(StatementImpl, self)._implicit_coercions(
                 original_element, resolved, argname=argname, **kw
             )
 
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index c48338303..dfa6c0f8f 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -307,9 +307,9 @@ class ClauseElement(
         return d
 
     def _execute_on_connection(
-        self, connection, multiparams, params, execution_options
+        self, connection, multiparams, params, execution_options, _force=False
     ):
-        if self.supports_execution:
+        if _force or self.supports_execution:
             return connection._execute_clauseelement(
                 self, multiparams, params, execution_options
             )
diff --git a/lib/sqlalchemy/sql/lambdas.py b/lib/sqlalchemy/sql/lambdas.py
index ebf576c8f..06db8f95e 100644
--- a/lib/sqlalchemy/sql/lambdas.py
+++ b/lib/sqlalchemy/sql/lambdas.py
@@ -100,7 +100,7 @@ def lambda_stmt(
 
     return StatementLambdaElement(
         lmb,
-        roles.CoerceTextStatementRole,
+        roles.StatementRole,
         LambdaOptions(
             enable_tracking=enable_tracking,
             track_on=track_on,
@@ -155,9 +155,7 @@ class LambdaElement(elements.ClauseElement):
         self.tracker_key = (fn.__code__,)
         self.opts = opts
 
-        if apply_propagate_attrs is None and (
-            role is roles.CoerceTextStatementRole
-        ):
+        if apply_propagate_attrs is None and (role is roles.StatementRole):
             apply_propagate_attrs = self
 
         rec = self._retrieve_tracker_rec(fn, apply_propagate_attrs, opts)
@@ -493,10 +491,6 @@ class StatementLambdaElement(roles.AllowsLambdaRole, LambdaElement):
         return self._rec.expected_expr._effective_plugin_target
 
     @property
-    def _is_future(self):
-        return self._rec.expected_expr._is_future
-
-    @property
     def _execution_options(self):
         return self._rec.expected_expr._execution_options
 
diff --git a/lib/sqlalchemy/sql/roles.py b/lib/sqlalchemy/sql/roles.py
index 1c8276eb6..7d64e8382 100644
--- a/lib/sqlalchemy/sql/roles.py
+++ b/lib/sqlalchemy/sql/roles.py
@@ -155,26 +155,20 @@ class AnonymizedFromClauseRole(StrictFromClauseRole):
         raise NotImplementedError()
 
 
-class CoerceTextStatementRole(SQLRole):
-    _role_name = "Executable SQL or text() construct"
+class ReturnsRowsRole(SQLRole):
+    _role_name = (
+        "Row returning expression such as a SELECT, a FROM clause, or an "
+        "INSERT/UPDATE/DELETE with RETURNING"
+    )
 
 
-class StatementRole(CoerceTextStatementRole):
+class StatementRole(SQLRole):
     _role_name = "Executable SQL or text() construct"
 
-    _is_future = False
-
     _propagate_attrs = util.immutabledict()
 
 
-class ReturnsRowsRole(StatementRole):
-    _role_name = (
-        "Row returning expression such as a SELECT, a FROM clause, or an "
-        "INSERT/UPDATE/DELETE with RETURNING"
-    )
-
-
-class SelectStatementRole(ReturnsRowsRole):
+class SelectStatementRole(StatementRole, ReturnsRowsRole):
     _role_name = "SELECT construct or equivalent text() construct"
 
     def subquery(self):
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index a2e5780f8..f12744cfa 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1575,9 +1575,6 @@ class AliasedReturnsRows(NoInit, FromClause):
         self.element = coercions.expect(
             roles.ReturnsRowsRole, selectable, apply_propagate_attrs=self
         )
-        self.supports_execution = selectable.supports_execution
-        if self.supports_execution:
-            self._execution_options = selectable._execution_options
         self.element = selectable
         self._orig_name = name
         if name is None:
@@ -2338,6 +2335,23 @@ class Subquery(AliasedReturnsRows):
     def as_scalar(self):
         return self.element.set_label_style(LABEL_STYLE_NONE).scalar_subquery()
 
+    def _execute_on_connection(
+        self,
+        connection,
+        multiparams,
+        params,
+        execution_options,
+    ):
+        util.warn_deprecated(
+            "Executing a subquery object is deprecated and will raise "
+            "ObjectNotExecutableError in an upcoming release.  Please "
+            "execute the underlying select() statement directly.",
+            "1.4",
+        )
+        return self.element._execute_on_connection(
+            connection, multiparams, params, execution_options, _force=True
+        )
+
 
 class FromGrouping(GroupedElement, FromClause):
     """Represent a grouping of a FROM clause"""
@@ -4485,7 +4499,6 @@ class Select(
             ("_distinct", InternalTraversal.dp_boolean),
             ("_distinct_on", InternalTraversal.dp_clauseelement_tuple),
             ("_label_style", InternalTraversal.dp_plain_obj),
-            ("_is_future", InternalTraversal.dp_boolean),
         ]
         + HasPrefixes._has_prefixes_traverse_internals
         + HasSuffixes._has_suffixes_traverse_internals
diff --git a/lib/sqlalchemy/testing/suite/test_results.py b/lib/sqlalchemy/testing/suite/test_results.py
index 6c2880ad4..e8ad88f24 100644
--- a/lib/sqlalchemy/testing/suite/test_results.py
+++ b/lib/sqlalchemy/testing/suite/test_results.py
@@ -333,14 +333,18 @@ class ServerSideCursorsTest(
 
     def test_aliases_and_ss(self):
         engine = self._fixture(False)
-        s1 = select(1).execution_options(stream_results=True).alias()
+        s1 = (
+            select(sql.literal_column("1").label("x"))
+            .execution_options(stream_results=True)
+            .subquery()
+        )
+
+        # options don't propagate out when subquery is used as a FROM clause
         with engine.begin() as conn:
-            result = conn.execute(s1)
-            assert self._is_server_side(result.cursor)
+            result = conn.execute(s1.select())
+            assert not self._is_server_side(result.cursor)
             result.close()
 
-        # s1's options shouldn't affect s2 when s2 is used as a
-        # from_obj.
         s2 = select(1).select_from(s1)
         with engine.begin() as conn:
             result = conn.execute(s2)
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 09219cfdb..0de5ed124 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -42,6 +42,7 @@ from sqlalchemy.testing import is_false
 from sqlalchemy.testing import is_not
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import expect_deprecated
 from sqlalchemy.testing.assertsql import CompiledSQL
 from sqlalchemy.testing.mock import call
 from sqlalchemy.testing.mock import Mock
@@ -349,6 +350,8 @@ class ExecuteTest(fixtures.TablesTest):
             tsa.and_(True).compile(),
             column("foo"),
             column("foo").compile(),
+            select(1).cte(),
+            # select(1).subquery(),
             MetaData(),
             Integer(),
             tsa.Index(name="foo"),
@@ -362,6 +365,15 @@ class ExecuteTest(fixtures.TablesTest):
                     obj,
                 )
 
+    def test_subquery_exec_warning(self):
+        for obj in (select(1).alias(), select(1).subquery()):
+            with testing.db.connect() as conn:
+                with expect_deprecated(
+                    "Executing a subquery object is deprecated and will "
+                    "raise ObjectNotExecutableError"
+                ):
+                    eq_(conn.execute(obj).scalar(), 1)
+
     def test_stmt_exception_bytestring_raised(self):
         name = util.u("méil")
         users = self.tables.users
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index b2d443438..afa6aecf8 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4550,20 +4550,20 @@ class ExecutionOptionsTest(fixtures.TestBase):
         eq_(compiled.execution_options, {"autocommit": True})
 
     def test_embedded_element_true_to_none(self):
-        stmt = table1.insert().cte()
+        stmt = table1.insert()
         eq_(stmt._execution_options, {"autocommit": True})
-        s2 = select(table1).select_from(stmt)
+        s2 = select(table1).select_from(stmt.cte())
         eq_(s2._execution_options, {})
 
         compiled = s2.compile()
         eq_(compiled.execution_options, {"autocommit": True})
 
     def test_embedded_element_true_to_false(self):
-        stmt = table1.insert().cte()
+        stmt = table1.insert()
         eq_(stmt._execution_options, {"autocommit": True})
         s2 = (
             select(table1)
-            .select_from(stmt)
+            .select_from(stmt.cte())
             .execution_options(autocommit=False)
         )
         eq_(s2._execution_options, {"autocommit": False})
diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index 92f3b215f..fc2c9b40d 100644
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -1150,7 +1150,7 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
         products = table("products", column("id"), column("price"))
 
         cte = products.select().cte("pd")
-        assert "autocommit" not in cte._execution_options
+        assert "autocommit" not in cte.select()._execution_options
 
         stmt = products.update().where(products.c.price == cte.c.price)
         eq_(stmt.compile().execution_options["autocommit"], True)
diff --git a/test/sql/test_roles.py b/test/sql/test_roles.py
index e47a7e889..997311071 100644
--- a/test/sql/test_roles.py
+++ b/test/sql/test_roles.py
@@ -209,24 +209,14 @@ class RoleTest(fixtures.TestBase):
         ):
             expect(roles.ExpressionElementRole, t.select().alias())
 
-    def test_statement_no_text_coercion(self):
-        assert_raises_message(
-            exc.ArgumentError,
-            r"Textual SQL expression 'select \* from table' should be "
-            r"explicitly declared",
-            expect,
-            roles.StatementRole,
-            "select * from table",
-        )
-
     def test_statement_text_coercion(self):
         with testing.expect_deprecated_20(
             "Using plain strings to indicate SQL statements"
         ):
             is_true(
-                expect(
-                    roles.CoerceTextStatementRole, "select * from table"
-                ).compare(text("select * from table"))
+                expect(roles.StatementRole, "select * from table").compare(
+                    text("select * from table")
+                )
             )
 
     def test_select_statement_no_text_coercion(self):
@@ -282,13 +272,11 @@ class RoleTest(fixtures.TestBase):
         )
 
     def test_statement_coercion_select(self):
-        is_true(
-            expect(roles.CoerceTextStatementRole, select(t)).compare(select(t))
-        )
+        is_true(expect(roles.StatementRole, select(t)).compare(select(t)))
 
     def test_statement_coercion_ddl(self):
         d1 = DDL("hi")
-        is_(expect(roles.CoerceTextStatementRole, d1), d1)
+        is_(expect(roles.StatementRole, d1), d1)
 
     def test_strict_from_clause_role(self):
         stmt = select(t).subquery()
@@ -325,7 +313,7 @@ class RoleTest(fixtures.TestBase):
 
     def test_statement_coercion_sequence(self):
         s1 = Sequence("hi")
-        is_(expect(roles.CoerceTextStatementRole, s1), s1)
+        is_(expect(roles.StatementRole, s1), s1)
 
     def test_columns_clause_role(self):
         is_(expect(roles.ColumnsClauseRole, t.c.q), t.c.q)