get rid of kwargs; support lsuffix & rsuffix

Author: sfc-gh-mayliu

src/snowflake/snowpark/dataframe.py

@@ -3528,8 +3528,10 @@ def lateral_join(
         right: "DataFrame",
         condition: Optional[Column] = None,
         how: Optional[str] = None,
+        *,
+        lsuffix: str = "",
+        rsuffix: str = "",
         _emit_ast: bool = True,
-        **kwargs,
     ) -> "DataFrame":
         """Performs a lateral join of the specified type (``how``) with the
         current DataFrame and another DataFrame (``right``).
@@ -3545,9 +3547,12 @@ def lateral_join(
                 - Left outer join: "left", "leftouter"
                 - Cross join: "cross"
 
-                You can also use ``join_type`` keyword to specify this condition.
-                Note that to avoid breaking changes, currently when ``join_type`` is specified,
-                it overrides ``how``.
+            lsuffix: Suffix to add to the overlapping columns of the left DataFrame.
+            rsuffix: Suffix to add to the overlapping columns of the right DataFrame.
+
+        Note:
+            When both ``lsuffix`` and ``rsuffix`` are empty, the overlapping columns will have random column names in the resulting DataFrame.
+            You can reference to these randomly named columns using :meth:`Column.alias`.
 
         Examples::
             >>> df1 = session.create_dataframe([[1, 2], [3, 4], [5, 6]], schema=["a", "b"])
@@ -3571,10 +3576,21 @@ def lateral_join(
             |5      |NULL   |6    |NULL  |
             ------------------------------
             <BLANKLINE>
+
+            >>> df1.lateral_join(df2, df1.b * 2 > df2.c, how="left", lsuffix="_l", rsuffix="_r").show()
+            ------------------------------
+            |"A_L"  |"B"  |"A_R"  |"C"   |
+            ------------------------------
+            |3      |4    |1      |7     |
+            |5      |6    |1      |7     |
+            |5      |6    |3      |8     |
+            |1      |2    |NULL   |NULL  |
+            ------------------------------
+            <BLANKLINE>
         """
-        join_type = create_join_type(kwargs.get("join_type") or how or "inner")
+        join_type = create_join_type(how or "inner")
         (lhs, rhs) = _disambiguate(
-            self, right, LateralJoin(join_type), [], lsuffix="", rsuffix=""
+            self, right, LateralJoin(join_type), [], lsuffix=lsuffix, rsuffix=rsuffix
         )
 
         condition_expr = condition._expression if condition is not None else None

tests/integ/test_dataframe.py

@@ -5653,7 +5653,7 @@ def test_df_join_how_on_overwrite(session):
     "config.getoption('local_testing_mode', default=False)",
     reason="Lateral join is not supported in Local Testing",
 )
-def test_lateral_join_how_on_overwrite(session):
+def test_lateral_join(session):
     table1 = Utils.random_table_name()
     table2 = Utils.random_table_name()
 
@@ -5666,17 +5666,16 @@ def test_lateral_join_how_on_overwrite(session):
         df1 = session.table(table1)
         df2 = session.table(table2)
 
-        # Test join_type overwriting how parameter
-        df_inner = df1.lateral_join(df2, df1.a == df2.a, how="left", join_type="inner")
-        df_left = df1.lateral_join(df2, df1.a == df2.a, how="inner", join_type="left")
+        df_inner = df1.lateral_join(df2, df1.a == df2.a, how="inner")
+        df_left = df1.lateral_join(df2, df1.a == df2.a, how="left")
 
-        # Compare with equivalent SQL - inner lateral join
+        # Test inner lateral join vs SQL
         sql_inner = session.sql(
             f"SELECT * FROM {table1}, LATERAL (SELECT * FROM {table2} WHERE {table1}.a = {table2}.a)"
         )
         Utils.check_answer(df_inner, sql_inner)
 
-        # Compare with equivalent SQL - left outer lateral join
+        # Test left outer lateral join vs SQL
         sql_left = session.sql(
             f"SELECT * FROM {table1} LEFT OUTER JOIN LATERAL (SELECT * FROM {table2} WHERE {table1}.a = {table2}.a) ON TRUE"
         )

tests/integ/test_simplifier_suite.py

@@ -1132,7 +1132,7 @@ def test_lateral_join_simplifier(session, use_simplified_query_generation):
         Utils.check_answer(df, [Row(3, 4, 3, 8)])
 
         df_chained = (
-            df1.lateral_join(df2, df1.id == df2.id)
+            df1.lateral_join(df2, df1.id == df2.id, lsuffix="_l", rsuffix="_r")
             .select("value", "amount")
             .filter(col("value") > 1)
             .sort("value")