Chen Fan created SPARK-26569:
--------------------------------
Summary: Fixed point for batch Operator Optimizations never
reached when optimize logicalPlan
Key: SPARK-26569
URL: https://issues.apache.org/jira/browse/SPARK-26569
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 2.1.0
Environment:
Reporter: Chen Fan
There is a bit complicated Spark App using DataSet api run once a day, and I
noticed the app will hang once in a while,
I add some log and compare two driver log which one belong to successful app,
another belong to faied, and here is some results of investigation
1. Usually the app will running correctly, but sometime it will hang after
finishing job 1
2. According to log I append , the successful app always reach the fixed point
when iteration is 7 on Batch Operator Optimizations, but failed app never
reached this fixed point.
{code:java}
2019-01-04,11:35:34,199 DEBUG org.apache.spark.sql.execution.SparkOptimizer:
=== Result of Batch Operator Optimizations ===
2019-01-04,14:00:42,847 INFO org.apache.spark.sql.execution.SparkOptimizer:
iteration is 6/100, for batch Operator Optimizations
2019-01-04,14:00:42,851 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule
org.apache.spark.sql.catalyst.optimizer.PushPredicateThroughJoin ===
2019-01-04,14:00:42,852 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicate ===
2019-01-04,14:00:42,903 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule
org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints ===
2019-01-04,14:00:42,939 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.BooleanSimplification
===
2019-01-04,14:00:42,951 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PruneFilters ===
2019-01-04,14:00:42,970 INFO org.apache.spark.sql.execution.SparkOptimizer:
iteration is 7/100, for batch Operator Optimizations
2019-01-04,14:00:42,971 INFO org.apache.spark.sql.execution.SparkOptimizer:
Fixed point reached for batch Operator Optimizations after 7 iterations.
2019-01-04,14:13:15,616 INFO org.apache.spark.sql.execution.SparkOptimizer:
iteration is 45/100, for batch Operator Optimizations
2019-01-04,14:13:15,619 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule
org.apache.spark.sql.catalyst.optimizer.PushPredicateThroughJoin ===
2019-01-04,14:13:15,620 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicate ===
2019-01-04,14:13:59,529 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule
org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints ===
2019-01-04,14:13:59,806 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.BooleanSimplification
===
2019-01-04,14:13:59,845 INFO org.apache.spark.sql.execution.SparkOptimizer:
iteration is 46/100, for batch Operator Optimizations
2019-01-04,14:13:59,849 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule
org.apache.spark.sql.catalyst.optimizer.PushPredicateThroughJoin ===
2019-01-04,14:13:59,849 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushDownPredicate ===
2019-01-04,14:14:45,340 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule
org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints ===
2019-01-04,14:14:45,631 INFO org.apache.spark.sql.execution.SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.BooleanSimplification
===
2019-01-04,14:14:45,678 INFO org.apache.spark.sql.execution.SparkOptimizer:
iteration is 47/100, for batch Operator Optimizations
{code}
3. The difference between two logical plan appear in BooleanSimplification on
iteration, before this rule, two logical plan is same:
{code:java}
// just a head part of plan
Project [model#2486, version#12, device#11, date#30, imei#13, pType#14,
releaseType#15, regionType#16, expType#17, tag#18, imeiCount#1586,
startdate#2360, enddate#2361, status#2362]
+- Join Inner, (((cast(cast(imeiCount#1586 as decimal(20,0)) as int) >
10000) || (model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))
&& model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
&& ((((((cast(cast(imeiCount#1586 as decimal(20,0)) as int) >
10000) || (model#2358 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))
&& model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
&& (date#30 >= startdate#2360)) && (date#30 <=
enddate#2361)) && (model#2486 = model#2358)) && (version#12 =
version#2359)))
:- Project [device#11, version#12, date#30, imei#13, pType#14,
releaseType#15, regionType#16, expType#17, tag#18, imeiCount#1586,
UDF(model#10, device#11) AS model#2486]
: +- Filter ((((((cast(cast(imeiCount#1586 as decimal(20,0)) as int) >
10000) || UDF(model#10, device#1584) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael)))
&& ((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) ||
UDF(model#1583, device#11) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
&& isnotnull(UDF(model#1583, device#11))) &&
isnotnull(UDF(model#10, device#1584))) && (((cast(cast(imeiCount#1586
as decimal(20,0)) as int) > 10000) || UDF(model#10, device#11) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael)))
&& isnotnull(UDF(model#10, device#11))))
: +- Join Inner, ((((model#10 = model#1583) && (device#11 =
device#1584)) && (version#12 = version#1585)) && (date#30 =
date#1592))
{code}
4. after BooleanSimplification, There is only one difference: Filter's
constraints format like (A || B) && (A || C) on successful app but A && (B ||
C) on failed app.
{code:java}
//successful app's logical plan
Filter ((((isnotnull(UDF(model#1583, device#11)) &&
((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) ||
UDF(model#1583, device#11) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
&& isnotnull(UDF(model#10, device#1584))) &&
((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) ||
UDF(model#10, device#1584) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
&& (((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000)
|| UDF(model#10, device#11) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael)))
&& isnotnull(UDF(model#10, device#11))))
// failed app's plan
Filter (((((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) ||
(UDF(model#10, device#1584) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))
&& UDF(model#1583, device#11) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
&& isnotnull(UDF(model#1583, device#11))) &&
isnotnull(UDF(model#10, device#1584))) && (((cast(cast(imeiCount#1586
as decimal(20,0)) as int) > 10000) || UDF(model#10, device#11) INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael)))
&& isnotnull(UDF(model#10, device#11))))
{code}
5. hereafter, error occurred with rule InferFiltersFromConstraints:
{code:java}
// code about upon log
case join @ Join(left, right, joinType, conditionOpt) =>
// Only consider constraints that can be pushed down completely to either
the left or the
// right child
val constraints = join.constraints.filter { c =>
c.references.subsetOf(left.outputSet) ||
c.references.subsetOf(right.outputSet)
}
// Remove those constraints that are already enforced by either the left
or the right child
val additionalConstraints = constraints -- (left.constraints ++
right.constraints)
logInfo(
s"""
| After remove constraints additional constraints is
${additionalConstraints.toList.toString}
| left constraints is ${left.constraints.toList.toString()}
| right constraints is ${right.constraints.toList.toString()}
""".stripMargin)
{code}
{code:java}
// successful app's log
2019-01-08,16:44:48,911 INFO
org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints:
After remove constraints additional constraints is
List(((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) ||
(model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))
&& model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael)))))
left constraints is List(isnotnull(model#2486), isnotnull(device#11),
isnotnull(date#30), ((cast(cast(imeiCount#1586 as decimal(20,0)) as int) >
10000) || model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))),
((((((((((((((((((((((((((((((20190107 <=> date#30) || (20190106
<=> date#30)) || (20190105 <=> date#30)) || (20190104 <=>
date#30)) || (20190103 <=> date#30)) || (20190102 <=> date#30)) ||
(20190101 <=> date#30)) || (20181231 <=> date#30)) || (20181230
<=> date#30)) || (20181229 <=> date#30)) || (20181228 <=>
date#30)) || (20181227 <=> date#30)) || (20181226 <=> date#30)) ||
(20181225 <=> date#30)) || (20181224 <=> date#30)) || (20181223
<=> date#30)) || (20181222 <=> date#30)) || (20181221 <=>
date#30)) || (20181220 <=> date#30)) || (20181219 <=> date#30)) ||
(20181218 <=> date#30)) || (20181217 <=> date#30)) || (20181216
<=> date#30)) || (20181215 <=> date#30)) || (20181214 <=>
date#30)) || (20181213 <=> date#30)) || (20181212 <=> date#30)) ||
(20181211 <=> date#30)) || (20181210 <=> date#30)) || (20181209
<=> date#30)), isnotnull(version#12))
right constraints is List(isnotnull(enddate#2361), isnotnull(startdate#2360),
isnotnull(model#2358), isnotnull(version#2359))
//failed app's log
2019-01-08,16:55:11,614 INFO
org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints:
After remove constraints additional constraints is List()
left constraints is List(isnotnull(date#30),
((((((((((((((((((((((((((((((20190107 <=> date#30) || (20190106
<=> date#30)) || (20190105 <=> date#30)) || (20190104 <=>
date#30)) || (20190103 <=> date#30)) || (20190102 <=> date#30)) ||
(20190101 <=> date#30)) || (20181231 <=> date#30)) || (20181230
<=> date#30)) || (20181229 <=> date#30)) || (20181228 <=>
date#30)) || (20181227 <=> date#30)) || (20181226 <=> date#30)) ||
(20181225 <=> date#30)) || (20181224 <=> date#30)) || (20181223
<=> date#30)) || (20181222 <=> date#30)) || (20181221 <=>
date#30)) || (20181220 <=> date#30)) || (20181219 <=> date#30)) ||
(20181218 <=> date#30)) || (20181217 <=> date#30)) || (20181216
<=> date#30)) || (20181215 <=> date#30)) || (20181214 <=>
date#30)) || (20181213 <=> date#30)) || (20181212 <=> date#30)) ||
(20181211 <=> date#30)) || (20181210 <=> date#30)) || (20181209
<=> date#30)), isnotnull(version#12), isnotnull(device#11),
isnotnull(model#2486), ((cast(cast(imeiCount#1586 as decimal(20,0)) as int)
> 10000) || model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))),
((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) ||
(model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))
&& model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael)))))
right constraints is List(isnotnull(enddate#2361), isnotnull(startdate#2360),
isnotnull(version#2359), isnotnull(model#2358))
{code}
Failed app plan's left child has an additional constraint:
{code:java}
((cast(cast(imeiCount#1586 as decimal(20,0)) as int) > 10000) || (model#2486
INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))
&& model#2486 INSET
(grus(grus),lyra(lyra),lavender(lavender),violet(violet),andromeda(andromeda),davinci(davinci),Cepheus(cepheus),onc(onc),aquila(aquila),tiare(tiare),raphael(raphael))))
{code}
5. Soon after the gap between two app's plan is getting bigger, one will
successful, another hang. It seems there are two possibly reason :
1. BooleanSimplification is not idempotent
2. InferFiltersFromConstraints's behavior not correct when child's constraints
has A || ( B && C) instead of (A || B) && (A || C)
I'm not sure which is root casue, could someone follow this question?
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
