Delta-Iteration: Runtime-Error "Could not set up runtime strategy for input channel to node"

[Stefan Bunk]

Hello,

I built a plan using Delta-Iteration, however, at runtime, when the plan is
executed, I get the following error:

Exception in thread "main"
eu.stratosphere.compiler.CompilerPostPassException: Could not set up
runtime strategy for input channel to node ''. Missing type information for
field 0
I have no idea, where to look now, can anyone help?

In case it is needed, please find below a description of the problem and
the code, as well as the stacktrace:
Usecase:
I have a  list of edges like this (lets call them base links):
a --> b
e --> f

And a set of (lets call them additional links):
b --> c
c --> d
f --> g

I want to transitively follow the base links using the additional links, so
that I finally have:
a --> d (over b and c)
e --> g

That's the problem, here's my delta iteration code (Not sure, what's best
practice for the code - append to the email or attach?):

override def getPlan(args: String*): Plan = {
// store the orig destination as well, because we need it for the key in
the delta iteration
case class BaseLink(from: String, origTo: String, to: String)
case class AdditionalLink(from: String, to: String)

// load datasets
val baseLinks       = DataSource(baseLinkPath,
CsvInputFormat[BaseLink]("\n", '\t'))
val additionalLinks = DataSource(additionalLinkPath,
CsvInputFormat[AdditionalLink]("\n", '\t'))

def iterate(s: DataSet[BaseLink], ws: DataSet[BaseLink]) = {
// first, join base links "to" with additional links "from", to do one
resolving step
val resolvedLinks = ws.join(additionalLinks)
.where { case BaseLink(from, to, orig) => to }
.isEqualTo { case AdditionalLink(from, to) => from }
.map { case (baseLink, additionalLink) =>
// If we had base link a --> b, and additional link b --> c, we now have a
--> c
BaseLink(baseLink.from, baseLink.origTo, additionalLink.to)
}
 // now, join with the solution set to build the solution delta and the
next working set
val nextWs = s.join(resolvedLinks)
.where { cl => (cl.to, cl.origTo) }
.isEqualTo { cl => (cl.to, cl.origTo) }
.map { (orig, resolved) =>
orig
}
// nextWs is solution delta as well
(nextWs, nextWs)
}
// start iteration, key is combination of from and orig to, e.g. a --> b
val resolvedLinks = baseLinks
.iterateWithDelta(baseLinks, { cl => (cl.from, cl.origTo) }, iterate, 10)
.map { cl => (cl.from, cl.to) }

val output = resolvedLinks.write(resolvedLinksPath,
CsvOutputFormat[(String, String)]("\n", "\t"))
val plan = new ScalaPlan(Seq(output))
plan
}

Here's the whole stacktrace:
Exception in thread "main"
eu.stratosphere.compiler.CompilerPostPassException: Could not set up
runtime strategy for input channel to node ''. Missing type information for
field 0
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:316)
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:228)
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.propagateToChannel(GenericFlatTypePostPass.java:499)
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:313)
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.propagateToChannel(GenericFlatTypePostPass.java:499)
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:91)
at
eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.postPass(GenericFlatTypePostPass.java:66)
at eu.stratosphere.api.scala.ScalaPostPass.postPass(ScalaPlan.scala:77)
at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:744)
at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:553)
at
eu.stratosphere.client.LocalExecutor.optimizerPlanAsJSON(LocalExecutor.java:320)
// error happens as well, when I directly call LocalExecutor.execute

Thanks in advance!
Stefan