docs: public <T> DataSet<T> groupReduceOnNeighbors(NeighborsFunction<K, VV, EV, T> neighborsFunction,
  			EdgeDirection direction, TypeInformation<T> typeInfo) throws IllegalArgumentException {
  		switch (direction) {
  			case IN:
  				// create <edge-sourceVertex> pairs
  				DataSet<Tuple3<K, Edge<K, EV>, Vertex<K, VV>>> edgesWithSources = edges
  						.join(this.vertices).where(0).equalTo(0)
  						.with(new ProjectVertexIdJoin<>(1))
  						.withForwardedFieldsFirst("f1->f0").name("Edge with source vertex ID");
  				return edgesWithSources.groupBy(0).reduceGroup(
  					new ApplyNeighborGroupReduceFunction<>(neighborsFunction))
  						.name("Neighbors function").returns(typeInfo);
  			case OUT:
  				// create <edge-targetVertex> pairs
  				DataSet<Tuple3<K, Edge<K, EV>, Vertex<K, VV>>> edgesWithTargets = edges
  						.join(this.vertices).where(1).equalTo(0)
  						.with(new ProjectVertexIdJoin<>(0))
  						.withForwardedFieldsFirst("f0").name("Edge with target vertex ID");
  				return edgesWithTargets.groupBy(0).reduceGroup(
  					new ApplyNeighborGroupReduceFunction<>(neighborsFunction))
  						.name("Neighbors function").returns(typeInfo);
  			case ALL:
  				// create <edge-sourceOrTargetVertex> pairs
  				DataSet<Tuple3<K, Edge<K, EV>, Vertex<K, VV>>> edgesWithNeighbors = edges
  						.flatMap(new EmitOneEdgeWithNeighborPerNode<>())
  						.join(this.vertices).where(1).equalTo(0)
  						.with(new ProjectEdgeWithNeighbor<>()).name("Edge with vertex ID");
  
  				return edgesWithNeighbors.groupBy(0).reduceGroup(
  					new ApplyNeighborGroupReduceFunction<>(neighborsFunction))
  						.name("Neighbors function").returns(typeInfo);
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  	}