[nsmallest] Need to verify the input nums is integer

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

@@ -7766,6 +7766,7 @@ def nsmallest(
         a  1
         d  2
         """
+        assert type(n) is int
         by_scols = self._prepare_sort_by_scols(columns)
         return self._sort(by=by_scols, ascending=True, na_position="last", keep=keep).head(n=n)
 

python/pyspark/pandas/groupby.py

@@ -3574,6 +3574,7 @@ def nsmallest(self, n: int = 5) -> Series:
         3  6    3
         Name: b, dtype: int64
         """
+        assert type(n) is int
         if self._psser._internal.index_level > 1:
             raise ValueError("nsmallest do not support multi-index now")
 

python/pyspark/pandas/series.py

@@ -3417,6 +3417,7 @@ def nsmallest(self, n: int = 5) -> "Series":
         2    3.0
         dtype: float64
         """
+        assert type(n) is int
         return self.sort_values(ascending=True).head(n)
 
     def nlargest(self, n: int = 5) -> "Series":

python/pyspark/pandas/tests/test_dataframe.py

@@ -1907,6 +1907,10 @@ def test_nsmallest(self):
         msg = 'keep must be either "first", "last" or "all".'
         with self.assertRaisesRegex(ValueError, msg):
             psdf.nlargest(5, columns=["c"], keep="xx")
+        with self.assertRaises(AssertionError):
+            psdf.nsmallest("test", columns=["c"], keep="last")
+        with self.assertRaises(AssertionError):
+            psdf.nsmallest(0.1, columns=["c"], keep="last")
 
     def test_xs(self):
         d = {

python/pyspark/pandas/tests/test_groupby.py

@@ -1765,6 +1765,10 @@ def test_nsmallest(self):
         )
         with self.assertRaisesRegex(ValueError, "nsmallest do not support multi-index now"):
             psdf.set_index(["a", "b"]).groupby(["c"])["d"].nsmallest(1)
+        with self.assertRaises(AssertionError):
+            psdf.groupby(["a"])["b"].nsmallest(0.1).sort_index()
+        with self.assertRaises(AssertionError):
+            psdf.groupby(["a"])["b"].nsmallest(False).sort_index()
 
     def test_nlargest(self):
         pdf = pd.DataFrame(

python/pyspark/pandas/tests/test_series.py

@@ -877,6 +877,11 @@ def test_nsmallest(self):
         self.assert_eq(psser.nsmallest(), pser.nsmallest())
         self.assert_eq((psser + 1).nsmallest(), (pser + 1).nsmallest())
 
+        with self.assertRaises(AssertionError):
+            psser.nsmallest("String")
+        with self.assertRaises(AssertionError):
+            psser.nsmallest(0.1)
+
     def test_nlargest(self):
         sample_lst = [1, 2, 3, 4, np.nan, 6]
         pser = pd.Series(sample_lst, name="x")

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
     }
   }