[SPARK-59250][SQL] Translate an empty runtime IN filter to AlwaysFalse in DSv2 runtime filtering

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala

@@ -848,6 +848,10 @@ abstract class InMemoryBaseTable(
     }
 
     override def filter(filters: Array[Filter]): Unit = {
+      if (filters.exists(_.isInstanceOf[AlwaysFalse])) {
+        this.data = Seq.empty
+        return
+      }
       if (partitioning.length == 1 && identityPartitionReferences.length == 1) {
         val ref = identityPartitionReferences.head
         filters.foreach {

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala

@@ -24,7 +24,7 @@ import org.scalatest.Assertions.assert
 import org.apache.spark.sql.connector.catalog.constraints.Constraint
 import org.apache.spark.sql.connector.distributions.{Distribution, Distributions}
 import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue, NamedReference, SortOrder, Transform}
-import org.apache.spark.sql.connector.expressions.filter.{And, Predicate}
+import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, And, Predicate}
 import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder, SupportsRuntimeV2Filtering}
 import org.apache.spark.sql.connector.write.{LogicalWriteInfo, SupportsOverwriteV2, WriteBuilder, WriterCommitMessage}
 import org.apache.spark.sql.types.StructType
@@ -81,6 +81,10 @@ class InMemoryTableWithV2Filter(
     }
 
     override def filter(filters: Array[Predicate]): Unit = {
+      if (filters.exists(_.isInstanceOf[AlwaysFalse])) {
+        data = Seq.empty
+        return
+      }
       if (partitioning.length == 1 && identityPartitionReferences.length == 1) {
         val ref = identityPartitionReferences.head
         filters.foreach {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

@@ -38,7 +38,7 @@ import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Dependency, Depend
 import org.apache.spark.sql.connector.catalog.TableChange
 import org.apache.spark.sql.connector.catalog.index.SupportsIndex
 import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue}
-import org.apache.spark.sql.connector.expressions.filter.{And => V2And, Not => V2Not, Or => V2Or, Predicate}
+import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, And => V2And, Not => V2Not, Or => V2Or, Predicate}
 import org.apache.spark.sql.connector.read.LocalScan
 import org.apache.spark.sql.connector.read.streaming.{ContinuousStream, MicroBatchStream, SupportsRealTimeMode}
 import org.apache.spark.sql.connector.write.{V1Write, Write}
@@ -975,6 +975,8 @@ private[sql] object DataSourceV2Strategy extends Logging {
     case TrueLiteral => None
     case in: InSubqueryExec if in.isResultUnavailable =>
       None
+    case in: InSubqueryExec if in.values().exists(_.isEmpty) =>
+      Some(new AlwaysFalse())
     case in @ InSubqueryExec(PushableColumnAndNestedColumn(name), _, _, _, _, _) =>
       val values = in.values().getOrElse {
         throw SparkException.internalError(

sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala

@@ -2659,12 +2659,76 @@ class DynamicPartitionPruningV1SuiteAEOn extends DynamicPartitionPruningV1Suite
 }
 
 abstract class DynamicPartitionPruningV2Suite extends DynamicPartitionPruningDataSourceSuiteBase {
+  import testImplicits._
+
   override protected def runAnalyzeColumnCommands: Boolean = false
 
   override protected def initState(): Unit = {
     spark.conf.set("spark.sql.catalog.testcat", classOf[InMemoryTableCatalog].getName)
     spark.conf.set("spark.sql.defaultCatalog", "testcat")
   }
+
+  private def collectFactScan(df: DataFrame): BatchScanExec = {
+    val scans = collectWithSubqueries(df.queryExecution.executedPlan) {
+      case b: BatchScanExec if b.runtimeFilters.nonEmpty => b
+    }
+    assert(scans.size == 1, s"expected exactly 1 scan with runtime filters, got:\n$scans")
+    scans.head
+  }
+
+  test("SPARK-59250: DPP prunes all DSv2 partitions when the runtime IN filter is empty",
+    DisableAdaptiveExecution("an empty build side collapses the join under AQE")) {
+    withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+      SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+      // no dim rows match, so the runtime filter degenerates into store_id IN ()
+      val df = sql(
+        """
+          |SELECT f.date_id, f.store_id FROM fact_sk f
+          |JOIN dim_store s ON f.store_id = s.store_id AND s.country = 'XX'
+        """.stripMargin)
+
+      checkPartitionPruningPredicate(df, withSubquery = true, withBroadcast = false)
+      checkAnswer(df, Nil)
+
+      val scan = collectFactScan(df)
+      assert(scan.filteredPartitions.flatten.isEmpty,
+        s"expected all partitions pruned by the empty runtime filter, " +
+          s"got ${scan.filteredPartitions.flatten.size} of ${scan.inputPartitions.size}")
+    }
+  }
+
+  test("SPARK-59250: DPP prunes all DSv2 partitions when the runtime IN filter " +
+    "on a cast key is empty",
+    DisableAdaptiveExecution("an empty build side collapses the join under AQE")) {
+    withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+      SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+      withTable("dim_big") {
+        Seq[(Long, String)]((1L, "NL"), (2L, "NL"), (3L, "DE"), (4L, "US"), (5L, "US"))
+          .toDF("store_id", "country")
+          .write
+          .format(tableFormat)
+          .saveAsTable("dim_big")
+
+        // the BIGINT dim key adds cast(f.store_id as bigint) on the pruning key, and no dim
+        // rows match, so the runtime filter degenerates into cast(store_id) IN ()
+        val df = sql(
+          """
+            |SELECT f.date_id, f.store_id FROM fact_sk f
+            |JOIN dim_big s ON f.store_id = s.store_id AND s.country = 'XX'
+          """.stripMargin)
+
+        checkPartitionPruningPredicate(df, withSubquery = true, withBroadcast = false)
+        checkAnswer(df, Nil)
+
+        val scan = collectFactScan(df)
+        assert(scan.filteredPartitions.flatten.isEmpty,
+          s"expected all partitions pruned by the empty runtime filter, " +
+            s"got ${scan.filteredPartitions.flatten.size} of ${scan.inputPartitions.size}")
+      }
+    }
+  }
 }
 
 class DynamicPartitionPruningV2SuiteAEOff extends DynamicPartitionPruningV2Suite