Oracle COLUMN_VALUE is a column name, not a keyword

Fixed issue in Oracle compiler where the syntax for
:meth:`.FunctionElement.column_valued` was incorrect, rendering the name
``COLUMN_VALUE`` without qualifying the source table correctly.

Fixes: #8945
Change-Id: Ia04bbdc68168e78b67a74bb3834a63f5d5000627

6 files changed, 58 insertions, 14 deletions

diff --git a/doc/build/changelog/unreleased_14/8827.rst b/doc/build/changelog/unreleased_14/8827.rst
index d71a30db5..b389ff771 100644
--- a/doc/build/changelog/unreleased_14/8827.rst
+++ b/doc/build/changelog/unreleased_14/8827.rst
@@ -3,8 +3,11 @@
     :tickets: 8827
     :versions: 2.0.0b4
 
-    Fixed a series of issues regarding positionally rendered bound parameters,
-    such as those used for SQLite, asyncpg, MySQL and others. Some compiled
-    forms would not maintain the order of parameters correctly, such as the
-    PostgreSQL ``regexp_replace()`` function as well as within the "nesting"
-    feature of the :class:`.CTE` construct first introduced in :ticket:`4123`.
+    Fixed a series of issues regarding the position and sometimes the identity
+    of rendered bound parameters, such as those used for SQLite, asyncpg,
+    MySQL, Oracle and others. Some compiled forms would not maintain the order
+    of parameters correctly, such as the PostgreSQL ``regexp_replace()``
+    function, the "nesting" feature of the :class:`.CTE` construct first
+    introduced in :ticket:`4123`, and selectable tables formed by using the
+    :meth:`.FunctionElement.column_valued` method with Oracle.
+
diff --git a/doc/build/changelog/unreleased_14/8945.rst b/doc/build/changelog/unreleased_14/8945.rst
new file mode 100644
index 000000000..e1b4bd693
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8945.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, oracle
+    :tickets: 8945
+    :versions: 2.0.0b5
+
+    Fixed issue in Oracle compiler where the syntax for
+    :meth:`.FunctionElement.column_valued` was incorrect, rendering the name
+    ``COLUMN_VALUE`` without qualifying the source table correctly.
diff --git a/doc/build/tutorial/data_select.rst b/doc/build/tutorial/data_select.rst
index 64b9cef8a..206171988 100644
--- a/doc/build/tutorial/data_select.rst
+++ b/doc/build/tutorial/data_select.rst
@@ -1753,7 +1753,7 @@ it is usable for custom SQL functions::
     >>> from sqlalchemy.dialects import oracle
     >>> stmt = select(func.scalar_strings(5).column_valued("s"))
     >>> print(stmt.compile(dialect=oracle.dialect()))
-    SELECT COLUMN_VALUE s
+    SELECT s.COLUMN_VALUE
     FROM TABLE (scalar_strings(:scalar_strings_1)) s
 
 
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 41b9ac43d..dc2b011af 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -859,7 +859,7 @@ class OracleCompiler(compiler.SQLCompiler):
 
     def visit_table_valued_column(self, element, **kw):
         text = super().visit_table_valued_column(element, **kw)
-        text = "COLUMN_VALUE " + text
+        text = text + ".COLUMN_VALUE"
         return text
 
     def default_from(self):
diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py
index 255efdb3c..603d54e4b 100644
--- a/test/dialect/oracle/test_compiler.py
+++ b/test/dialect/oracle/test_compiler.py
@@ -1796,19 +1796,41 @@ class TableValuedFunctionTest(fixtures.TestBase, testing.AssertsCompiledSQL):
         stmt = select(fn.alias().column)
         self.assert_compile(
             stmt,
-            "SELECT COLUMN_VALUE anon_1 "
+            "SELECT anon_1.COLUMN_VALUE "
             "FROM TABLE (scalar_strings(:scalar_strings_1)) anon_1",
         )
 
+    def test_scalar_alias_multi_columns(self):
+        fn1 = func.scalar_strings(5)
+        fn2 = func.scalar_strings(3)
+        stmt = select(fn1.alias().column, fn2.alias().column)
+        self.assert_compile(
+            stmt,
+            "SELECT anon_1.COLUMN_VALUE, anon_2.COLUMN_VALUE FROM TABLE "
+            "(scalar_strings(:scalar_strings_1)) anon_1, "
+            "TABLE (scalar_strings(:scalar_strings_2)) anon_2",
+        )
+
     def test_column_valued(self):
         fn = func.scalar_strings(5)
         stmt = select(fn.column_valued())
         self.assert_compile(
             stmt,
-            "SELECT COLUMN_VALUE anon_1 "
+            "SELECT anon_1.COLUMN_VALUE "
             "FROM TABLE (scalar_strings(:scalar_strings_1)) anon_1",
         )
 
+    def test_multi_column_valued(self):
+        fn1 = func.scalar_strings(5)
+        fn2 = func.scalar_strings(3)
+        stmt = select(fn1.column_valued(), fn2.column_valued().label("x"))
+        self.assert_compile(
+            stmt,
+            "SELECT anon_1.COLUMN_VALUE, anon_2.COLUMN_VALUE AS x FROM "
+            "TABLE (scalar_strings(:scalar_strings_1)) anon_1, "
+            "TABLE (scalar_strings(:scalar_strings_2)) anon_2",
+        )
+
     def test_table_valued(self):
         fn = func.three_pairs().table_valued("string1", "string2")
         stmt = select(fn.c.string1, fn.c.string2)
diff --git a/test/dialect/oracle/test_dialect.py b/test/dialect/oracle/test_dialect.py
index 4370992e8..38f4b176b 100644
--- a/test/dialect/oracle/test_dialect.py
+++ b/test/dialect/oracle/test_dialect.py
@@ -1,3 +1,4 @@
+import itertools
 from multiprocessing import get_context
 import re
 from unittest import mock
@@ -1113,7 +1114,7 @@ class TableValuedTest(fixtures.TestBase):
         connection.exec_driver_sql(
             r"""
 CREATE OR REPLACE FUNCTION scalar_strings (
-   count_in IN INTEGER)
+   count_in IN INTEGER, string_in IN VARCHAR2)
    RETURN strings_t
    AUTHID DEFINER
 IS
@@ -1123,7 +1124,7 @@ BEGIN
 
    FOR indx IN 1 .. count_in
    LOOP
-      l_strings (indx) := 'some string';
+      l_strings (indx) := string_in;
    END LOOP;
 
    RETURN l_strings;
@@ -1173,7 +1174,8 @@ END;
     def test_scalar_strings_control(self, scalar_strings, connection):
         result = (
             connection.exec_driver_sql(
-                "SELECT COLUMN_VALUE my_string FROM TABLE (scalar_strings (5))"
+                "SELECT COLUMN_VALUE my_string FROM TABLE "
+                "(scalar_strings (5, 'some string'))"
             )
             .scalars()
             .all()
@@ -1184,7 +1186,7 @@ END;
         result = (
             connection.exec_driver_sql(
                 "SELECT COLUMN_VALUE anon_1 "
-                "FROM TABLE (scalar_strings (5)) anon_1"
+                "FROM TABLE (scalar_strings (5, 'some string')) anon_1"
             )
             .scalars()
             .all()
@@ -1192,7 +1194,7 @@ END;
         eq_(result, ["some string"] * 5)
 
     def test_scalar_strings(self, scalar_strings, connection):
-        fn = func.scalar_strings(5)
+        fn = func.scalar_strings(5, "some string")
         result = connection.execute(select(fn.column_valued())).scalars().all()
         eq_(result, ["some string"] * 5)
 
@@ -1207,6 +1209,15 @@ END;
         result = connection.execute(select(fn.c.string1, fn.c.string2)).all()
         eq_(result, [("a", "b"), ("c", "d"), ("e", "f")])
 
+    def test_two_independent_tables(self, scalar_strings, connection):
+        fn1 = func.scalar_strings(5, "string one").column_valued()
+        fn2 = func.scalar_strings(3, "string two").column_valued()
+        result = connection.execute(select(fn1, fn2).where(fn1 != fn2)).all()
+        eq_(
+            result,
+            list(itertools.product(["string one"] * 5, ["string two"] * 3)),
+        )
+
 
 class OptimizedFetchLimitOffsetTest(test_select.FetchLimitOffsetTest):
     __only_on__ = "oracle"