Peter Toth created SPARK-60260:
----------------------------------
Summary: Fix wrong results when
spark.sql.sources.v2.bucketing.enabled changes after a storage-partitioned join
is planned
Key: SPARK-60260
URL: https://issues.apache.org/jira/browse/SPARK-60260
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.2.0, 4.1.3, 4.0.4
Reporter: Peter Toth
A storage-partitioned join can return wrong results, or fail, when
{{spark.sql.sources.v2.bucketing.enabled}} is turned off between planning and
executing a query. It needs dynamic partition pruning on the keyed scan of a
join with no {{GroupPartitionsExec}} above it, e.g. two sources that already
report one split per partition key.
Measured with
{{spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=false}} and no
broadcast, on {{items}} and {{purchases}} identity-partitioned on {{id}} /
{{item_id}}, 7 keys, one split each:
{code:scala}
val df = sql("SELECT i.id, i.name, p.price FROM items i JOIN purchases p ON
i.id = p.item_id WHERE p.price > 0.0")
df.queryExecution.executedPlan // planned with the config on
spark.conf.set("spark.sql.sources.v2.bucketing.enabled", "false")
df.collect()
{code}
* On master and branch-4.2, with AQE on, it returns 2 rows instead of 7. When
the runtime filter prunes splits ({{p.price > 5.0}}), it fails with {{Can't zip
RDDs with unequal numbers of partitions: List(4, 7)}}. With AQE off it is
correct.
* On branch-4.1 and branch-4.0 the same happens with AQE on, and with AQE off
it returns no rows.
The cause on master: {{PlanAdaptiveDynamicPruningFilters}} makes a new
{{BatchScanExec}} when the stage is created, and the new instance reads the
config again. It then reports no {{KeyedPartitioning}}, so
{{PushDownUtils.replanWithRuntimeFilters}} returns the splits in the
connector's order and without the empty partitions for pruned keys. The other
side keeps the sorted and padded layout the join was planned with, so the
sort-merge join pairs the wrong partitions.
This is the third bug of this kind in storage-partitioned join: a config read
again after planning changes what a planned node does (SPARK-59279,
SPARK-60044). The fixes so far freeze one config in one node. It may be worth
looking at it more generally. A node that is rebuilt or executed after planning
reads the session config as it is then, so the plan that runs is not always the
plan that was planned.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
