docs: @Override
  	public DataSet<Vertex<K, VV>> createResult() {
  		if (vertexDataSet == null) {
  			throw new IllegalStateException("The input data set has not been set.");
  		}
  
  		// Prepare type information
  		TypeInformation<K> keyType = ((TupleTypeInfo<?>) vertexDataSet.getType()).getTypeAt(0);
  		TypeInformation<M> messageType = TypeExtractor.createTypeInfo(gather, GatherFunction.class, gather.getClass(), 2);
  		TypeInformation<Tuple2<K, M>> innerType = new TupleTypeInfo<>(keyType, messageType);
  		TypeInformation<Vertex<K, VV>> outputType = vertexDataSet.getType();
  
  		// check whether the numVertices option is set and, if so, compute the total number of vertices
  		// and set it within the gather, sum and apply functions
  
  		DataSet<LongValue> numberOfVertices = null;
  		if (this.configuration != null && this.configuration.isOptNumVertices()) {
  			try {
  				numberOfVertices = GraphUtils.count(this.vertexDataSet);
  			} catch (Exception e) {
  				e.printStackTrace();
  			}
  		}
  
  		// Prepare UDFs
  		GatherUdf<K, VV, EV, M> gatherUdf = new GatherUdf<>(gather, innerType);
  		SumUdf<K, VV, EV, M> sumUdf = new SumUdf<>(sum, innerType);
  		ApplyUdf<K, VV, EV, M> applyUdf = new ApplyUdf<>(apply, outputType);
  
  		final int[] zeroKeyPos = new int[] {0};
  		final DeltaIteration<Vertex<K, VV>, Vertex<K, VV>> iteration =
  				vertexDataSet.iterateDelta(vertexDataSet, maximumNumberOfIterations, zeroKeyPos);
  
  		// set up the iteration operator
  		if (this.configuration != null) {
  
  			iteration.name(this.configuration.getName(
  					"Gather-sum-apply iteration (" + gather + " | " + sum + " | " + apply + ")"));
  			iteration.parallelism(this.configuration.getParallelism());
  			iteration.setSolutionSetUnManaged(this.configuration.isSolutionSetUnmanagedMemory());
  
  			// register all aggregators
  			for (Map.Entry<String, Aggregator<?>> entry : this.configuration.getAggregators().entrySet()) {
  				iteration.registerAggregator(entry.getKey(), entry.getValue());
  			}
  		}
  		else {
  			// no configuration provided; set default name
  			iteration.name("Gather-sum-apply iteration (" + gather + " | " + sum + " | " + apply + ")");
  		}
  
  		// Prepare the neighbors
  		if (this.configuration != null) {
  			direction = this.configuration.getDirection();
  		}
  		DataSet<Tuple2<K, Neighbor<VV, EV>>> neighbors;
  		switch(direction) {
  			case OUT:
  				neighbors = iteration
  				.getWorkset().join(edgeDataSet)
  				.where(0).equalTo(0).with(new ProjectKeyWithNeighborOUT<>());
  				break;
  			case IN:
  				neighbors = iteration
  				.getWorkset().join(edgeDataSet)
  				.where(0).equalTo(1).with(new ProjectKeyWithNeighborIN<>());
  				break;
  			case ALL:
  				neighbors =  iteration
  						.getWorkset().join(edgeDataSet)
  						.where(0).equalTo(0).with(new ProjectKeyWithNeighborOUT<>()).union(iteration
  								.getWorkset().join(edgeDataSet)
  								.where(0).equalTo(1).with(new ProjectKeyWithNeighborIN<>()));
  				break;
  			default:
  				neighbors = iteration
  						.getWorkset().join(edgeDataSet)
  						.where(0).equalTo(0).with(new ProjectKeyWithNeighborOUT<>());
  				break;
  		}
  
  		// Gather, sum and apply
  		MapOperator<Tuple2<K, Neighbor<VV, EV>>, Tuple2<K, M>> gatherMapOperator = neighbors.map(gatherUdf);
  
  		// configure map gather function with name and broadcast variables
  		gatherMapOperator = gatherMapOperator.name("Gather");
  
  		if (this.configuration != null) {
  			for (Tuple2<String, DataSet<?>> e : this.configuration.getGatherBcastVars()) {
  				gatherMapOperator = gatherMapOperator.withBroadcastSet(e.f1, e.f0);
  			}
  			if (this.configuration.isOptNumVertices()) {
  				gatherMapOperator = gatherMapOperator.withBroadcastSet(numberOfVertices, "number of vertices");
  			}
  		}
  		DataSet<Tuple2<K, M>> gatheredSet = gatherMapOperator;
  
  		ReduceOperator<Tuple2<K, M>> sumReduceOperator = gatheredSet.groupBy(0).reduce(sumUdf);
  
  		// configure reduce sum function with name and broadcast variables
  		sumReduceOperator = sumReduceOperator.name("Sum");
  
  		if (this.configuration != null) {
  			for (Tuple2<String, DataSet<?>> e : this.configuration.getSumBcastVars()) {
  				sumReduceOperator = sumReduceOperator.withBroadcastSet(e.f1, e.f0);
  			}
  			if (this.configuration.isOptNumVertices()) {
  				sumReduceOperator = sumReduceOperator.withBroadcastSet(numberOfVertices, "number of vertices");
  			}
  		}
  		DataSet<Tuple2<K, M>> summedSet = sumReduceOperator;
  
  		JoinOperator<?, ?, Vertex<K, VV>> appliedSet = summedSet
  				.join(iteration.getSolutionSet())
  				.where(0)
  				.equalTo(0)
  				.with(applyUdf);
  
  		// configure join apply function with name and broadcast variables
  		appliedSet = appliedSet.name("Apply");
  
  		if (this.configuration != null) {
  			for (Tuple2<String, DataSet<?>> e : this.configuration.getApplyBcastVars()) {
  				appliedSet = appliedSet.withBroadcastSet(e.f1, e.f0);
  			}
  			if (this.configuration.isOptNumVertices()) {
  				appliedSet = appliedSet.withBroadcastSet(numberOfVertices, "number of vertices");
  			}
  		}
  
  		// let the operator know that we preserve the key field
  		appliedSet.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0");
  
  		return iteration.closeWith(appliedSet, appliedSet);
  	}