Check for column expr in Oracle RETURNING check

Fixed regression in Oracle dialect introduced by :ticket:`4894` in
SQLAlchemy 1.3.11 where use of a SQL expression in RETURNING for an UPDATE
would fail to compile, due to a check for "server_default" when an
arbitrary SQL expression is not a column.

Fixes: #5813
Change-Id: I1977bb49bc971399195015ae45e761f774f4008d

diff --git a/doc/build/changelog/unreleased_13/5813.rst b/doc/build/changelog/unreleased_13/5813.rst
new file mode 100644 (file)
index 0000000..d6483a2
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5813.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, oracle
+    :tickets: 5813
+
+    Fixed regression in Oracle dialect introduced by :ticket:`4894` in
+    SQLAlchemy 1.3.11 where use of a SQL expression in RETURNING for an UPDATE
+    would fail to compile, due to a check for "server_default" when an
+    arbitrary SQL expression is not a column.
+

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 2b4b268c2c88354b1dc70d5edd60a3d4b5f4b707..1cc8b7aef07584b3ca1784d4a7f1cc457b6be05b 100644 (file)
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1017,6 +1017,7 @@ class OracleCompiler(compiler.SQLCompiler):
         ):
             if (
                 self.isupdate
+                and isinstance(column, sa_schema.Column)
                 and isinstance(column.server_default, Computed)
                 and not self.dialect._supports_update_returning_computed_cols
             ):

diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py
index 9f2afd7b7dad1fc33b8c96f3a14b0f17ae29701f..187cf0dd023f7935fe62f08c99022142a0f77a04 100644 (file)
--- a/test/sql/test_returning.py
+++ b/test/sql/test_returning.py
@@ -11,6 +11,7 @@ from sqlalchemy import select
 from sqlalchemy import Sequence
 from sqlalchemy import String
 from sqlalchemy import testing
+from sqlalchemy import type_coerce
 from sqlalchemy import update
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
@@ -121,6 +122,7 @@ class ReturningTest(fixtures.TablesTest, AssertsExecutionResults):
             Column("persons", Integer),
             Column("full", Boolean),
             Column("goofy", GoofyType(50)),
+            Column("strval", String(50)),
         )
 
     def test_column_targeting(self, connection):
@@ -197,6 +199,88 @@ class ReturningTest(fixtures.TablesTest, AssertsExecutionResults):
         )
         eq_(result2.fetchall(), [(1, True), (2, False)])
 
+    @testing.fails_on(
+        "mssql",
+        "driver has unknown issue with string concatenation "
+        "in INSERT RETURNING",
+    )
+    def test_insert_returning_w_expression_one(self, connection):
+        table = self.tables.tables
+        result = connection.execute(
+            table.insert().returning(table.c.strval + "hi"),
+            {"persons": 5, "full": False, "strval": "str1"},
+        )
+
+        eq_(result.fetchall(), [("str1hi",)])
+
+        result2 = connection.execute(
+            select(table.c.id, table.c.strval).order_by(table.c.id)
+        )
+        eq_(result2.fetchall(), [(1, "str1")])
+
+    def test_insert_returning_w_type_coerce_expression(self, connection):
+        table = self.tables.tables
+        result = connection.execute(
+            table.insert().returning(type_coerce(table.c.goofy, String)),
+            {"persons": 5, "goofy": "somegoofy"},
+        )
+
+        eq_(result.fetchall(), [("FOOsomegoofy",)])
+
+        result2 = connection.execute(
+            select(table.c.id, table.c.goofy).order_by(table.c.id)
+        )
+        eq_(result2.fetchall(), [(1, "FOOsomegoofyBAR")])
+
+    def test_update_returning_w_expression_one(self, connection):
+        table = self.tables.tables
+        connection.execute(
+            table.insert(),
+            [
+                {"persons": 5, "full": False, "strval": "str1"},
+                {"persons": 3, "full": False, "strval": "str2"},
+            ],
+        )
+
+        result = connection.execute(
+            table.update()
+            .where(table.c.persons > 4)
+            .values(full=True)
+            .returning(table.c.strval + "hi")
+        )
+        eq_(result.fetchall(), [("str1hi",)])
+
+        result2 = connection.execute(
+            select(table.c.id, table.c.strval).order_by(table.c.id)
+        )
+        eq_(result2.fetchall(), [(1, "str1"), (2, "str2")])
+
+    def test_update_returning_w_type_coerce_expression(self, connection):
+        table = self.tables.tables
+        connection.execute(
+            table.insert(),
+            [
+                {"persons": 5, "goofy": "somegoofy1"},
+                {"persons": 3, "goofy": "somegoofy2"},
+            ],
+        )
+
+        result = connection.execute(
+            table.update()
+            .where(table.c.persons > 4)
+            .values(goofy="newgoofy")
+            .returning(type_coerce(table.c.goofy, String))
+        )
+        eq_(result.fetchall(), [("FOOnewgoofy",)])
+
+        result2 = connection.execute(
+            select(table.c.id, table.c.goofy).order_by(table.c.id)
+        )
+        eq_(
+            result2.fetchall(),
+            [(1, "FOOnewgoofyBAR"), (2, "FOOsomegoofy2BAR")],
+        )
+
     @testing.requires.full_returning
     def test_update_full_returning(self, connection):
         table = self.tables.tables