Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-2814

DeltaIteration: DualInputPlanNode cannot be cast to SingleInputPlanNode

    Details

    • Type: Bug
    • Status: Closed
    • Priority: Major
    • Resolution: Fixed
    • Affects Version/s: 0.10.0
    • Fix Version/s: 1.3.0
    • Component/s: Optimizer
    • Labels:
      None

      Description

      A delta iteration that closes with a solution set which is a JoinOperator throws the following exception:

      org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.
      	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:444)
      	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:345)
      	at org.apache.flink.client.program.Client.runBlocking(Client.java:289)
      	at org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:669)
      	at org.apache.flink.client.CliFrontend.run(CliFrontend.java:324)
      	at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:969)
      	at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1019)
      Caused by: java.lang.ClassCastException: org.apache.flink.optimizer.plan.DualInputPlanNode cannot be cast to org.apache.flink.optimizer.plan.SingleInputPlanNode
      	at org.apache.flink.optimizer.dag.WorksetIterationNode.instantiate(WorksetIterationNode.java:432)
      	at org.apache.flink.optimizer.dag.TwoInputNode.addLocalCandidates(TwoInputNode.java:557)
      	at org.apache.flink.optimizer.dag.TwoInputNode.getAlternativePlans(TwoInputNode.java:478)
      	at org.apache.flink.optimizer.dag.DataSinkNode.getAlternativePlans(DataSinkNode.java:204)
      	at org.apache.flink.optimizer.dag.TwoInputNode.getAlternativePlans(TwoInputNode.java:309)
      	at org.apache.flink.optimizer.dag.TwoInputNode.getAlternativePlans(TwoInputNode.java:308)
      	at org.apache.flink.optimizer.Optimizer.compile(Optimizer.java:500)
      	at org.apache.flink.optimizer.Optimizer.compile(Optimizer.java:402)
      	at org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:271)
      	at org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:543)
      	at org.apache.flink.client.program.Client.runBlocking(Client.java:350)
      	at org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:64)
      	at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:796)
      	at org.apache.flink.api.java.DataSet.collect(DataSet.java:424)
      	at org.apache.flink.api.java.DataSet.print(DataSet.java:1365)
      	at Driver.main(Driver.java:366)
      	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      	at java.lang.reflect.Method.invoke(Method.java:497)
      	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:429)
      	... 6 more
      

      Temporary fix is to attach an identity mapper.

        Issue Links

          Activity

          Hide
          githubbot ASF GitHub Bot added a comment -

          GitHub user rekhajoshm opened a pull request:

          https://github.com/apache/flink/pull/2029

          FLINK-2814 Fix for DualInputPlanNode cannot be cast to SingleInputPlanNode

          FLINK-2814 Fix for DeltaIteration: DualInputPlanNode cannot be cast to SingleInputPlanNode

          You can merge this pull request into a Git repository by running:

          $ git pull https://github.com/rekhajoshm/flink FLINK-2814

          Alternatively you can review and apply these changes as the patch at:

          https://github.com/apache/flink/pull/2029.patch

          To close this pull request, make a commit to your master/trunk branch
          with (at least) the following in the commit message:

          This closes #2029


          commit 166407b6e35ffb104d11d8b0c41ace34bfbb0ecd
          Author: Joshi <rekhajoshm@gmail.com>
          Date: 2016-05-24T19:14:15Z

          FLINK-2814 DeltaIteration: DualInputPlanNode cannot be cast to SingleInputPlanNode


          Show
          githubbot ASF GitHub Bot added a comment - GitHub user rekhajoshm opened a pull request: https://github.com/apache/flink/pull/2029 FLINK-2814 Fix for DualInputPlanNode cannot be cast to SingleInputPlanNode FLINK-2814 Fix for DeltaIteration: DualInputPlanNode cannot be cast to SingleInputPlanNode You can merge this pull request into a Git repository by running: $ git pull https://github.com/rekhajoshm/flink FLINK-2814 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2029.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2029 commit 166407b6e35ffb104d11d8b0c41ace34bfbb0ecd Author: Joshi <rekhajoshm@gmail.com> Date: 2016-05-24T19:14:15Z FLINK-2814 DeltaIteration: DualInputPlanNode cannot be cast to SingleInputPlanNode
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the pull request:

          https://github.com/apache/flink/pull/2029#issuecomment-221849475

          Thanks for this fix! I think it may be sufficient, but to be sure, this would need a test.
          Maybe @greghogan can contribute his example that caused the error as a test?

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the pull request: https://github.com/apache/flink/pull/2029#issuecomment-221849475 Thanks for this fix! I think it may be sufficient, but to be sure, this would need a test. Maybe @greghogan can contribute his example that caused the error as a test?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/2029

          @StephanEwen it is line 454: `solutionSetCandidate = siSolutionDeltaCandidate.getInput().getSource();` where I am seeing `solutionSetCandidate` changed from a `SingleInputPlanNode` to `DualInputPlanNode`. Why are we not exiting the loops after creating and adding the `WorksetIterationPlanNode`?

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/2029 @StephanEwen it is line 454: `solutionSetCandidate = siSolutionDeltaCandidate.getInput().getSource();` where I am seeing `solutionSetCandidate` changed from a `SingleInputPlanNode` to `DualInputPlanNode`. Why are we not exiting the loops after creating and adding the `WorksetIterationPlanNode`?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/2029

          @StephanEwen should I create an alternate PR?

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/2029 @StephanEwen should I create an alternate PR?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/2029

          @greghogan I think you are right with your suggestion. Would be great if you can create an alternate fix.

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/2029 @greghogan I think you are right with your suggestion. Would be great if you can create an alternate fix.
          Hide
          githubbot ASF GitHub Bot added a comment -

          GitHub user greghogan opened a pull request:

          https://github.com/apache/flink/pull/3563

          FLINK-2814 [optimizer] DualInputPlanNode cannot be cast to SingleInputPlanNode

          WorksetIterationNode#instantiate loops over all solution and work set candidates. Since the solution set reference is modified in place when the predecessor node can be used in its place, swith this variable to the inner loop.

          @StephanEwen this is similar to #2029 but resets the reference in the loop. I believe my prior suggestion to immediately return upon adding a node was incorrect as the `instantiate` methods look to be compiling all valid combinations.

          IntelliJ code coverage on `flink-optimizer` shows 105 hits through `WorksetIterationNode#instantiate` and it does fix this issue with my Katz Centrality algorithm (which should not be using delta iterations, but I was young and naive when I wrote it).

          You can merge this pull request into a Git repository by running:

          $ git pull https://github.com/greghogan/flink 2814_deltaiteration_dualinputplannode_cannot_be_cast_to_singleinputplannode

          Alternatively you can review and apply these changes as the patch at:

          https://github.com/apache/flink/pull/3563.patch

          To close this pull request, make a commit to your master/trunk branch
          with (at least) the following in the commit message:

          This closes #3563


          commit 34f017834e17fa69e2b7c72bd95e1a819e4e6aa3
          Author: Greg Hogan <code@greghogan.com>
          Date: 2017-03-17T16:09:34Z

          FLINK-2814 [optimizer] DualInputPlanNode cannot be cast to SingleInputPlanNode

          WorksetIterationNode#instantiate loops over all solution and work set
          candidates. Since the solution set reference is modified in place when
          the predecessor node can be used in its place, swith this variable to
          the inner loop.


          Show
          githubbot ASF GitHub Bot added a comment - GitHub user greghogan opened a pull request: https://github.com/apache/flink/pull/3563 FLINK-2814 [optimizer] DualInputPlanNode cannot be cast to SingleInputPlanNode WorksetIterationNode#instantiate loops over all solution and work set candidates. Since the solution set reference is modified in place when the predecessor node can be used in its place, swith this variable to the inner loop. @StephanEwen this is similar to #2029 but resets the reference in the loop. I believe my prior suggestion to immediately return upon adding a node was incorrect as the `instantiate` methods look to be compiling all valid combinations. IntelliJ code coverage on `flink-optimizer` shows 105 hits through `WorksetIterationNode#instantiate` and it does fix this issue with my Katz Centrality algorithm (which should not be using delta iterations, but I was young and naive when I wrote it). You can merge this pull request into a Git repository by running: $ git pull https://github.com/greghogan/flink 2814_deltaiteration_dualinputplannode_cannot_be_cast_to_singleinputplannode Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3563.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3563 commit 34f017834e17fa69e2b7c72bd95e1a819e4e6aa3 Author: Greg Hogan <code@greghogan.com> Date: 2017-03-17T16:09:34Z FLINK-2814 [optimizer] DualInputPlanNode cannot be cast to SingleInputPlanNode WorksetIterationNode#instantiate loops over all solution and work set candidates. Since the solution set reference is modified in place when the predecessor node can be used in its place, swith this variable to the inner loop.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/2029

          I created #3563 which combines this PR and my suggestion.

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/2029 I created #3563 which combines this PR and my suggestion.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3563

          Nice, looks like a compact fix.
          With the explanation about test coverage, +1 from my side

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3563 Nice, looks like a compact fix. With the explanation about test coverage, +1 from my side
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3563

          @greghogan I think you can go ahead and merge this...

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3563 @greghogan I think you can go ahead and merge this...
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/3563

          @StephanEwen thanks for the reminder. Do you think this should also be merged to 1.2 or 1.1?

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/3563 @StephanEwen thanks for the reminder. Do you think this should also be merged to 1.2 or 1.1?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user asfgit closed the pull request at:

          https://github.com/apache/flink/pull/3563

          Show
          githubbot ASF GitHub Bot added a comment - Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/3563
          Hide
          greghogan Greg Hogan added a comment -

          Fixed in 43158a86d9c3f3d6e3d612c8871625e9d22e7ba3

          Show
          greghogan Greg Hogan added a comment - Fixed in 43158a86d9c3f3d6e3d612c8871625e9d22e7ba3
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/2029

          @rehevkor5 thanks for this identifying bug and submitting this PR. My apologies for taking so long to look into this issue. With #3563 accepted I think you can go ahead and close this PR.

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/2029 @rehevkor5 thanks for this identifying bug and submitting this PR. My apologies for taking so long to look into this issue. With #3563 accepted I think you can go ahead and close this PR.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3563

          I think cherry-picking this into the `release-1.2` branch would be nice

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3563 I think cherry-picking this into the `release-1.2` branch would be nice
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user asfgit closed the pull request at:

          https://github.com/apache/flink/pull/2029

          Show
          githubbot ASF GitHub Bot added a comment - Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/2029

            People

            • Assignee:
              rekhajoshm Rekha Joshi
              Reporter:
              greghogan Greg Hogan
            • Votes:
              0 Vote for this issue
              Watchers:
              3 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved:

                Development