docs: public DataSet<Tuple2<K, VV>> reduceOnNeighbors(ReduceNeighborsFunction<VV> reduceNeighborsFunction,
  			EdgeDirection direction) throws IllegalArgumentException {
  		switch (direction) {
  			case IN:
  				// create <vertex-source value> pairs
  				final DataSet<Tuple2<K, VV>> verticesWithSourceNeighborValues = edges
  						.join(this.vertices).where(0).equalTo(0)
  						.with(new ProjectVertexWithNeighborValueJoin<>(1))
  						.withForwardedFieldsFirst("f1->f0").name("Vertex with in-neighbor value");
  				return verticesWithSourceNeighborValues.groupBy(0).reduce(new ApplyNeighborReduceFunction<>(
  					reduceNeighborsFunction)).name("Neighbors function");
  			case OUT:
  				// create <vertex-target value> pairs
  				DataSet<Tuple2<K, VV>> verticesWithTargetNeighborValues = edges
  						.join(this.vertices).where(1).equalTo(0)
  						.with(new ProjectVertexWithNeighborValueJoin<>(0))
  						.withForwardedFieldsFirst("f0").name("Vertex with out-neighbor value");
  				return verticesWithTargetNeighborValues.groupBy(0).reduce(new ApplyNeighborReduceFunction<>(
  					reduceNeighborsFunction)).name("Neighbors function");
  			case ALL:
  				// create <vertex-neighbor value> pairs
  				DataSet<Tuple2<K, VV>> verticesWithNeighborValues = edges
  						.flatMap(new EmitOneEdgeWithNeighborPerNode<>())
  						.join(this.vertices).where(1).equalTo(0)
  						.with(new ProjectNeighborValue<>()).name("Vertex with neighbor value");
  
  				return verticesWithNeighborValues.groupBy(0).reduce(new ApplyNeighborReduceFunction<>(
  					reduceNeighborsFunction)).name("Neighbors function");
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  	}