[
https://issues.apache.org/jira/browse/SPARK-49030?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17870359#comment-17870359
]
Gera Shegalov edited comment on SPARK-49030 at 8/2/24 5:03 AM:
---------------------------------------------------------------
I think UNION ALL of the same CTE which is LIMIT 1 out of spark.range(100) is
a better illustration of the nondeterminism through the duplicate CTE
evaluation.
Using Spark 3.5.1:
{code:java}
$ ~/dist/spark-3.5.1-bin-hadoop3/bin/pyspark --master local-cluster[4,1,1024]
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/__ / .__/\_,_/_/ /_/\_\ version 3.5.1
/_/
Using Python version 3.10.12 (main, Mar 22 2024 16:50:05)
Spark context Web UI available at http://view-localhost:4040
Spark context available as 'sc' (master = local-cluster[4,1,1024], app id =
app-20240801213303-0000).
SparkSession available as 'spark'.
>>> spark.range(100).repartition(100).write.mode("overwrite").parquet("/tmp/range")
>>>
>>>
>>> spark.read.parquet("/tmp/range").createOrReplaceTempView('tbl')
>>>
>>>
>>> df = sql("WITH c AS (SELECT * FROM tbl LIMIT 1) SELECT * FROM c UNION ALL
>>> SELECT * FROM c")
>>>
>>> df.collect()
>>>
>>>
[Row(id=86), Row(id=14)]
>>> df.explain("extended")
>>>
== Parsed Logical Plan ==
CTE [c]
: +- 'SubqueryAlias c
: +- 'GlobalLimit 1
: +- 'LocalLimit 1
: +- 'Project [*]
: +- 'UnresolvedRelation [tbl], [], false
+- 'Union false, false
:- 'Project [*]
: +- 'UnresolvedRelation [c], [], false
+- 'Project [*]
+- 'UnresolvedRelation [c], [], false
== Analyzed Logical Plan ==
id: bigint
WithCTE
:- CTERelationDef 0, false
: +- SubqueryAlias c
: +- GlobalLimit 1
: +- LocalLimit 1
: +- Project [id#5L]
: +- SubqueryAlias tbl
: +- View (`tbl`, [id#5L])
: +- Relation [id#5L] parquet
+- Union false, false
:- Project [id#5L]
: +- SubqueryAlias c
: +- CTERelationRef 0, true, [id#5L], false
+- Project [id#7L]
+- SubqueryAlias c
+- CTERelationRef 0, true, [id#7L], false
== Optimized Logical Plan ==
Union false, false
:- GlobalLimit 1
: +- LocalLimit 1
: +- Relation [id#5L] parquet
+- GlobalLimit 1
+- LocalLimit 1
+- Relation [id#10L] parquet
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
Union
:- *(3) GlobalLimit 1, 0
: +- ShuffleQueryStage 0
: +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=79]
: +- *(1) LocalLimit 1
: +- *(1) ColumnarToRow
: +- FileScan parquet [id#5L] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/range],
PartitionFilters: [], PushedFilters: [],
ReadSchema: struct<id:bigint>
+- *(4) GlobalLimit 1, 0
+- ShuffleQueryStage 2
+- ReusedExchange [id#10L], Exchange SinglePartition,
ENSURE_REQUIREMENTS, [plan_id=79]
+- == Initial Plan ==
Union
:- GlobalLimit 1, 0
: +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=64]
: +- LocalLimit 1
: +- FileScan parquet [id#5L] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/range],
PartitionFilters: [], PushedFilters: [], ReadSc
hema: struct<id:bigint>
+- GlobalLimit 1, 0
+- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=66]
+- LocalLimit 1
+- FileScan parquet [id#10L] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/range],
PartitionFilters: [], PushedFilters: [], ReadS
chema: struct<id:bigint>
{code}
Because GlobalLimit is not part of the reused exchange
{{ReusedExchange [id#10L|#10L], Exchange SinglePartition, ENSURE_REQUIREMENTS,
[plan_id=79]}}
There are two non-identical rows in the result: {{ [Row(id=86), Row(id=14)] }}
was (Author: jira.shegalov):
I think UNION ALL of the same CTE which is LIMIT 1 out of spark.range(100) is
a better illustration of the nondeterminism duplicate CTE evaluation.
Using Spark 3.5.1:
{code:java}
$ ~/dist/spark-3.5.1-bin-hadoop3/bin/pyspark --master local-cluster[4,1,1024]
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/__ / .__/\_,_/_/ /_/\_\ version 3.5.1
/_/
Using Python version 3.10.12 (main, Mar 22 2024 16:50:05)
Spark context Web UI available at http://view-localhost:4040
Spark context available as 'sc' (master = local-cluster[4,1,1024], app id =
app-20240801213303-0000).
SparkSession available as 'spark'.
>>> spark.range(100).repartition(100).write.mode("overwrite").parquet("/tmp/range")
>>>
>>>
>>> spark.read.parquet("/tmp/range").createOrReplaceTempView('tbl')
>>>
>>>
>>> df = sql("WITH c AS (SELECT * FROM tbl LIMIT 1) SELECT * FROM c UNION ALL
>>> SELECT * FROM c")
>>>
>>> df.collect()
>>>
>>>
[Row(id=86), Row(id=14)]
>>> df.explain("extended")
>>>
== Parsed Logical Plan ==
CTE [c]
: +- 'SubqueryAlias c
: +- 'GlobalLimit 1
: +- 'LocalLimit 1
: +- 'Project [*]
: +- 'UnresolvedRelation [tbl], [], false
+- 'Union false, false
:- 'Project [*]
: +- 'UnresolvedRelation [c], [], false
+- 'Project [*]
+- 'UnresolvedRelation [c], [], false
== Analyzed Logical Plan ==
id: bigint
WithCTE
:- CTERelationDef 0, false
: +- SubqueryAlias c
: +- GlobalLimit 1
: +- LocalLimit 1
: +- Project [id#5L]
: +- SubqueryAlias tbl
: +- View (`tbl`, [id#5L])
: +- Relation [id#5L] parquet
+- Union false, false
:- Project [id#5L]
: +- SubqueryAlias c
: +- CTERelationRef 0, true, [id#5L], false
+- Project [id#7L]
+- SubqueryAlias c
+- CTERelationRef 0, true, [id#7L], false
== Optimized Logical Plan ==
Union false, false
:- GlobalLimit 1
: +- LocalLimit 1
: +- Relation [id#5L] parquet
+- GlobalLimit 1
+- LocalLimit 1
+- Relation [id#10L] parquet
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
Union
:- *(3) GlobalLimit 1, 0
: +- ShuffleQueryStage 0
: +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=79]
: +- *(1) LocalLimit 1
: +- *(1) ColumnarToRow
: +- FileScan parquet [id#5L] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/range],
PartitionFilters: [], PushedFilters: [],
ReadSchema: struct<id:bigint>
+- *(4) GlobalLimit 1, 0
+- ShuffleQueryStage 2
+- ReusedExchange [id#10L], Exchange SinglePartition,
ENSURE_REQUIREMENTS, [plan_id=79]
+- == Initial Plan ==
Union
:- GlobalLimit 1, 0
: +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=64]
: +- LocalLimit 1
: +- FileScan parquet [id#5L] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/range],
PartitionFilters: [], PushedFilters: [], ReadSc
hema: struct<id:bigint>
+- GlobalLimit 1, 0
+- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=66]
+- LocalLimit 1
+- FileScan parquet [id#10L] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/range],
PartitionFilters: [], PushedFilters: [], ReadS
chema: struct<id:bigint>
{code}
Because GlobalLimit is not part of the reused exchange
{{ReusedExchange [id#10L|#10L], Exchange SinglePartition, ENSURE_REQUIREMENTS,
[plan_id=79]}}
There are two non-identical rows in the result: {{ [Row(id=86), Row(id=14)] }}
> Self join of a CTE seems non-deterministic
> ------------------------------------------
>
> Key: SPARK-49030
> URL: https://issues.apache.org/jira/browse/SPARK-49030
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.0.0
> Environment: Tested with Spark 3.4.1, 3.5.1, and 4.0.0-preview.
> Reporter: Jihoon Son
> Priority: Minor
> Fix For: 3.5.3
>
> Attachments: screenshot-1.png
>
>
> {code:java}
> WITH c AS (SELECT * FROM customer LIMIT 10)
> SELECT count(*)
> FROM c c1, c c2
> WHERE c1.c_customer_sk > c2.c_customer_sk{code}
> Suppose a self join query on a CTE such as the one above.
> Spark generates a physical plan like the one below for this query.
> {code:java}
> == Physical Plan ==
> AdaptiveSparkPlan isFinalPlan=false
> +- HashAggregate(keys=[], functions=[count(1)], output=[count(1)#194L])
> +- HashAggregate(keys=[], functions=[partial_count(1)],
> output=[count#233L])
> +- Project
> +- BroadcastNestedLoopJoin BuildRight, Inner, (c_customer_sk#0 >
> c_customer_sk#214)
> :- Filter isnotnull(c_customer_sk#0)
> : +- GlobalLimit 10, 0
> : +- Exchange SinglePartition, ENSURE_REQUIREMENTS,
> [plan_id=256]
> : +- LocalLimit 10
> : +- FileScan parquet [c_customer_sk#0] Batched: true,
> DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1
> paths)[file:/some/path/customer], PartitionFilters: [], PushedFilters: [],
> ReadSchema: struct<c_customer_sk:int>
> +- BroadcastExchange IdentityBroadcastMode, [plan_id=263]
> +- Filter isnotnull(c_customer_sk#214)
> +- GlobalLimit 10, 0
> +- Exchange SinglePartition, ENSURE_REQUIREMENTS,
> [plan_id=259]
> +- LocalLimit 10
> +- FileScan parquet [c_customer_sk#214] Batched:
> true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1
> paths)[file:/some/path/customer], PartitionFilters: [], PushedFilters: [],
> ReadSchema: struct<c_customer_sk:int>{code}
> Evaluating this plan produces non-deterministic result because the limit is
> independently pushed into the two sides of the join. Each limit can produce
> different data, and thus the join can produce results that vary across runs.
> I understand that the query in question is not deterministic (and thus not
> very practical) as, due to the nature of the limit in distributed engines, it
> is not expected to produce the same result anyway across repeated runs.
> However, I would still expect that the query plan evaluation remains
> deterministic.
> Per extended analysis as seen below, it seems that the query plan has changed
> at some point during optimization.
> {code:java}
> == Analyzed Logical Plan ==
> count(1): bigint
> WithCTE
> :- CTERelationDef 2, false
> : +- SubqueryAlias c
> : +- GlobalLimit 10
> : +- LocalLimit 10
> : +- Project [c_customer_sk#0, c_customer_id#1,
> c_current_cdemo_sk#2, c_current_hdemo_sk#3, c_current_addr_sk#4,
> c_first_shipto_date_sk#5, c_first_sales_date_sk#6, c_salutation#7,
> c_first_name#8, c_last_name#9, c_preferred_cust_flag#10, c_birth_day#11L,
> c_birth_month#12L, c_birth_year#13L, c_birth_country#14, c_login#15,
> c_email_address#16, c_last_review_date_sk#17]
> : +- SubqueryAlias customer
> : +- View (`customer`, [c_customer_sk#0, c_customer_id#1,
> c_current_cdemo_sk#2, c_current_hdemo_sk#3, c_current_addr_sk#4,
> c_first_shipto_date_sk#5, c_first_sales_date_sk#6, c_salutation#7,
> c_first_name#8, c_last_name#9, c_preferred_cust_flag#10, c_birth_day#11L,
> c_birth_month#12L, c_birth_year#13L, c_birth_country#14, c_login#15,
> c_email_address#16, c_last_review_date_sk#17])
> : +- Relation
> [c_customer_sk#0,c_customer_id#1,c_current_cdemo_sk#2,c_current_hdemo_sk#3,c_current_addr_sk#4,c_first_shipto_date_sk#5,c_first_sales_date_sk#6,c_salutation#7,c_first_name#8,c_last_name#9,c_preferred_cust_flag#10,c_birth_day#11L,c_birth_month#12L,c_birth_year#13L,c_birth_country#14,c_login#15,c_email_address#16,c_last_review_date_sk#17]
> parquet
> +- Aggregate [count(1) AS count(1)#194L]
> +- Filter (c_customer_sk#0 > c_customer_sk#176)
> +- Join Inner
> :- SubqueryAlias c1
> : +- SubqueryAlias c
> : +- CTERelationRef 2, true, [c_customer_sk#0, c_customer_id#1,
> c_current_cdemo_sk#2, c_current_hdemo_sk#3, c_current_addr_sk#4,
> c_first_shipto_date_sk#5, c_first_sales_date_sk#6, c_salutation#7,
> c_first_name#8, c_last_name#9, c_preferred_cust_flag#10, c_birth_day#11L,
> c_birth_month#12L, c_birth_year#13L, c_birth_country#14, c_login#15,
> c_email_address#16, c_last_review_date_sk#17], false
> +- SubqueryAlias c2
> +- SubqueryAlias c
> +- CTERelationRef 2, true, [c_customer_sk#176,
> c_customer_id#177, c_current_cdemo_sk#178, c_current_hdemo_sk#179,
> c_current_addr_sk#180, c_first_shipto_date_sk#181, c_first_sales_date_sk#182,
> c_salutation#183, c_first_name#184, c_last_name#185,
> c_preferred_cust_flag#186, c_birth_day#187L, c_birth_month#188L,
> c_birth_year#189L, c_birth_country#190, c_login#191, c_email_address#192,
> c_last_review_date_sk#193], false
> == Optimized Logical Plan ==
> Aggregate [count(1) AS count(1)#194L]
> +- Project
> +- Join Inner, (c_customer_sk#0 > c_customer_sk#214)
> :- Filter isnotnull(c_customer_sk#0)
> : +- GlobalLimit 10
> : +- LocalLimit 10
> : +- Project [c_customer_sk#0]
> : +- Relation
> [c_customer_sk#0,c_customer_id#1,c_current_cdemo_sk#2,c_current_hdemo_sk#3,c_current_addr_sk#4,c_first_shipto_date_sk#5,c_first_sales_date_sk#6,c_salutation#7,c_first_name#8,c_last_name#9,c_preferred_cust_flag#10,c_birth_day#11L,c_birth_month#12L,c_birth_year#13L,c_birth_country#14,c_login#15,c_email_address#16,c_last_review_date_sk#17]
> parquet
> +- Filter isnotnull(c_customer_sk#214)
> +- GlobalLimit 10
> +- LocalLimit 10
> +- Project [c_customer_sk#214]
> +- Relation
> [c_customer_sk#214,c_customer_id#215,c_current_cdemo_sk#216,c_current_hdemo_sk#217,c_current_addr_sk#218,c_first_shipto_date_sk#219,c_first_sales_date_sk#220,c_salutation#221,c_first_name#222,c_last_name#223,c_preferred_cust_flag#224,c_birth_day#225L,c_birth_month#226L,c_birth_year#227L,c_birth_country#228,c_login#229,c_email_address#230,c_last_review_date_sk#231]
> parquet {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
