-
Notifications
You must be signed in to change notification settings - Fork 29.4k
[SPARK-59250][SQL] Translate an empty runtime IN filter to AlwaysFalse in DSv2 runtime filtering #58524
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[SPARK-59250][SQL] Translate an empty runtime IN filter to AlwaysFalse in DSv2 runtime filtering #58524
Changes from 2 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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", | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could we also cover the
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Moved the tests into DynamicPartitionPruningV2Suite, so they now run through V1 Filter, V2 Predicate, and Catalyst runtime filtering. |
||
| 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 | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: can we put it on top or more natural place ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Moved the import and helper into the shared DynamicPartitionPruningV2Suite.