[
https://issues.apache.org/jira/browse/SPARK-60260?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131155#comment-18131155
]
Naveen Kumar Puppala commented on SPARK-60260:
----------------------------------------------
I’d like to take this one. It’s the same SPJ pattern as SPARK-59279 and
SPARK-60044, which I’ve been working in. I’ll reproduce the wrong-results case
first, then scope the fix.
On your note that it’s bigger than one node: a node that gets rebuilt or
executed after planning reads the config as it is then, so the plan that runs
isn’t always the one that was planned. Once I’ve got the repro I’ll look at
whether to just freeze the config in the node like the earlier two, or go for
something more general. Will open a PR.
> 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
> Priority: Major
>
> 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
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
