[jira] [Commented] (PIG-5207) BugFix e2e tests fail on spark

["Adam Szita (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/PIG-5207?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15950940#comment-15950940 ] 

Adam Szita commented on PIG-5207:
---------------------------------

This test case uses COR UDF to calculate correlation between 3 vars.
The cause of the issue is that when execution reaches the Final implementation of the COR UDF there are some junk bags in the input Tuple:
{code}
--input
  |--DefaultBag: {((943629.1899999954,19810.98000000007,476680.0,52620.35740000006,2.5723842E7)}
  |--DefaultBag: {((157499.53767599948,19810.98000000007,52620.35740000006,52620.35740000006,503441.4849212208)}
  |--InternalCachedBag: 
     |--0 = {BinSedesTuple@7900} "(157499.53767599948,19810.98000000007,52620.35740000006,52620.35740000006,503441.4849212208)"
     |--1 = {Long@7901} "10000"
     |--2 = {BinSedesTuple@7902} "(943629.1899999954,19810.98000000007,476680.0,52620.35740000006,2.5723842E7)"
     |--3 = {Long@7903} "10000"
     |--4 = {BinSedesTuple@7904} "(2509050.8495000014,52620.35740000006,476680.0,503441.4849212208,2.5723842E7)"
     |--5 = {Long@7905} "10000"
{code}

The real result to be consumed is at position 2, but of course since we expect 1 entry here the implementation queries for {{.get(0)}} and shortly after we will get a ClassCastException.
*This is because of a fault in the CombinerOptimizer of Spark, it doesn't remove all original input projections in POUserFunc.Final part* as seen [here|https://github.com/apache/pig/blob/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/optimizer/CombinerOptimizer.java#L299]: {{PhysicalOperator udfInput = pplan.getPredecessors(combineUdf).get(0);}} //only the first input (this is why we didn't see this issue when using for example the AVG UDF)
As seen in the plan:
{code}
Spark node scope-38
D: Store(file:/tmp/temp-1343607396/tmp944122059:org.apache.pig.impl.io.InterStorage) - scope-37
|
|---D: New For Each(false,true)[tuple] - scope-47 (postReduceFE)
    |   |
    |   Project[chararray][0] - scope-39
    |   |
    |   Project[bag][1] - scope-43
    |   |
    |   |---Project[bag][1] - scope-42
    |   
    |   POUserFunc(org.apache.pig.builtin.COR$Final)[bag] - scope-46
    |   |
    |   |---Project[bag][0] - scope-41   << residual unneeded projection, remains from the original UDF's input
    |   |   |
    |   |   |---Project[bag][1] - scope-40
    |   |
    |   |---Project[bag][2] - scope-45   << residual unneeded projection, remains from the original UDF's input
    |   |   |
    |   |   |---Project[bag][1] - scope-44
    |   |
    |   |---Project[bag][1] - scope-67   << actual subresult comes in here
    |
    |---C: Reduce By(false,false)[tuple] - scope-57  (cfe)
        |   |
        |   Project[chararray][0] - scope-58
{code}

After fixing this I saw that although now I get good results, the order of the results is off. The actual correlation values between var0-var1, etc.. are shifted with respect to MR's output.
This is due to another plan generation bug and it is actually observable in the plan above, that {{POUserFunc}} has Project {{0,2,1}} as its input (instead of {{0,1,2}}).

The root of this problem is the cloning of the ForEach operator [here|https://github.com/apache/pig/blob/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/optimizer/CombinerOptimizer.java#L141]. Cloning this will trigger cloning the associated PhysicalPlan instance and unfortunately [that method|https://github.com/apache/pig/blob/spark/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhysicalPlan.java#L217] has a bug:

It doesn't keep the order of the {{List<PhysicalOperator>}} lists in the {{mToEdges}} because it is only considering {{mFromEdges}} to connect the cloned Operator instances.
Keeping the ordering cannot be achieved by looking either only at {{mToEdges}} or only at {{mFromEdges}} since both operate with lists.

As a fix to this I'm sorting these lists in the cloned plan according to what was in the original plan (by the use of the {{matches}} map that maps original and cloned ops.

This doesn't come up in MR mode because the ForEach op is not cloned there but rather modified in-place during combiner optimization. We could do the same in Spark too but I feel this should rather be fixed in the common code for future clone() convenience. 

[~kellyzly] can you review the spark parts in CombinerOptimizer.java please?
[~rohini] can you please review the common parts in PhysicalPlan.java?

Please find the fix in my patch: [^PIG-5207.0.patch]

> BugFix e2e tests fail on spark
> ------------------------------
>
>                 Key: PIG-5207
>                 URL: https://issues.apache.org/jira/browse/PIG-5207
>             Project: Pig
>          Issue Type: Sub-task
>          Components: spark
>            Reporter: Adam Szita
>            Assignee: Adam Szita
>             Fix For: spark-branch
>
>         Attachments: PIG-5207.0.patch
>
>
> Observed ClassCastException in BugFix 1 and 2 test cases. The exception is thrown from and UDF: COR.Final



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)