Merge "Disallow AliasedReturnsRows from execution"

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 90ae1b700..38f1d26b4 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -3254,8 +3254,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 2115c32a7..d7a74ee2e 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4635,20 +4635,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)