docs: public <T> DataSet<T> groupReduceOnEdges(EdgesFunction<K, EV, T> edgesFunction,
  			EdgeDirection direction, TypeInformation<T> typeInfo) throws IllegalArgumentException {
  
  		switch (direction) {
  			case IN:
  				return edges.map(new ProjectVertexIdMap<>(1)).name("Vertex ID")
  						.withForwardedFields("f1->f0")
  						.groupBy(0).reduceGroup(new ApplyGroupReduceFunction<>(edgesFunction))
  							.name("GroupReduce on in-edges").returns(typeInfo);
  			case OUT:
  				return edges.map(new ProjectVertexIdMap<>(0)).name("Vertex ID")
  						.withForwardedFields("f0")
  						.groupBy(0).reduceGroup(new ApplyGroupReduceFunction<>(edgesFunction))
  							.name("GroupReduce on out-edges").returns(typeInfo);
  			case ALL:
  				return edges.flatMap(new EmitOneEdgePerNode<>()).name("Emit edge")
  						.groupBy(0).reduceGroup(new ApplyGroupReduceFunction<>(edgesFunction))
  							.name("GroupReduce on in- and out-edges").returns(typeInfo);
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  	}