docs: @Override
  	public DataSet<Vertex<K, VV>> createResult() {
  		if (this.initialVertices == null) {
  			throw new IllegalStateException("The input data set has not been set.");
  		}
  
  		// prepare the type information
  		TypeInformation<K> keyType = ((TupleTypeInfo<?>) initialVertices.getType()).getTypeAt(0);
  		TypeInformation<Tuple2<K, Message>> messageTypeInfo =
  			new TupleTypeInfo<>(keyType, messageType);
  		TypeInformation<Vertex<K, VV>> vertexType = initialVertices.getType();
  		TypeInformation<Either<Vertex<K, VV>, Tuple2<K, Message>>> intermediateTypeInfo =
  			new EitherTypeInfo<>(vertexType, messageTypeInfo);
  		TypeInformation<Either<NullValue, Message>> nullableMsgTypeInfo =
  			new EitherTypeInfo<>(TypeExtractor.getForClass(NullValue.class), messageType);
  		TypeInformation<Tuple2<K, Either<NullValue, Message>>> workSetTypeInfo =
  			new TupleTypeInfo<>(keyType, nullableMsgTypeInfo);
  
  		DataSet<Tuple2<K, Either<NullValue, Message>>> initialWorkSet = initialVertices.map(
  				new InitializeWorkSet<K, VV, Message>()).returns(workSetTypeInfo);
  
  		final DeltaIteration<Vertex<K, VV>, Tuple2<K, Either<NullValue, Message>>> iteration =
  				initialVertices.iterateDelta(initialWorkSet, this.maximumNumberOfIterations, 0);
  		setUpIteration(iteration);
  
  		// join with the current state to get vertex values
  		DataSet<Tuple2<Vertex<K, VV>, Either<NullValue, Message>>> verticesWithMsgs =
  				iteration.getSolutionSet().join(iteration.getWorkset())
  				.where(0).equalTo(0)
  				.with(new AppendVertexState<>())
  				.returns(new TupleTypeInfo<>(
  					vertexType, nullableMsgTypeInfo));
  
  		VertexComputeUdf<K, VV, EV, Message> vertexUdf =
  			new VertexComputeUdf<>(computeFunction, intermediateTypeInfo);
  
  		CoGroupOperator<?, ?, Either<Vertex<K, VV>, Tuple2<K, Message>>> superstepComputation =
  				verticesWithMsgs.coGroup(edgesWithValue)
  				.where("f0.f0").equalTo(0)
  				.with(vertexUdf);
  
  		// compute the solution set delta
  		DataSet<Vertex<K, VV>> solutionSetDelta = superstepComputation.flatMap(
  			new ProjectNewVertexValue<>()).returns(vertexType);
  
  		// compute the inbox of each vertex for the next superstep (new workset)
  		DataSet<Tuple2<K, Either<NullValue, Message>>> allMessages = superstepComputation.flatMap(
  			new ProjectMessages<>()).returns(workSetTypeInfo);
  
  		DataSet<Tuple2<K, Either<NullValue, Message>>> newWorkSet = allMessages;
  
  		// check if a combiner has been provided
  		if (combineFunction != null) {
  
  			MessageCombinerUdf<K, Message> combinerUdf =
  				new MessageCombinerUdf<>(combineFunction, workSetTypeInfo);
  
  			DataSet<Tuple2<K, Either<NullValue, Message>>> combinedMessages = allMessages
  					.groupBy(0).reduceGroup(combinerUdf)
  					.setCombinable(true);
  
  			newWorkSet = combinedMessages;
  		}
  
  		// configure the compute function
  		superstepComputation = superstepComputation.name("Compute Function");
  		if (this.configuration != null) {
  			for (Tuple2<String, DataSet<?>> e : this.configuration.getBcastVars()) {
  				superstepComputation = superstepComputation.withBroadcastSet(e.f1, e.f0);
  			}
  		}
  
  		return iteration.closeWith(solutionSetDelta, newWorkSet);
  	}