docs: @Override
  	public boolean preVisit(PlanNode node) {
  		// check if we have visited this node before. in non-tree graphs, this happens
  		if (this.vertices.containsKey(node) || this.chainedTasks.containsKey(node) || this.iterations.containsKey(node)) {
  			// return false to prevent further descend
  			return false;
  		}
  
  		// the vertex to be created for the current node
  		final JobVertex vertex;
  		try {
  			if (node instanceof SinkPlanNode) {
  				vertex = createDataSinkVertex((SinkPlanNode) node);
  			}
  			else if (node instanceof SourcePlanNode) {
  				vertex = createDataSourceVertex((SourcePlanNode) node);
  			}
  			else if (node instanceof BulkIterationPlanNode) {
  				BulkIterationPlanNode iterationNode = (BulkIterationPlanNode) node;
  				// for the bulk iteration, we skip creating anything for now. we create the graph
  				// for the step function in the post visit.
  				
  				// check that the root of the step function has the same parallelism as the iteration.
  				// because the tail must have the same parallelism as the head, we can only merge the last
  				// operator with the tail, if they have the same parallelism. not merging is currently not
  				// implemented
  				PlanNode root = iterationNode.getRootOfStepFunction();
  				if (root.getParallelism() != node.getParallelism())
  				{
  					throw new CompilerException("Error: The final operator of the step " +
  							"function has a different parallelism than the iteration operator itself.");
  				}
  				
  				IterationDescriptor descr = new IterationDescriptor(iterationNode, this.iterationIdEnumerator++);
  				this.iterations.put(iterationNode, descr);
  				vertex = null;
  			}
  			else if (node instanceof WorksetIterationPlanNode) {
  				WorksetIterationPlanNode iterationNode = (WorksetIterationPlanNode) node;
  
  				// we have the same constraints as for the bulk iteration
  				PlanNode nextWorkSet = iterationNode.getNextWorkSetPlanNode();
  				PlanNode solutionSetDelta  = iterationNode.getSolutionSetDeltaPlanNode();
  				
  				if (nextWorkSet.getParallelism() != node.getParallelism())
  				{
  					throw new CompilerException("It is currently not supported that the final operator of the step " +
  							"function has a different parallelism than the iteration operator itself.");
  				}
  				if (solutionSetDelta.getParallelism() != node.getParallelism())
  				{
  					throw new CompilerException("It is currently not supported that the final operator of the step " +
  							"function has a different parallelism than the iteration operator itself.");
  				}
  				
  				IterationDescriptor descr = new IterationDescriptor(iterationNode, this.iterationIdEnumerator++);
  				this.iterations.put(iterationNode, descr);
  				vertex = null;
  			}
  			else if (node instanceof SingleInputPlanNode) {
  				vertex = createSingleInputVertex((SingleInputPlanNode) node);
  			}
  			else if (node instanceof DualInputPlanNode) {
  				vertex = createDualInputVertex((DualInputPlanNode) node);
  			}
  			else if (node instanceof NAryUnionPlanNode) {
  				// skip the union for now
  				vertex = null;
  			}
  			else if (node instanceof BulkPartialSolutionPlanNode) {
  				// create a head node (or not, if it is merged into its successor)
  				vertex = createBulkIterationHead((BulkPartialSolutionPlanNode) node);
  			}
  			else if (node instanceof SolutionSetPlanNode) {
  				// this represents an access into the solution set index.
  				// we do not create a vertex for the solution set here (we create the head at the workset place holder)
  				
  				// we adjust the joins / cogroups that go into the solution set here
  				for (Channel c : node.getOutgoingChannels()) {
  					DualInputPlanNode target = (DualInputPlanNode) c.getTarget();
  					JobVertex accessingVertex = this.vertices.get(target);
  					TaskConfig conf = new TaskConfig(accessingVertex.getConfiguration());
  					int inputNum = c == target.getInput1() ? 0 : c == target.getInput2() ? 1 : -1;
  					
  					// sanity checks
  					if (inputNum == -1) {
  						throw new CompilerException();
  					}
  					
  					// adjust the driver
  					if (conf.getDriver().equals(JoinDriver.class)) {
  						conf.setDriver(inputNum == 0 ? JoinWithSolutionSetFirstDriver.class : JoinWithSolutionSetSecondDriver.class);
  					}
  					else if (conf.getDriver().equals(CoGroupDriver.class)) {
  						conf.setDriver(inputNum == 0 ? CoGroupWithSolutionSetFirstDriver.class : CoGroupWithSolutionSetSecondDriver.class);
  					}
  					else {
  						throw new CompilerException("Found join with solution set using incompatible operator (only Join/CoGroup are valid).");
  					}
  				}
  				
  				// make sure we do not visit this node again. for that, we add a 'already seen' entry into one of the sets
  				this.chainedTasks.put(node, ALREADY_VISITED_PLACEHOLDER);
  				
  				vertex = null;
  			}
  			else if (node instanceof WorksetPlanNode) {
  				// create the iteration head here
  				vertex = createWorksetIterationHead((WorksetPlanNode) node);
  			}
  			else {
  				throw new CompilerException("Unrecognized node type: " + node.getClass().getName());
  			}
  		}
  		catch (Exception e) {
  			throw new CompilerException("Error translating node '" + node + "': " + e.getMessage(), e);
  		}
  		
  		// check if a vertex was created, or if it was chained or skipped
  		if (vertex != null) {
  			// set parallelism
  			int pd = node.getParallelism();
  			vertex.setParallelism(pd);
  			vertex.setMaxParallelism(pd);
  			
  			vertex.setSlotSharingGroup(sharingGroup);
  			
  			// check whether this vertex is part of an iteration step function
  			if (this.currentIteration != null) {
  				// check that the task has the same parallelism as the iteration as such
  				PlanNode iterationNode = (PlanNode) this.currentIteration;
  				if (iterationNode.getParallelism() < pd) {
  					throw new CompilerException("Error: All functions that are part of an iteration must have the same, or a lower, parallelism than the iteration operator.");
  				}
  
  				// store the id of the iterations the step functions participate in
  				IterationDescriptor descr = this.iterations.get(this.currentIteration);
  				new TaskConfig(vertex.getConfiguration()).setIterationId(descr.getId());
  			}
  	
  			// store in the map
  			this.vertices.put(node, vertex);
  		}
  
  		// returning true causes deeper descend
  		return true;
  	}