[SPARK-33538][SQL] Directly push IN/NOT predicates to the Hive Metastore

### What changes were proposed in this pull request?

This PR refactors the way IN and NOT IN predicates are converted in `HiveShim`. Previously, these predicates were expanded into chains of OR/AND comparisons (e.g., `a = 1` or `a = 2`). This change updates the logic to use SQL-native IN and NOT IN syntax (e.g., `a in (1, 2)`), improving readability and possibly query performance.

### Why are the changes needed?

Efficiency:
Many SQL engines optimize IN/NOT IN natively, which can improve performance and reduce query complexity.
Maintainability:
Simplifies predicate generation logic, making the codebase easier to maintain and extend.

### Does this PR introduce _any_ user-facing change?

No user-facing changes; this is an internal refactor that affects query generation logic.

### How was this patch tested?

Existing unit tests for predicate conversion and query generation should be sufficient to cover the updated logic.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #51132 from yuexing/SPARK-33538-hive.

Lead-authored-by: xingyue <[email protected]>
Co-authored-by: Yue <[email protected]>
Signed-off-by: yangjie01 <[email protected]>

Author: yuexing

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala

@@ -749,12 +749,12 @@ private[client] class Shim_v2_0 extends Shim with Logging {
       }
     }
 
-    def convertInToOr(name: String, values: Seq[String]): String = {
-      values.map(value => s"$name = $value").mkString("(", " or ", ")")
+    def convertIn(name: String, values: Seq[String]): String = {
+      s"($name) in (${values.mkString(", ")})"
     }
 
-    def convertNotInToAnd(name: String, values: Seq[String]): String = {
-      values.map(value => s"$name != $value").mkString("(", " and ", ")")
+    def convertNotIn(name: String, values: Seq[String]): String = {
+      s"($name) not in (${values.mkString(", ")})"
     }
 
     def hasNullLiteral(list: Seq[Expression]): Boolean = list.exists {
@@ -786,11 +786,11 @@ private[client] class Shim_v2_0 extends Shim with Logging {
 
       case In(ExtractAttribute(SupportedAttribute(name)), ExtractableLiterals(values))
           if useAdvanced =>
-        Some(convertInToOr(name, values))
+        Some(convertIn(name, values))
 
       case Not(In(ExtractAttribute(SupportedAttribute(name)), ExtractableLiterals(values)))
           if useAdvanced =>
-        Some(convertNotInToAnd(name, values))
+        Some(convertNotIn(name, values))
 
       case InSet(child, values) if useAdvanced && values.size > inSetThreshold =>
         val dataType = child.dataType
@@ -802,19 +802,19 @@ private[client] class Shim_v2_0 extends Shim with Logging {
 
       case InSet(child @ ExtractAttribute(SupportedAttribute(name)), ExtractableDateValues(values))
           if useAdvanced && child.dataType == DateType =>
-        Some(convertInToOr(name, values))
+        Some(convertIn(name, values))
 
       case Not(InSet(child @ ExtractAttribute(SupportedAttribute(name)),
         ExtractableDateValues(values))) if useAdvanced && child.dataType == DateType =>
-        Some(convertNotInToAnd(name, values))
+        Some(convertNotIn(name, values))
 
       case InSet(ExtractAttribute(SupportedAttribute(name)), ExtractableValues(values))
           if useAdvanced =>
-        Some(convertInToOr(name, values))
+        Some(convertIn(name, values))
 
       case Not(InSet(ExtractAttribute(SupportedAttribute(name)), ExtractableValues(values)))
           if useAdvanced =>
-        Some(convertNotInToAnd(name, values))
+        Some(convertNotIn(name, values))
 
       case op @ SpecialBinaryComparison(
           ExtractAttribute(SupportedAttribute(name)), ExtractableLiteral(value)) =>

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/FiltersSuite.scala

@@ -71,7 +71,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
   filterTest("date filter with IN predicate",
     (a("datecol", DateType) in
       (Literal(Date.valueOf("2019-01-01")), Literal(Date.valueOf("2019-01-07")))) :: Nil,
-    "(datecol = \"2019-01-01\" or datecol = \"2019-01-07\")")
+    "(datecol) in (\"2019-01-01\", \"2019-01-07\")")
 
   filterTest("date and string filter",
     (Literal(Date.valueOf("2019-01-01")) === a("datecol", DateType)) ::
@@ -84,7 +84,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
 
   filterTest("string filter with InSet predicate",
     InSet(a("strcol", StringType), Set("1", "2").map(s => UTF8String.fromString(s))) :: Nil,
-    "(strcol = \"1\" or strcol = \"2\")")
+    "(strcol) in (\"1\", \"2\")")
 
   filterTest("skip varchar",
     (Literal("") === a("varchar", StringType)) :: Nil,
@@ -97,7 +97,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
 
   filterTest("SPARK-24879 null literals should be ignored for IN constructs",
     (a("intcol", IntegerType) in (Literal(1), Literal(null))) :: Nil,
-    "(intcol = 1)")
+    "(intcol) in (1)")
 
   filterTest("NOT: int and string filters",
     (a("intcol", IntegerType) =!= Literal(1)) :: (Literal("a") =!= a("strcol", IntegerType)) :: Nil,
@@ -109,7 +109,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
 
   filterTest("not-in, string filter",
     (Not(In(a("strcol", StringType), Seq(Literal("a"), Literal("b"))))) :: Nil,
-    """(strcol != "a" and strcol != "b")""")
+    """(strcol) not in ("a", "b")""")
 
   filterTest("not-in, string filter with null",
     (Not(In(a("strcol", StringType), Seq(Literal("a"), Literal("b"), Literal(null))))) :: Nil,
@@ -118,7 +118,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
   filterTest("not-in, date filter",
     (Not(In(a("datecol", DateType),
       Seq(Literal(Date.valueOf("2021-01-01")), Literal(Date.valueOf("2021-01-02")))))) :: Nil,
-    """(datecol != "2021-01-01" and datecol != "2021-01-02")""")
+    """(datecol) not in ("2021-01-01", "2021-01-02")""")
 
   filterTest("not-in, date filter with null",
     (Not(In(a("datecol", DateType),
@@ -128,7 +128,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
 
   filterTest("not-inset, string filter",
     (Not(InSet(a("strcol", StringType), Set(Literal("a").eval(), Literal("b").eval())))) :: Nil,
-    """(strcol != "a" and strcol != "b")""")
+    """(strcol) not in ("a", "b")""")
 
   filterTest("not-inset, string filter with null",
     (Not(InSet(a("strcol", StringType),
@@ -139,7 +139,7 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
     (Not(InSet(a("datecol", DateType),
       Set(Literal(Date.valueOf("2020-01-01")).eval(),
         Literal(Date.valueOf("2020-01-02")).eval())))) :: Nil,
-    """(datecol != "2020-01-01" and datecol != "2020-01-02")""")
+    """(datecol) not in ("2020-01-01", "2020-01-02")""")
 
   filterTest("not-inset, date filter with null",
     (Not(InSet(a("datecol", DateType),
@@ -239,14 +239,14 @@ class FiltersSuite extends SparkFunSuite with PlanTest {
     withSQLConf(SQLConf.HIVE_METASTORE_PARTITION_PRUNING_INSET_THRESHOLD.key -> "3") {
       val intFilter = InSet(a("p", IntegerType), Set(null, 1, 2))
       val intConverted = shim.convertFilters(testTable, Seq(intFilter))
-      assert(intConverted == "(p = 1 or p = 2)")
+      assert(intConverted == "(p) in (1, 2)")
     }
 
     withSQLConf(SQLConf.HIVE_METASTORE_PARTITION_PRUNING_INSET_THRESHOLD.key -> "3") {
       val dateFilter = InSet(a("p", DateType), Set(null,
         Literal(Date.valueOf("2020-01-01")).eval(), Literal(Date.valueOf("2021-01-01")).eval()))
       val dateConverted = shim.convertFilters(testTable, Seq(dateFilter))
-      assert(dateConverted == "(p = \"2020-01-01\" or p = \"2021-01-01\")")
+      assert(dateConverted == "(p) in (\"2020-01-01\", \"2021-01-01\")")
     }
   }