[FLINK-5563] [gelly] Add density to vertex metrics

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/VertexMetrics.java

@@ -251,12 +251,27 @@ public long getNumberOfUndirectedEdges() {
 		}
 
 		/**
-		 * Get the average degree.
+		 * Get the average degree, the average number of in- plus out-edges per vertex.
+		 *
+		 * A result of {@code Float.NaN} is returned for an empty graph for
+		 * which both the number of edges and number of vertices is zero.
 		 *
 		 * @return average degree
 		 */
 		public float getAverageDegree() {
-			return getNumberOfEdges() / (float)vertexCount;
+			return vertexCount == 0 ? Float.NaN : getNumberOfEdges() / (float)vertexCount;
+		}
+
+		/**
+		 * Get the density, the ratio of actual to potential edges between vertices.
+		 *
+		 * A result of {@code Float.NaN} is returned for a graph with fewer than
+		 * two vertices for which the number of edges is zero.
+		 *
+		 * @return density
+		 */
+		public float getDensity() {
+			return vertexCount <= 1 ? Float.NaN : getNumberOfEdges() / (float)(vertexCount*(vertexCount-1));
 		}
 
 		/**
@@ -313,6 +328,7 @@ public String toString() {
 				+ "; unidirectional edge count: " + nf.format(unidirectionalEdgeCount)
 				+ "; bidirectional edge count: " + nf.format(bidirectionalEdgeCount)
 				+ "; average degree: " + nf.format(getAverageDegree())
+				+ "; density: " + nf.format(getDensity())
 				+ "; triplet count: " + nf.format(tripletCount)
 				+ "; maximum degree: " + nf.format(maximumDegree)
 				+ "; maximum out degree: " + nf.format(maximumOutDegree)

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/VertexMetrics.java

@@ -208,7 +208,8 @@ public long getNumberOfVertices() {
 		}
 
 		/**
-		 * Get the number of edges.
+		 * Get the number of edges. Each edge is counted once even though Gelly
+		 * stores undirected edges twice, once in each direction.
 		 *
 		 * @return number of edges
 		 */
@@ -217,12 +218,28 @@ public long getNumberOfEdges() {
 		}
 
 		/**
-		 * Get the average degree.
+		 * Get the average degree, the average number of edges per vertex.
+		 *
+		 * A result of {@code Float.NaN} is returned for an empty graph for
+		 * which both the number of edges and number of vertices is zero.
 		 *
 		 * @return average degree
 		 */
 		public float getAverageDegree() {
-			return edgeCount / (float)vertexCount;
+			// each edge is incident on two vertices
+			return vertexCount == 0 ? Float.NaN : 2 * edgeCount / (float)vertexCount;
+		}
+
+		/**
+		 * Get the density, the ratio of actual to potential edges between vertices.
+		 *
+		 * A result of {@code Float.NaN} is returned for a graph with fewer than
+		 * two vertices for which the number of edges is zero.
+		 *
+		 * @return density
+		 */
+		public float getDensity() {
+			return vertexCount <= 1 ? Float.NaN : edgeCount / (float)(vertexCount*(vertexCount-1)/2);
 		}
 
 		/**
@@ -259,6 +276,7 @@ public String toString() {
 			return "vertex count: " + nf.format(vertexCount)
 				+ "; edge count: " + nf.format(edgeCount)
 				+ "; average degree: " + nf.format(getAverageDegree())
+				+ "; density: " + nf.format(getDensity())
 				+ "; triplet count: " + nf.format(tripletCount)
 				+ "; maximum degree: " + nf.format(maximumDegree)
 				+ "; maximum triplets: " + nf.format(maximumTriplets);

flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/asm/AsmTestBase.java

@@ -37,6 +37,8 @@ public class AsmTestBase {
 
 	protected ExecutionEnvironment env;
 
+	protected final double ACCURACY = 0.000001;
+
 	// simple graph
 	protected Graph<IntValue, NullValue, NullValue> directedSimpleGraph;
 
@@ -61,6 +63,7 @@ public class AsmTestBase {
 	public void setup()
 			throws Exception {
 		env = ExecutionEnvironment.createCollectionsEnvironment();
+		env.getConfig().enableObjectReuse();
 
 		// the "fish" graph
 		Object[][] edges = new Object[][] {
@@ -98,9 +101,17 @@ public void setup()
 		Graph<LongValue, NullValue, NullValue> rmatGraph = new RMatGraph<>(env, new JDKRandomGeneratorFactory(), rmatVertexCount, rmatEdgeCount)
 			.generate();
 
+		/*
+			./bin/flink run -c org.apache.flink.graph.drivers.Graph500 flink-gelly-examples_2.10-1.2-SNAPSHOT.jar \
+				--directed true --simplify true --scale 10 --edge_factor 16 --output csv --filename directedRMatGraph.csv
+		 */
 		directedRMatGraph = rmatGraph
 			.run(new org.apache.flink.graph.asm.simple.directed.Simplify<LongValue, NullValue, NullValue>());
 
+		/*
+			./bin/flink run -c org.apache.flink.graph.drivers.Graph500 flink-gelly-examples_2.10-1.2-SNAPSHOT.jar \
+				--directed false --simplify true --scale 10 --edge_factor 16 --output csv --filename undirectedRMatGraph.csv
+		 */
 		undirectedRMatGraph = rmatGraph
 			.run(new org.apache.flink.graph.asm.simple.undirected.Simplify<LongValue, NullValue, NullValue>(false));
 	}

flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/VertexMetricsTest.java

@@ -59,6 +59,8 @@ public void testWithCompleteGraph()
 			.execute();
 
 		assertEquals(expectedResult, vertexMetrics);
+		assertEquals(expectedDegree, vertexMetrics.getAverageDegree(), ACCURACY);
+		assertEquals(1.0f, vertexMetrics.getDensity(), ACCURACY);
 	}
 
 	@Test
@@ -73,7 +75,9 @@ public void testWithEmptyGraph()
 			.run(emptyGraph)
 			.execute();
 
-		assertEquals(withoutZeroDegreeVertices, expectedResult);
+		assertEquals(expectedResult, withoutZeroDegreeVertices);
+		assertEquals(Float.NaN, withoutZeroDegreeVertices.getAverageDegree(), ACCURACY);
+		assertEquals(Float.NaN, withoutZeroDegreeVertices.getDensity(), ACCURACY);
 
 		expectedResult = new Result(3, 0, 0, 0, 0, 0, 0, 0);
 
@@ -83,6 +87,8 @@ public void testWithEmptyGraph()
 			.execute();
 
 		assertEquals(expectedResult, withZeroDegreeVertices);
+		assertEquals(0.0f, withZeroDegreeVertices.getAverageDegree(), ACCURACY);
+		assertEquals(0.0f, withZeroDegreeVertices.getDensity(), ACCURACY);
 	}
 
 	@Test

flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/VertexMetricsTest.java

@@ -59,6 +59,8 @@ public void testWithCompleteGraph()
 			.execute();
 
 		assertEquals(expectedResult, vertexMetrics);
+		assertEquals(expectedDegree, vertexMetrics.getAverageDegree(), ACCURACY);
+		assertEquals(1.0f, vertexMetrics.getDensity(), ACCURACY);
 	}
 
 	@Test
@@ -73,7 +75,9 @@ public void testWithEmptyGraph()
 			.run(emptyGraph)
 			.execute();
 
-		assertEquals(withoutZeroDegreeVertices, expectedResult);
+		assertEquals(expectedResult, withoutZeroDegreeVertices);
+		assertEquals(Float.NaN, withoutZeroDegreeVertices.getAverageDegree(), ACCURACY);
+		assertEquals(Float.NaN, withoutZeroDegreeVertices.getDensity(), ACCURACY);
 
 		expectedResult = new Result(3, 0, 0, 0, 0);
 
@@ -83,6 +87,8 @@ public void testWithEmptyGraph()
 			.execute();
 
 		assertEquals(expectedResult, withZeroDegreeVertices);
+		assertEquals(0.0f, withZeroDegreeVertices.getAverageDegree(), ACCURACY);
+		assertEquals(0.0f, withZeroDegreeVertices.getDensity(), ACCURACY);
 	}
 
 	@Test