From 71aedc6c9eeec230b8f657211524853a3ecfaa06 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Fri, 4 Sep 2026 21:15:22 +0800 Subject: [PATCH 1/3] [SPARK-59250][SQL] Translate an empty runtime IN filter to AlwaysFalse in DSv2 runtime filtering Assisted-by: Claude Fable 5 --- .../catalog/InMemoryTableWithV2Filter.scala | 6 +- .../datasources/v2/DataSourceV2Strategy.scala | 4 +- .../sql/DynamicPartitionPruningSuite.scala | 64 +++++++++++++++++++ 3 files changed, 72 insertions(+), 2 deletions(-) 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 1228b70b103d0..6922c874cf73a 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 8805bfe75298f..7ee68bc39448f 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 9f497680b7fdb..e62e49b485b62 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 @@ -2680,6 +2680,70 @@ abstract class DynamicPartitionPruningV2FilterSuite super.initState() spark.conf.set("spark.sql.catalog.testcat", classOf[InMemoryTableWithV2FilterCatalog].getName) } + + import testImplicits._ + + 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 DynamicPartitionPruningV2FilterSuiteAEOff From 5ccb9f19895a1496d212cc16c5b30d716618079f Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Sat, 5 Sep 2026 00:45:28 +0800 Subject: [PATCH 2/3] [SPARK-59250][SQL] Honor AlwaysFalse in InMemoryBatchScan runtime filtering Assisted-by: Claude Fable 5 --- .../spark/sql/connector/catalog/InMemoryBaseTable.scala | 4 ++++ 1 file changed, 4 insertions(+) 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 d0f7a19583874..39634eb4c8343 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 { From 9ceb03f349609f97f15ef1e16ee117e98eb89af5 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Sat, 5 Sep 2026 14:25:54 +0800 Subject: [PATCH 3/3] [SPARK-59250][SQL] Cover V1 runtime filtering in empty IN tests Assisted-by: Claude Fable 5 --- .../sql/DynamicPartitionPruningSuite.scala | 34 +++++++++---------- 1 file changed, 17 insertions(+), 17 deletions(-) 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 e62e49b485b62..e09a170734bbb 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,29 +2659,14 @@ 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") } -} - -class DynamicPartitionPruningV2SuiteAEOff extends DynamicPartitionPruningV2Suite - with DisableAdaptiveExecutionSuite - -class DynamicPartitionPruningV2SuiteAEOn extends DynamicPartitionPruningV2Suite - with EnableAdaptiveExecutionSuite - -abstract class DynamicPartitionPruningV2FilterSuite - extends DynamicPartitionPruningV2Suite { - - override protected def initState(): Unit = { - super.initState() - spark.conf.set("spark.sql.catalog.testcat", classOf[InMemoryTableWithV2FilterCatalog].getName) - } - - import testImplicits._ private def collectFactScan(df: DataFrame): BatchScanExec = { val scans = collectWithSubqueries(df.queryExecution.executedPlan) { @@ -2746,6 +2731,21 @@ abstract class DynamicPartitionPruningV2FilterSuite } } +class DynamicPartitionPruningV2SuiteAEOff extends DynamicPartitionPruningV2Suite + with DisableAdaptiveExecutionSuite + +class DynamicPartitionPruningV2SuiteAEOn extends DynamicPartitionPruningV2Suite + with EnableAdaptiveExecutionSuite + +abstract class DynamicPartitionPruningV2FilterSuite + extends DynamicPartitionPruningV2Suite { + + override protected def initState(): Unit = { + super.initState() + spark.conf.set("spark.sql.catalog.testcat", classOf[InMemoryTableWithV2FilterCatalog].getName) + } +} + class DynamicPartitionPruningV2FilterSuiteAEOff extends DynamicPartitionPruningV2FilterSuite with DisableAdaptiveExecutionSuite