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

2 files changed, 40 insertions, 7 deletions

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"