This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 87c75f1c6a5a [SPARK-58783][4.1][SQL] Fix SPJ with runtime filtering
when join keys are a subset of partition keys
87c75f1c6a5a is described below
commit 87c75f1c6a5a15b8546f3149183d50450a35a8e6
Author: Peter Toth <[email protected]>
AuthorDate: Sat Aug 15 06:21:03 2026 -0700
[SPARK-58783][4.1][SQL] Fix SPJ with runtime filtering when join keys are a
subset of partition keys
### What changes were proposed in this pull request?
This PR makes `BatchScanExec.filteredPartitions` validate the partitions a
data source reports after runtime filtering against the **unprojected**
partitioning (`super.outputPartitioning`) instead of `outputPartitioning`.
`BatchScanExec.outputPartitioning` projects the partition expressions and
values down to the join keys when SPJ params were pushed down
(`spjParams.joinKeyPositions`), and it also replaces the partition values with
the common partition values of the join. But `filteredPartitions` wraps the
re-planned partitions with their full, unprojected
`HasPartitionKey.partitionKey()`, so the two sides of the comparison are in
different spaces.
Nothing else needs to change, because everything after the check already
works in the full partition key space: `groupPartitions` groups by the
unprojected keys, and the projection, the reducers, the common partition values
and the empty-partition padding all happen later, in `inputRDD`.
One further consequence worth calling out: when `commonPartitionValues` was
pushed down, the values the check compared against were the *merged* set from
both sides of the join, so it also accepted keys that only the other side had
ever reported. It now compares against this scan's own pre-filter keys, which
is what the error message claims. Every data source that only prunes during
filtering still passes, since its post-filter keys are a subset of its own
pre-filter keys.
This PR targets `branch-4.1` (and later `branch-4.0`) only. `master` and
`branch-4.2` are not affected: SPARK-55535 / SPARK-55092 moved the join key
projection out of the scan into `GroupPartitionsExec`, so there both sides of
the comparison are already unprojected. That refactor is too large for a
maintenance branch, hence this targeted fix.
### Why are the changes needed?
A storage-partitioned join and runtime filtering cannot coexist on the same
scan when the join keys are a subset of the partition keys. The query fails on
the driver, in `BatchScanExec.filteredPartitions`, before a single task runs:
```
org.apache.spark.SparkException: During runtime filtering, data source must
not report new partition values that are not present in the original
partitioning.
```
The general trigger is that `outputPartitioning`'s partition values are not
a superset of the keys the scan itself reported, so a post-filter key looks new
when it isn't. The practically reachable way to get there is a projected
partitioning:
1. SPJ with
`spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true`
(off by default), so Spark projects the scan partitioning down to the join
keys.
2. A scan reporting `KeyGroupedPartitioning` where the join keys are **not
a leading prefix** of the partition keys. `InternalRowComparableWrapper` builds
its `StructType` and its ordering from the projected data types, and
`InterpretedOrdering.compare` walks `ordering.size` positions, so the
comparison is positional over the first k fields of the raw key. When the
projection is a leading prefix (`Seq(0)`, `Seq(0, 1)`, …) it happens to compare
the same columns and the mismatch goes un [...]
3. A runtime filter on that same scan.
`spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled` (the
values are reduced into the smaller bucket space while the keys the scan
reports are not) and `spark.sql.sources.v2.bucketing.partition.filter.enabled`
(the values become the intersection of the two sides, or one side's for an
outer join) are two further ways to make the value set not a superset, and this
change repairs those as well. All three have tests, see below.
With `allowJoinKeysSubsetOfPartitionKeys` enabled, a copy-on-write `MERGE
INTO` hits the other two conditions by default, because the row-level operation
group filter
(`spark.sql.optimizer.runtime.rowLevelOperationGroupFilter.enabled`) is on: the
target is partitioned by e.g. `months(created), bucket(4, id)` and the merge
joins on `id`, which is the second partition key. This makes SPJ unusable for
`MERGE`, which is the case it is most worth having. The only workaround is to
turn the [...]
### Does this PR introduce _any_ user-facing change?
Yes, it fixes the bug above: a query that failed with the exception now
runs. There is no change for queries that don't combine SPJ, join keys that are
a non-prefix subset of the partition keys, and a runtime filter on the same
scan. This is a change compared to released 4.0.x and 4.1.x; `master` and
`branch-4.2` already behave this way.
### How was this patch tested?
Six new tests in `KeyGroupedPartitioningSuite`, each asserting that the
dynamic filter really is pushed to the scan the bug needs it on and that SPJ is
triggered:
| shape | failure on `branch-4.1` without the source change |
|---|---|
| join key is the second of two partition keys, projection `Seq(1)` |
`SparkException`, must not report new partition values |
| join keys are the first and third of three, projection `Seq(0, 2)` | same
|
| projection changes the key types, `Seq(1)` over `(int, string)` |
`ClassCastException: class java.lang.Integer cannot be cast to class
org.apache.spark.unsafe.types.UTF8String` in
`InternalRowComparableWrapper.equals` |
| `partition.filter.enabled`, join keys cover all partition keys, **no
projection** | `SparkException`, "Before: 2 partition values. After: 3
partition values" |
| `partition.filter.enabled` on a `LEFT OUTER` join, so the values become
one side's | `SparkException`, must not report new partition values |
| `allowCompatibleTransforms.enabled` with `bucket(4, ...)` against
`bucket(2, ...)`, so the values are reduced | same |
All six pass with the source change. The last three carry no projection at
all, which is why the fix is about the value set rather than about
`joinKeyPositions`.
Note that `InMemoryBatchScan.filter` only prunes for a single-transform
table, so these tests cover the driver-side validation rather than the pruning
that follows it; the existing single-key dynamic-filtering tests in the suite
cover the pruning and padding path.
Also ran the full `KeyGroupedPartitioningSuite` (76 tests, covering the
partially-clustered and dynamic-filtering combinations that exercise the common
partition values path), plus `GroupBasedMergeIntoTableSuite`,
`GroupBasedDeleteFromTableSuite`, `GroupBasedUpdateTableSuite` and
`DataSourceV2Suite` — 322 tests in total.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes #58013 from peter-toth/SPARK-58783-spj-runtime-filtering-4.1.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
---
.../execution/datasources/v2/BatchScanExec.scala | 7 +-
.../connector/KeyGroupedPartitioningSuite.scala | 279 +++++++++++++++++++++
2 files changed, 285 insertions(+), 1 deletion(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
index 55866cc85840..a7f8b3640204 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
@@ -68,7 +68,12 @@ case class BatchScanExec(
}
if (dataSourceFilters.nonEmpty) {
- val originalPartitioning = outputPartitioning
+ // The partitions reported by the data source after filtering are
compared below by their
+ // full partition keys, so the partitioning they are compared against
must not be the
+ // projected one: `outputPartitioning` projects the partition keys down
to the join keys and
+ // replaces the partition values with the common ones of the join when
SPJ params were pushed
+ // down. Projecting and re-grouping the filtered partitions happens
later, in `inputRDD`.
+ val originalPartitioning = super.outputPartitioning
// the cast is safe as runtime filters are only assigned if the scan can
be filtered
val filterableScan = scan.asInstanceOf[SupportsRuntimeV2Filtering]
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index e8d51dedcda3..77324eb3b77c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -2334,6 +2334,285 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-58783: SPJ with runtime filtering when the join key is a
trailing partition key") {
+ val table1 = "table1"
+ val table2 = "table2"
+ // the join key is the second partition key of `table1`, so its scan
partitioning is projected
+ // down to `dept_id` when SPJ is applied
+ createTable(table1, columns2, Array(identity("store_id"),
identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
+ "(100, 1, 'aa'), " +
+ "(100, 2, 'ab'), " +
+ "(200, 1, 'ac'), " +
+ "(200, 3, 'ad')")
+
+ createTable(table2, columns2, Array(identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
+ "(100, 1, 'x'), " +
+ "(100, 2, 'y'), " +
+ "(100, 3, 'x')")
+
+ Seq(true, false).foreach { partiallyClustered =>
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
+ partiallyClustered.toString) {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("t1", "t2")}
+ |t1.store_id, t1.dept_id, t1.data AS t1data, t2.data AS t2data
+ |FROM testcat.ns.$table1 t1 JOIN testcat.ns.$table2 t2
+ |ON t1.dept_id = t2.dept_id WHERE t2.data = 'x'
+ |ORDER BY t1.store_id, t1.dept_id, t1data
+ |""".stripMargin)
+ // the dynamic filter must land on the `table1` scan, the side whose
partitioning is
+ // projected down to a non-prefix of its partition keys
+ val scans = collectScans(df.queryExecution.executedPlan)
+ assert(
+ scans.exists(s =>
+ s.runtimeFilters.nonEmpty &&
s.spjParams.joinKeyPositions.contains(Seq(1))),
+ "the dynamic filter should be pushed to the scan with the projected
partitioning")
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "SPJ should be triggered")
+ checkAnswer(df, Seq(
+ Row(100, 1, "aa", "x"),
+ Row(200, 1, "ac", "x"),
+ Row(200, 3, "ad", "x")))
+ }
+ }
+ }
+
+ test("SPARK-58783: SPJ with runtime filtering when the join keys have a gap
in the " +
+ "partition keys") {
+ val table1 = "table1"
+ val table2 = "table2"
+ // the join keys are the first and the third partition key of `table1`, so
its scan partitioning
+ // is projected down to `data, dept_id` when SPJ is applied. The partition
key the projection
+ // drops, `store_id`, sits between them, which is what makes the values it
is compared against
+ // line up with the wrong column.
+ createTable(table1, columns2,
+ Array(identity("data"), identity("store_id"), identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
+ "(100, 1, 'aa'), " +
+ "(200, 2, 'aa'), " +
+ "(100, 3, 'bb')")
+
+ createTable(table2, columns2, Array(identity("data"), identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
+ "(500, 1, 'aa'), " +
+ "(500, 2, 'aa'), " +
+ "(500, 3, 'bb')")
+
+ Seq(true, false).foreach { partiallyClustered =>
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
+ partiallyClustered.toString) {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("t1", "t2")}
+ |t1.store_id, t1.dept_id, t1.data, t2.store_id AS t2store_id
+ |FROM testcat.ns.$table1 t1 JOIN testcat.ns.$table2 t2
+ |ON t1.data = t2.data AND t1.dept_id = t2.dept_id WHERE t2.data =
'aa'
+ |ORDER BY t1.store_id, t1.dept_id
+ |""".stripMargin)
+ val scans = collectScans(df.queryExecution.executedPlan)
+ assert(
+ scans.exists(s =>
+ s.runtimeFilters.nonEmpty &&
s.spjParams.joinKeyPositions.contains(Seq(0, 2))),
+ "the dynamic filter should be pushed to the scan with the projected
partitioning")
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "SPJ should be triggered")
+ checkAnswer(df, Seq(
+ Row(100, 1, "aa", 500),
+ Row(200, 2, "aa", 500)))
+ }
+ }
+ }
+
+ test("SPARK-58783: SPJ with runtime filtering when the projection changes
the partition key " +
+ "types") {
+ val table1 = "table1"
+ val table2 = "table2"
+ // `data` is the second partition key, so the projection down to it leaves
the wrapper with a
+ // single `StringType`, which it then reads off the first field of the raw
key - an
+ // `IntegerType`
+ createTable(table1, columns2, Array(identity("dept_id"), identity("data")))
+ sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
+ "(100, 1, 'aa'), " +
+ "(200, 2, 'aa'), " +
+ "(300, 3, 'bb')")
+
+ createTable(table2, columns2, Array(identity("data")))
+ sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
+ "(500, 1, 'aa'), " +
+ "(500, 2, 'bb')")
+
+ Seq(true, false).foreach { partiallyClustered =>
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
+ partiallyClustered.toString) {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("t1", "t2")}
+ |t1.store_id, t1.dept_id, t1.data, t2.store_id AS t2store_id
+ |FROM testcat.ns.$table1 t1 JOIN testcat.ns.$table2 t2
+ |ON t1.data = t2.data WHERE t2.dept_id = 1
+ |ORDER BY t1.store_id, t1.dept_id
+ |""".stripMargin)
+ val scans = collectScans(df.queryExecution.executedPlan)
+ assert(
+ scans.exists(s =>
+ s.runtimeFilters.nonEmpty &&
s.spjParams.joinKeyPositions.contains(Seq(1))),
+ "the dynamic filter should be pushed to the scan with the projected
partitioning")
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "SPJ should be triggered")
+ checkAnswer(df, Seq(
+ Row(100, 1, "aa", 500),
+ Row(200, 2, "aa", 500)))
+ }
+ }
+ }
+
+ test("SPARK-58783: SPJ with runtime filtering and partition filtering,
without a projection") {
+ val table1 = "table1"
+ val table2 = "table2"
+ // no projection here - the join keys are all of the partition keys. The
partition values are
+ // replaced by the intersection of the two sides instead, and `table1` has
a partition key that
+ // `table2` does not, so the keys it reports back after filtering are not
all in that
+ // intersection.
+ createTable(table1, columns2, Array(identity("store_id"),
identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
+ "(100, 1, 'aa'), " +
+ "(100, 2, 'ab'), " +
+ "(200, 3, 'ac')")
+
+ createTable(table2, columns2, Array(identity("store_id"),
identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
+ "(100, 1, 'x'), " +
+ "(100, 2, 'y')")
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("t1", "t2")}
+ |t1.store_id, t1.dept_id, t1.data AS t1data, t2.data AS t2data
+ |FROM testcat.ns.$table1 t1 JOIN testcat.ns.$table2 t2
+ |ON t1.store_id = t2.store_id AND t1.dept_id = t2.dept_id WHERE
t2.data = 'x'
+ |ORDER BY t1.store_id, t1.dept_id
+ |""".stripMargin)
+ val scans = collectScans(df.queryExecution.executedPlan)
+ assert(
+ scans.exists(s => s.runtimeFilters.nonEmpty &&
s.table.name().endsWith(table1) &&
+ s.spjParams.joinKeyPositions.isEmpty),
+ s"the dynamic filter should be pushed to the unprojected $table1 scan")
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "SPJ should be triggered")
+ checkAnswer(df, Seq(Row(100, 1, "aa", "x")))
+ }
+ }
+
+ test("SPARK-58783: SPJ with runtime filtering and partition filtering on an
outer join") {
+ val table1 = "table1"
+ val table2 = "table2"
+ // for an outer join the partition values become one side's, so here they
are `table1`'s and the
+ // scan that reports keys outside them is `table2`, the side an outer join
can prune
+ createTable(table1, columns2, Array(identity("store_id"),
identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
+ "(100, 1, 'aa'), " +
+ "(300, 5, 'ab')")
+
+ createTable(table2, columns2, Array(identity("store_id"),
identity("dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
+ "(100, 1, 'x'), " +
+ "(200, 2, 'y')")
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("t1", "t2")}
+ |t1.store_id, t1.dept_id, t1.data AS t1data, t2.data AS t2data
+ |FROM testcat.ns.$table1 t1 LEFT OUTER JOIN testcat.ns.$table2 t2
+ |ON t1.store_id = t2.store_id AND t1.dept_id = t2.dept_id WHERE
t1.data = 'aa'
+ |ORDER BY t1.store_id, t1.dept_id
+ |""".stripMargin)
+ val scans = collectScans(df.queryExecution.executedPlan)
+ assert(
+ scans.exists(s => s.runtimeFilters.nonEmpty &&
s.table.name().endsWith(table2)),
+ s"the dynamic filter should be pushed to the $table2 scan")
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "SPJ should be triggered")
+ checkAnswer(df, Seq(Row(100, 1, "aa", "x")))
+ }
+ }
+
+ test("SPARK-58783: SPJ with runtime filtering and compatible bucket
transforms") {
+ val table1 = "table1"
+ val table2 = "table2"
+ // the two sides bucket into different numbers of buckets, so the
partition values pushed down
+ // are reduced into the smaller bucket space while the keys the scan
reports stay in its own
+ createTable(table1, columns2, Array(bucket(4, "store_id"), bucket(4,
"dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
+ "(0, 0, 'aa'), " +
+ "(2, 3, 'ab')")
+
+ createTable(table2, columns2, Array(bucket(2, "store_id"), bucket(2,
"dept_id")))
+ sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
+ "(0, 0, 'x'), " +
+ "(1, 1, 'y')")
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("t1", "t2")}
+ |t1.store_id, t1.dept_id, t1.data AS t1data, t2.data AS t2data
+ |FROM testcat.ns.$table1 t1 JOIN testcat.ns.$table2 t2
+ |ON t1.store_id = t2.store_id AND t1.dept_id = t2.dept_id WHERE
t2.data = 'x'
+ |ORDER BY t1.store_id, t1.dept_id
+ |""".stripMargin)
+ val scans = collectScans(df.queryExecution.executedPlan)
+ assert(
+ scans.exists(s => s.runtimeFilters.nonEmpty &&
s.spjParams.reducers.nonEmpty),
+ "the dynamic filter should be pushed to the scan whose partition keys
are reduced")
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "SPJ should be triggered")
+ checkAnswer(df, Seq(Row(0, 0, "aa", "x")))
+ }
+ }
+
test("SPARK-48012: one-side shuffle with partition transforms") {
val items_partitions = Array(bucket(2, "id"), identity("arrive_time"))
val items_partitions2 = Array(identity("arrive_time"), bucket(2, "id"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
