add joins_implicitly to column_valued()

Added parameter
:paramref:`.FunctionElement.column_valued.joins_implicitly`, which is
useful in preventing the "cartesian product" warning when making use of
table-valued or column-valued functions. This parameter was already
introduced for :meth:`.FunctionElement.table_valued` in :ticket:`7845`,
however it failed to be added for :meth:`.FunctionElement.column_valued`
as well.

Fixes: #9009
Change-Id: Ifb72fbcb4f4d2998e730d6f85ec7280df3bf3d47

3 files changed, 61 insertions, 19 deletions

diff --git a/doc/build/changelog/unreleased_14/9009.rst b/doc/build/changelog/unreleased_14/9009.rst
new file mode 100644
index 000000000..9520b3e13
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/9009.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, sql
+    :tickets: 9009
+    :versions: 2.0.0b5
+
+    Added parameter
+    :paramref:`.FunctionElement.column_valued.joins_implicitly`, which is
+    useful in preventing the "cartesian product" warning when making use of
+    table-valued or column-valued functions. This parameter was already
+    introduced for :meth:`.FunctionElement.table_valued` in :ticket:`7845`,
+    however it failed to be added for :meth:`.FunctionElement.column_valued`
+    as well.
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 5ed89bc82..742f211c9 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -292,7 +292,7 @@ class FunctionElement(Executable, ColumnElement[_T], FromClause, Generative):
 
         return new_func.alias(name=name, joins_implicitly=joins_implicitly)
 
-    def column_valued(self, name=None):
+    def column_valued(self, name=None, joins_implicitly=False):
         """Return this :class:`_functions.FunctionElement` as a column expression that
         selects from itself as a FROM clause.
 
@@ -308,6 +308,16 @@ class FunctionElement(Executable, ColumnElement[_T], FromClause, Generative):
 
             gs = func.generate_series(1, 5, -1).alias().column
 
+        :param name: optional name to assign to the alias name that's generated.
+         If omitted, a unique anonymizing name is used.
+
+        :param joins_implicitly: when True, the "table" portion of the column
+         valued function may be a member of the FROM clause without any
+         explicit JOIN to other tables in the SQL query, and no "cartesian
+         product" warning will be generated. May be useful for SQL functions
+         such as ``func.json_array_elements()``.
+
+         .. versionadded:: 1.4.46
 
         .. seealso::
 
@@ -319,7 +329,7 @@ class FunctionElement(Executable, ColumnElement[_T], FromClause, Generative):
 
         """  # noqa: 501
 
-        return self.alias(name=name).column
+        return self.alias(name=name, joins_implicitly=joins_implicitly).column
 
     @util.ro_non_memoized_property
     def columns(self):
diff --git a/test/sql/test_from_linter.py b/test/sql/test_from_linter.py
index 1fa3aff36..49370b1e6 100644
--- a/test/sql/test_from_linter.py
+++ b/test/sql/test_from_linter.py
@@ -165,16 +165,16 @@ class TestFindUnmatchingFroms(fixtures.TablesTest):
         assert start is p3
         assert froms == {p1}
 
-    @testing.combinations(
-        "render_derived", "alias", None, argnames="additional_transformation"
+    @testing.variation("additional_transformation", ["alias", "none"])
+    @testing.variation("joins_implicitly", [True, False])
+    @testing.variation(
+        "type_", ["table_valued", "table_valued_derived", "column_valued"]
     )
-    @testing.combinations(True, False, argnames="joins_implicitly")
-    def test_table_valued(
-        self,
-        joins_implicitly,
-        additional_transformation,
+    def test_fn_valued(
+        self, joins_implicitly, additional_transformation, type_
     ):
-        """test #7845"""
+        """test #7845, #9009"""
+
         my_table = table(
             "tbl",
             column("id", Integer),
@@ -183,25 +183,45 @@ class TestFindUnmatchingFroms(fixtures.TablesTest):
 
         sub_dict = my_table.c.data["d"]
 
-        tv = func.json_each(sub_dict)
+        if type_.table_valued or type_.table_valued_derived:
+            tv = func.json_each(sub_dict)
+
+            tv = tv.table_valued("key", joins_implicitly=joins_implicitly)
+
+            if type_.table_valued_derived:
+                tv = tv.render_derived(name="tv", with_types=True)
+
+            if additional_transformation.alias:
+                tv = tv.alias()
+
+            has_key = tv.c.key == "f"
+            stmt = select(my_table.c.id).where(has_key)
+        elif type_.column_valued:
+            tv = func.json_array_elements(sub_dict)
 
-        tv = tv.table_valued("key", joins_implicitly=joins_implicitly)
+            if additional_transformation.alias:
+                tv = tv.alias(joins_implicitly=joins_implicitly).column
+            else:
+                tv = tv.column_valued("key", joins_implicitly=joins_implicitly)
 
-        if additional_transformation == "render_derived":
-            tv = tv.render_derived(name="tv", with_types=True)
-        elif additional_transformation == "alias":
-            tv = tv.alias()
+            stmt = select(my_table.c.id, tv)
+        else:
+            type_.fail()
 
-        has_key = tv.c.key == "f"
-        stmt = select(my_table.c.id).where(has_key)
         froms, start = find_unmatching_froms(stmt, my_table)
 
         if joins_implicitly:
             is_(start, None)
             is_(froms, None)
-        else:
+        elif type_.column_valued:
+            assert start == my_table
+            assert froms == {tv.scalar_alias}
+
+        elif type_.table_valued or type_.table_valued_derived:
             assert start == my_table
             assert froms == {tv}
+        else:
+            type_.fail()
 
     def test_count_non_eq_comparison_operators(self):
         query = select(self.a).where(self.a.c.col_a > self.b.c.col_b)