[prefix/suffix] accept all inputs which implement __str__

pom.xml

@@ -128,7 +128,7 @@
     <!-- Version used for internal directory structure -->
     <hive.version.short>2.3</hive.version.short>
     <!-- note that this should be compatible with Kafka brokers version 0.10 and up -->
-    <kafka.version>3.2.0</kafka.version>
+    <kafka.version>3.2.1</kafka.version>
     <!-- After 10.15.1.3, the minimum required version is JDK9 -->
     <derby.version>10.14.2.0</derby.version>
     <parquet.version>1.12.3</parquet.version>

python/pyspark/pandas/frame.py

@@ -9012,9 +9012,9 @@ def add_prefix(self, prefix: str) -> "DataFrame":
         2      3      5
         3      4      6
         """
-        assert isinstance(prefix, str)
+        f = partial("{prefix}{}".format, prefix=prefix)
         return self._apply_series_op(
-            lambda psser: psser.rename(tuple([prefix + i for i in psser._column_label]))
+            lambda psser: psser.rename(tuple([f(i) for i in psser._column_label]))
         )
 
     def add_suffix(self, suffix: str) -> "DataFrame":
@@ -9057,9 +9057,9 @@ def add_suffix(self, suffix: str) -> "DataFrame":
         2      3      5
         3      4      6
         """
-        assert isinstance(suffix, str)
+        f = partial("{}{suffix}".format, suffix=suffix)
         return self._apply_series_op(
-            lambda psser: psser.rename(tuple([i + suffix for i in psser._column_label]))
+            lambda psser: psser.rename(tuple([f(i) for i in psser._column_label]))
         )
 
     # TODO: include, and exclude should be implemented.

python/pyspark/pandas/series.py

@@ -3170,11 +3170,10 @@ def add_prefix(self, prefix: str) -> "Series":
         item_3    4
         dtype: int64
         """
-        assert isinstance(prefix, str)
         internal = self._internal.resolved_copy
         sdf = internal.spark_frame.select(
             [
-                F.concat(SF.lit(prefix), index_spark_column).alias(index_spark_column_name)
+                F.concat(SF.lit(str(prefix)), index_spark_column).alias(index_spark_column_name)
                 for index_spark_column, index_spark_column_name in zip(
                     internal.index_spark_columns, internal.index_spark_column_names
                 )
@@ -3225,11 +3224,10 @@ def add_suffix(self, suffix: str) -> "Series":
         3_item    4
         dtype: int64
         """
-        assert isinstance(suffix, str)
         internal = self._internal.resolved_copy
         sdf = internal.spark_frame.select(
             [
-                F.concat(index_spark_column, SF.lit(suffix)).alias(index_spark_column_name)
+                F.concat(index_spark_column, SF.lit(str(suffix))).alias(index_spark_column_name)
                 for index_spark_column, index_spark_column_name in zip(
                     internal.index_spark_columns, internal.index_spark_column_names
                 )

python/pyspark/pandas/tests/test_dataframe.py

@@ -2701,6 +2701,8 @@ def test_add_prefix(self):
         pdf = pd.DataFrame({"A": [1, 2, 3, 4], "B": [3, 4, 5, 6]}, index=np.random.rand(4))
         psdf = ps.from_pandas(pdf)
         self.assert_eq(pdf.add_prefix("col_"), psdf.add_prefix("col_"))
+        self.assert_eq(pdf.add_prefix(1.1), psdf.add_prefix(1.1))
+        self.assert_eq(pdf.add_prefix(True), psdf.add_prefix(True))
 
         columns = pd.MultiIndex.from_tuples([("X", "A"), ("X", "B")])
         pdf.columns = columns
@@ -2711,6 +2713,8 @@ def test_add_suffix(self):
         pdf = pd.DataFrame({"A": [1, 2, 3, 4], "B": [3, 4, 5, 6]}, index=np.random.rand(4))
         psdf = ps.from_pandas(pdf)
         self.assert_eq(pdf.add_suffix("first_series"), psdf.add_suffix("first_series"))
+        self.assert_eq(pdf.add_suffix(1.1), psdf.add_suffix(1.1))
+        self.assert_eq(pdf.add_suffix(True), psdf.add_suffix(True))
 
         columns = pd.MultiIndex.from_tuples([("X", "A"), ("X", "B")])
         pdf.columns = columns

python/pyspark/pandas/tests/test_series.py

@@ -1293,6 +1293,8 @@ def test_add_prefix(self):
         pser = pd.Series([1, 2, 3, 4], name="0")
         psser = ps.from_pandas(pser)
         self.assert_eq(pser.add_prefix("item_"), psser.add_prefix("item_"))
+        self.assert_eq(pser.add_prefix(1.1), psser.add_prefix(1.1))
+        self.assert_eq(pser.add_prefix(False), psser.add_prefix(False))
 
         pser = pd.Series(
             [1, 2, 3],
@@ -1306,6 +1308,8 @@ def test_add_suffix(self):
         pser = pd.Series([1, 2, 3, 4], name="0")
         psser = ps.from_pandas(pser)
         self.assert_eq(pser.add_suffix("_item"), psser.add_suffix("_item"))
+        self.assert_eq(pser.add_suffix(1.1), psser.add_suffix(1.1))
+        self.assert_eq(pser.add_suffix(False), psser.add_suffix(False))
 
         pser = pd.Series(
             [1, 2, 3],

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PredicateUtils.scala

@@ -19,14 +19,25 @@ package org.apache.spark.sql.internal.connector
 
 import org.apache.spark.sql.catalyst.CatalystTypeConverters
 import org.apache.spark.sql.connector.expressions.{LiteralValue, NamedReference}
-import org.apache.spark.sql.connector.expressions.filter.Predicate
-import org.apache.spark.sql.sources.{Filter, In}
+import org.apache.spark.sql.connector.expressions.filter.{And => V2And, Not => V2Not, Or => V2Or, Predicate}
+import org.apache.spark.sql.sources.{AlwaysFalse, AlwaysTrue, And, EqualNullSafe, EqualTo, Filter, GreaterThan, GreaterThanOrEqual, In, IsNotNull, IsNull, LessThan, LessThanOrEqual, Not, Or, StringContains, StringEndsWith, StringStartsWith}
+import org.apache.spark.sql.types.StringType
 
 private[sql] object PredicateUtils {
 
   def toV1(predicate: Predicate): Option[Filter] = {
+
+    def isValidBinaryPredicate(): Boolean = {
+      if (predicate.children().length == 2 &&
+        predicate.children()(0).isInstanceOf[NamedReference] &&
+        predicate.children()(1).isInstanceOf[LiteralValue[_]]) {
+        true
+      } else {
+        false
+      }
+    }
+
     predicate.name() match {
-      // TODO: add conversion for other V2 Predicate
       case "IN" if predicate.children()(0).isInstanceOf[NamedReference] =>
         val attribute = predicate.children()(0).toString
         val values = predicate.children().drop(1)
@@ -43,6 +54,81 @@ private[sql] object PredicateUtils {
           Some(In(attribute, Array.empty[Any]))
         }
 
+      case "=" | "<=>" | ">" | "<" | ">=" | "<=" if isValidBinaryPredicate =>
+        val attribute = predicate.children()(0).toString
+        val value = predicate.children()(1).asInstanceOf[LiteralValue[_]]
+        val v1Value = CatalystTypeConverters.convertToScala(value.value, value.dataType)
+        val v1Filter = predicate.name() match {
+          case "=" => EqualTo(attribute, v1Value)
+          case "<=>" => EqualNullSafe(attribute, v1Value)
+          case ">" => GreaterThan(attribute, v1Value)
+          case ">=" => GreaterThanOrEqual(attribute, v1Value)
+          case "<" => LessThan(attribute, v1Value)
+          case "<=" => LessThanOrEqual(attribute, v1Value)
+        }
+        Some(v1Filter)
+
+      case "IS_NULL" | "IS_NOT_NULL" if predicate.children().length == 1 &&
+          predicate.children()(0).isInstanceOf[NamedReference] =>
+        val attribute = predicate.children()(0).toString
+        val v1Filter = predicate.name() match {
+          case "IS_NULL" => IsNull(attribute)
+          case "IS_NOT_NULL" => IsNotNull(attribute)
+        }
+        Some(v1Filter)
+
+      case "STARTS_WITH" | "ENDS_WITH" | "CONTAINS" if isValidBinaryPredicate =>
+        val attribute = predicate.children()(0).toString
+        val value = predicate.children()(1).asInstanceOf[LiteralValue[_]]
+        if (!value.dataType.sameType(StringType)) return None
+        val v1Value = value.value.toString
+        val v1Filter = predicate.name() match {
+          case "STARTS_WITH" =>
+            StringStartsWith(attribute, v1Value)
+          case "ENDS_WITH" =>
+            StringEndsWith(attribute, v1Value)
+          case "CONTAINS" =>
+            StringContains(attribute, v1Value)
+        }
+        Some(v1Filter)
+
+      case "ALWAYS_TRUE" | "ALWAYS_FALSE" if predicate.children().isEmpty =>
+        val v1Filter = predicate.name() match {
+          case "ALWAYS_TRUE" => AlwaysTrue()
+          case "ALWAYS_FALSE" => AlwaysFalse()
+        }
+        Some(v1Filter)
+
+      case "AND" =>
+        val and = predicate.asInstanceOf[V2And]
+        val left = toV1(and.left())
+        val right = toV1(and.right())
+        if (left.nonEmpty && right.nonEmpty) {
+          Some(And(left.get, right.get))
+        } else {
+          None
+        }
+
+      case "OR" =>
+        val or = predicate.asInstanceOf[V2Or]
+        val left = toV1(or.left())
+        val right = toV1(or.right())
+        if (left.nonEmpty && right.nonEmpty) {
+          Some(Or(left.get, right.get))
+        } else if (left.nonEmpty) {
+          left
+        } else {
+          right
+        }
+
+      case "NOT" =>
+        val child = toV1(predicate.asInstanceOf[V2Not].child())
+        if (child.nonEmpty) {
+          Some(Not(child.get))
+        } else {
+          None
+        }
+
       case _ => None
     }
   }