Add testing and include PySparkType

Author: jhaberstroh-sharethis

python/pyspark/sql/tests/test_dataframe.py

@@ -1077,6 +1077,17 @@ def test_join_without_on(self):
             expected = [Row(a=0, b=0)]
             self.assertEqual(actual, expected)
 
+    def test_join_on_types(self):
+        df1 = self.spark.range(1).toDF("a").withColumn('key', lit(1))
+        df2 = self.spark.range(1).toDF("b").withColumn('key', lit(1))
+        expected_result = df1.join(df2, on='key')
+
+        self.assertEqual(expected_result, df1.join(df2, on=['key']))
+        self.assertEqual(expected_result, df1.join(df2, on=('key',)))
+        self.assertEqual(expected_result, df1.join(df2, on=col('key')))
+        self.assertEqual(expected_result, df1.join(df2, on=[col('key')]))
+        self.assertEqual(expected_result, df1.join(df2, on=(col('key'),)))
+
     # Regression test for invalid join methods when on is None, Spark-14761
     def test_invalid_join_method(self):
         df1 = self.spark.createDataFrame([("Alice", 5), ("Bob", 8)], ["name", "age"])

python/pyspark/sql/tests/test_utils.py

@@ -24,8 +24,11 @@
     SparkUpgradeException,
 )
 from pyspark.testing.sqlutils import ReusedSQLTestCase
-from pyspark.sql.functions import to_date, unix_timestamp, from_unixtime
-
+from pyspark.sql.functions import col, to_date, unix_timestamp, from_unixtime
+from pyspark.sql.utils import (
+    isinstance_iterable,
+    to_list_column_style,
+)
 
 class UtilsTests(ReusedSQLTestCase):
     def test_capture_analysis_exception(self):
@@ -75,6 +78,32 @@ def test_get_error_class_state(self):
             self.assertEquals(e.getErrorClass(), "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION")
             self.assertEquals(e.getSqlState(), "42703")
 
+    def test_isinstance_iterable(self):
+        a_col = col("")
+        a_str = ""
+
+        self.assertFalse(isinstance_iterable(a_col))
+        self.assertTrue(isinstance_iterable([a_col]))
+        self.assertTrue(isinstance_iterable((a_col,)))
+
+        self.assertFalse(isinstance_iterable(a_str))
+        self.assertTrue(isinstance_iterable([a_str]))
+        self.assertTrue(isinstance_iterable((a_str,)))
+
+        self.assertIsInstance(to_list_column_style(a_col), list)
+        self.assertIsInstance(to_list_column_style([a_col]), list)
+        self.assertIsInstance(to_list_column_style((a_col,)), list)
+        self.assertFalse(isinstance_iterable(to_list_column_style(a_col)[0]))
+        self.assertFalse(isinstance_iterable(to_list_column_style([a_col])[0]))
+        self.assertFalse(isinstance_iterable(to_list_column_style((a_col,))[0]))
+
+        self.assertIsInstance(to_list_column_style(a_str), list)
+        self.assertIsInstance(to_list_column_style([a_str]), list)
+        self.assertIsInstance(to_list_column_style((a_str,)), list)
+        self.assertFalse(isinstance_iterable(to_list_column_style(a_str)[0]))
+        self.assertFalse(isinstance_iterable(to_list_column_style([a_str])[0]))
+        self.assertFalse(isinstance_iterable(to_list_column_style((a_str,))[0]))
+
 
 if __name__ == "__main__":
     import unittest

python/pyspark/sql/utils.py

@@ -294,7 +294,7 @@ def isinstance_iterable(obj: Any) -> bool:
     try:
         iter(obj)
         return True
-    except TypeError, PySparkTypeError:
+    except (TypeError, PySparkTypeError):
         return False
 
 def to_list_column_style(