diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala index d0f7a1958387..39634eb4c834 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala +++ b/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 { diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala index 1228b70b103d..6922c874cf73 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala +++ b/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 { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala index 8805bfe75298..7ee68bc39448 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala +++ b/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( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala index 9f497680b7fd..e09a170734bb 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala +++ b/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