You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gr...@apache.org on 2017/07/26 14:32:06 UTC
[1/5] flink git commit: [hotfix] [gelly] Reduce time to run
integration tests
Repository: flink
Updated Branches:
refs/heads/master 3e12673a5 -> 94559bae0
[hotfix] [gelly] Reduce time to run integration tests
Removes the "large" tests and increases the size of the "small" tests
from scale=7 to scale=8.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/94559bae
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/94559bae
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/94559bae
Branch: refs/heads/master
Commit: 94559bae086e4285b24e6fc29bb89007cfa71a33
Parents: 4a88f65
Author: Greg Hogan <co...@greghogan.com>
Authored: Wed Jul 26 07:08:46 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed Jul 26 10:31:20 2017 -0400
----------------------------------------------------------------------
.../flink/graph/drivers/AdamicAdarITCase.java | 7 +-
.../drivers/ClusteringCoefficientITCase.java | 62 +++-------
.../drivers/ConnectedComponentsITCase.java | 18 +--
.../flink/graph/drivers/EdgeListITCase.java | 3 +-
.../flink/graph/drivers/GraphMetricsITCase.java | 68 +----------
.../apache/flink/graph/drivers/HITSITCase.java | 16 +--
.../flink/graph/drivers/JaccardIndexITCase.java | 22 +---
.../flink/graph/drivers/PageRankITCase.java | 16 +--
.../graph/drivers/TriangleListingITCase.java | 115 ++++++-------------
9 files changed, 63 insertions(+), 264 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
index b048f08..6650d0d 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
@@ -30,8 +30,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link AdamicAdar}.
*/
@RunWith(Parameterized.class)
-public class AdamicAdarITCase
-extends CopyableValueDriverBaseITCase {
+public class AdamicAdarITCase extends CopyableValueDriverBaseITCase {
public AdamicAdarITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -61,8 +60,6 @@ extends CopyableValueDriverBaseITCase {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- expectedCount(
- parameters(7, "print"),
- 11388);
+ expectedCount(parameters(8, "print"), 39276);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
index 6bcab6b..a31e862 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
@@ -21,7 +21,6 @@ package org.apache.flink.graph.drivers;
import org.apache.flink.client.program.ProgramParametrizationException;
import org.apache.flink.graph.asm.dataset.ChecksumHashCode.Checksum;
-import org.junit.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -30,8 +29,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link ClusteringCoefficient}.
*/
@RunWith(Parameterized.class)
-public class ClusteringCoefficientITCase
-extends CopyableValueDriverBaseITCase {
+public class ClusteringCoefficientITCase extends CopyableValueDriverBaseITCase {
public ClusteringCoefficientITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -55,58 +53,24 @@ extends CopyableValueDriverBaseITCase {
}
@Test
- public void testHashWithSmallDirectedRMatGraph() throws Exception {
+ public void testHashWithDirectedRMatGraph() throws Exception {
String expected = "\n" +
- new Checksum(117, 0x0000003621c62ca1L) + "\n\n" +
- "triplet count: 29286, triangle count: 11466, global clustering coefficient: 0.39151813[0-9]+\n" +
- "vertex count: 117, average clustering coefficient: 0.45125697[0-9]+\n";
+ new Checksum(233, 0x00000075d99a55b6L) + "\n\n" +
+ "triplet count: 97315, triangle count: 30705, global clustering coefficient: 0.31552175[0-9]+\n" +
+ "vertex count: 233, average clustering coefficient: 0.41178524[0-9]+\n";
- expectedOutput(parameters(7, "directed", "directed", "hash"), expected);
+ expectedOutput(parameters(8, "directed", "directed", "hash"), expected);
}
@Test
- public void testHashWithSmallUndirectedRMatGraph() throws Exception {
+ public void testHashWithUndirectedRMatGraph() throws Exception {
String expected = "\n\n" +
- "triplet count: 29286, triangle count: 11466, global clustering coefficient: 0.39151813[0-9]+\n" +
- "vertex count: 117, average clustering coefficient: 0.57438679[0-9]+\n";
+ "triplet count: 97315, triangle count: 30705, global clustering coefficient: 0.31552175[0-9]+\n" +
+ "vertex count: 233, average clustering coefficient: 0.50945459[0-9]+\n";
- expectedOutput(parameters(7, "directed", "undirected", "hash"),
- "\n" + new Checksum(117, 0x0000003875b38c43L) + expected);
- expectedOutput(parameters(7, "undirected", "undirected", "hash"),
- "\n" + new Checksum(117, 0x0000003c20344c75L) + expected);
- }
-
- @Test
- public void testHashWithLargeDirectedRMatGraph() throws Exception {
- // computation is too large for collection mode
- Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- String expected = "\n" +
- new Checksum(3349, 0x0000067a9d18e7f3L) + "\n\n" +
- "triplet count: 9276207, triangle count: 1439454, global clustering coefficient: 0.15517700[0-9]+\n" +
- "vertex count: 3349, average clustering coefficient: 0.24571815[0-9]+\n";
-
- expectedOutput(parameters(12, "directed", "directed", "hash"), expected);
- }
-
- @Test
- public void testHashWithLargeUndirectedRMatGraph() throws Exception {
- // computation is too large for collection mode
- Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- String expected = "\n\n" +
- "triplet count: 9276207, triangle count: 1439454, global clustering coefficient: 0.15517700[0-9]+\n" +
- "vertex count: 3349, average clustering coefficient: 0.33029442[0-9]+\n";
-
- expectedOutput(parameters(12, "directed", "undirected", "hash"),
- "\n" + new Checksum(3349, 0x00000681fad1587eL) + expected);
- expectedOutput(parameters(12, "undirected", "undirected", "hash"),
- "\n" + new Checksum(3349, 0x0000068713b3b7f1L) + expected);
+ expectedOutput(parameters(8, "directed", "undirected", "hash"),
+ "\n" + new Checksum(233, 0x00000076635e00e2L) + expected);
+ expectedOutput(parameters(8, "undirected", "undirected", "hash"),
+ "\n" + new Checksum(233, 0x000000743ef6d14bL) + expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
index d6913c4..d14412d 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
@@ -30,8 +30,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link ConnectedComponents}.
*/
@RunWith(Parameterized.class)
-public class ConnectedComponentsITCase
-extends NonTransformableDriverBaseITCase {
+public class ConnectedComponentsITCase extends NonTransformableDriverBaseITCase {
public ConnectedComponentsITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -56,23 +55,12 @@ extends NonTransformableDriverBaseITCase {
}
@Test
- public void testHashWithSmallRMatGraph() throws Exception {
+ public void testHashWithRMatGraph() throws Exception {
expectedChecksum(parameters(7, "hash"), 106, 0x0000000000033e88L);
}
@Test
- public void testHashWithLargeRMatGraph() throws Exception {
- // computation is too large for collection mode
- Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- expectedChecksum(parameters(15, "hash"), 25572, 0x00000003094ffba2L);
- }
-
- @Test
- public void testPrintWithSmallRMatGraph() throws Exception {
+ public void testPrintWithRMatGraph() throws Exception {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
index 2f8eb0d..6d84833 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
@@ -31,8 +31,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link EdgeList}.
*/
@RunWith(Parameterized.class)
-public class EdgeListITCase
-extends NonTransformableDriverBaseITCase {
+public class EdgeListITCase extends NonTransformableDriverBaseITCase {
public EdgeListITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/GraphMetricsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/GraphMetricsITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/GraphMetricsITCase.java
index 1045a38..f417f53 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/GraphMetricsITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/GraphMetricsITCase.java
@@ -20,7 +20,6 @@ package org.apache.flink.graph.drivers;
import org.apache.flink.client.program.ProgramParametrizationException;
-import org.junit.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -29,8 +28,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link GraphMetrics}.
*/
@RunWith(Parameterized.class)
-public class GraphMetricsITCase
-extends DriverBaseITCase {
+public class GraphMetricsITCase extends DriverBaseITCase {
public GraphMetricsITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -54,7 +52,7 @@ extends DriverBaseITCase {
}
@Test
- public void testWithSmallDirectedRMatIntegerGraph() throws Exception {
+ public void testWithDirectedRMatGraph() throws Exception {
String expected = "\n" +
"Vertex metrics:\n" +
" vertex count: 117\n" +
@@ -80,39 +78,7 @@ extends DriverBaseITCase {
}
@Test
- public void testWithLargeDirectedRMatIntegerGraph() throws Exception {
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- // skip 'string' which does not compare numerically and generates a different triangle count
- Assume.assumeFalse(idType.equals("string") || idType.equals("nativeString"));
-
- String expected = "\n" +
- "Vertex metrics:\n" +
- " vertex count: 3,349\n" +
- " edge count: 53,368\n" +
- " unidirectional edge count: 43,602\n" +
- " bidirectional edge count: 4,883\n" +
- " average degree: 15.936\n" +
- " density: 0.00475971\n" +
- " triplet count: 9,276,207\n" +
- " maximum degree: 1,356\n" +
- " maximum out degree: 921\n" +
- " maximum in degree: 966\n" +
- " maximum triplets: 918,690\n" +
- "\n" +
- "Edge metrics:\n" +
- " triangle triplet count: 779,202\n" +
- " rectangle triplet count: 2,506,371\n" +
- " maximum triangle triplets: 3,160\n" +
- " maximum rectangle triplets: 16,835\n";
-
- expectedOutput(parameters(12, "directed", "hash"), expected);
- expectedOutput(parameters(12, "directed", "print"), expected);
- }
-
- @Test
- public void testWithSmallUndirectedRMatIntegerGraph() throws Exception {
+ public void testWithUndirectedRMatGraph() throws Exception {
String expected = "\n" +
"Vertex metrics:\n" +
" vertex count: 117\n" +
@@ -132,32 +98,4 @@ extends DriverBaseITCase {
expectedOutput(parameters(7, "undirected", "hash"), expected);
expectedOutput(parameters(7, "undirected", "print"), expected);
}
-
- @Test
- public void testWithLargelUndirectedRMatIntegerGraph() throws Exception {
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- // skip 'string' which does not compare numerically and generates a different triangle count
- Assume.assumeFalse(idType.equals("string") || idType.equals("nativeString"));
-
- String expected = "\n" +
- "Vertex metrics:\n" +
- " vertex count: 3,349\n" +
- " edge count: 48,485\n" +
- " average degree: 28.955\n" +
- " density: 0.00864842\n" +
- " triplet count: 9,276,207\n" +
- " maximum degree: 1,356\n" +
- " maximum triplets: 918,690\n" +
- "\n" +
- "Edge metrics:\n" +
- " triangle triplet count: 779,202\n" +
- " rectangle triplet count: 2,506,371\n" +
- " maximum triangle triplets: 3,160\n" +
- " maximum rectangle triplets: 16,835\n";
-
- expectedOutput(parameters(12, "undirected", "hash"), expected);
- expectedOutput(parameters(12, "undirected", "print"), expected);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/HITSITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/HITSITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/HITSITCase.java
index 51a63eb..72048ee 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/HITSITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/HITSITCase.java
@@ -29,8 +29,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link HITS}.
*/
@RunWith(Parameterized.class)
-public class HITSITCase
-extends DriverBaseITCase {
+public class HITSITCase extends DriverBaseITCase {
public HITSITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -54,21 +53,10 @@ extends DriverBaseITCase {
}
@Test
- public void testPrintWithSmallRMatGraph() throws Exception {
+ public void testPrintWithRMatGraph() throws Exception {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
expectedCount(parameters(8, "print"), 233);
}
-
- @Test
- public void testPrintWithLargeRMatGraph() throws Exception {
- // skip 'char' since it is not printed as a number
- Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- expectedCount(parameters(12, "print"), 3349);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
index 25eec22..a350bf0 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
@@ -31,8 +31,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link JaccardIndex}.
*/
@RunWith(Parameterized.class)
-public class JaccardIndexITCase
-extends CopyableValueDriverBaseITCase {
+public class JaccardIndexITCase extends CopyableValueDriverBaseITCase {
public JaccardIndexITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -58,26 +57,15 @@ extends CopyableValueDriverBaseITCase {
}
@Test
- public void testHashWithSmallRMatGraph() throws Exception {
- expectedChecksum(parameters(7, "hash"), 11388, 0x0000164757052eebL);
+ public void testHashWithRMatGraph() throws Exception {
+ expectedChecksum(parameters(8, "hash"), 39276, 0x00004caba2e663d5L);
}
@Test
- public void testHashWithLargeRMatGraph() throws Exception {
- // computation is too large for collection mode
- Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- expectedChecksum(parameters(12, "hash"), 4432058, 0x0021ce158d811c4eL);
- }
-
- @Test
- public void testPrintWithSmallRMatGraph() throws Exception {
+ public void testPrintWithRMatGraph() throws Exception {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- expectedOutputChecksum(parameters(7, "print"), new Checksum(11388, 0x0000163b17088256L));
+ expectedOutputChecksum(parameters(8, "print"), new Checksum(39276, 0x00004c5a726220c0L));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/PageRankITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/PageRankITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/PageRankITCase.java
index 8e86810..2d772ef 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/PageRankITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/PageRankITCase.java
@@ -29,8 +29,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link PageRank}.
*/
@RunWith(Parameterized.class)
-public class PageRankITCase
-extends DriverBaseITCase {
+public class PageRankITCase extends DriverBaseITCase {
public PageRankITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -54,21 +53,10 @@ extends DriverBaseITCase {
}
@Test
- public void testPrintWithSmallRMatGraph() throws Exception {
+ public void testPrintWithRMatGraph() throws Exception {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
expectedCount(parameters(8, "print"), 233);
}
-
- @Test
- public void testPrintWithLargeRMatGraph() throws Exception {
- // skip 'char' since it is not printed as a number
- Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- expectedCount(parameters(12, "print"), 3349);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94559bae/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
index 687610e..4331702 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
@@ -31,8 +31,7 @@ import org.junit.runners.Parameterized;
* Tests for {@link TriangleListing}.
*/
@RunWith(Parameterized.class)
-public class TriangleListingITCase
-extends CopyableValueDriverBaseITCase {
+public class TriangleListingITCase extends CopyableValueDriverBaseITCase {
public TriangleListingITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -62,106 +61,56 @@ extends CopyableValueDriverBaseITCase {
}
@Test
- public void testHashWithSmallDirectedRMatGraph() throws Exception {
+ public void testHashWithDirectedRMatGraph() throws Exception {
String expected = "\n" +
- new Checksum(22932, 0x00002cdfb573650dL) + "\n\n" +
+ new Checksum(61410, 0x000077d3e5e69fa3L) + "\n\n" +
"Triadic census:\n" +
- " 003: 178,989\n" +
- " 012: 47,736\n" +
- " 102: 11,763\n" +
- " 021d: 2,258\n" +
- " 021u: 2,064\n" +
- " 021c: 4,426\n" +
- " 111d: 3,359\n" +
- " 111u: 3,747\n" +
- " 030t: 624\n" +
- " 030c: 220\n" +
- " 201: 1,966\n" +
- " 120d: 352\n" +
- " 120u: 394\n" +
- " 120c: 704\n" +
- " 210: 1,120\n" +
- " 300: 408\n";
-
- expectedOutput(parameters(7, "directed", "hash"), expected);
+ " 003: 1,679,209\n" +
+ " 012: 267,130\n" +
+ " 102: 57,972\n" +
+ " 021d: 8,496\n" +
+ " 021u: 8,847\n" +
+ " 021c: 17,501\n" +
+ " 111d: 13,223\n" +
+ " 111u: 12,865\n" +
+ " 030t: 1,674\n" +
+ " 030c: 572\n" +
+ " 201: 5,678\n" +
+ " 120d: 1,066\n" +
+ " 120u: 896\n" +
+ " 120c: 2,011\n" +
+ " 210: 2,867\n" +
+ " 300: 1,149\n";
+
+ expectedOutput(parameters(8, "directed", "hash"), expected);
}
@Test
- public void testHashWithSmallUndirectedRMatGraph() throws Exception {
+ public void testHashWithUndirectedRMatGraph() throws Exception {
String expected = "\n" +
- new Checksum(22932, 0x00002c7fb95ec78eL) + "\n\n" +
+ new Checksum(61410, 0x000077ea1798a4e0L) + "\n\n" +
"Triadic census:\n" +
- " 03: 178,989\n" +
- " 12: 59,499\n" +
- " 21: 17,820\n" +
- " 30: 3,822\n";
+ " 03: 1,679,209\n" +
+ " 12: 325,102\n" +
+ " 21: 66,610\n" +
+ " 30: 10,235\n";
- expectedOutput(parameters(7, "undirected", "hash"), expected);
+ expectedOutput(parameters(8, "undirected", "hash"), expected);
}
@Test
- public void testHashWithLargeDirectedRMatGraph() throws Exception {
- // computation is too large for collection mode
- Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- String expected = "\n" +
- new Checksum(2878908, 0x0015f77b8984c7caL) + "\n\n" +
- "Triadic census:\n" +
- " 003: 6,101,196,568\n" +
- " 012: 132,051,207\n" +
- " 102: 13,115,128\n" +
- " 021d: 1,330,423\n" +
- " 021u: 1,336,897\n" +
- " 021c: 2,669,285\n" +
- " 111d: 1,112,144\n" +
- " 111u: 1,097,452\n" +
- " 030t: 132,048\n" +
- " 030c: 44,127\n" +
- " 201: 290,552\n" +
- " 120d: 47,734\n" +
- " 120u: 47,780\n" +
- " 120c: 95,855\n" +
- " 210: 90,618\n" +
- " 300: 21,656\n";
-
- expectedOutput(parameters(12, "directed", "hash"), expected);
- }
-
- @Test
- public void testHashWithLargeUndirectedRMatGraph() throws Exception {
- // computation is too large for collection mode
- Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
-
- // skip 'byte' which cannot store vertex IDs for scale > 8
- Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
-
- String expected = "\n" +
- new Checksum(2878908, 0x0015f51b1336df61L) + "\n\n" +
- "Triadic census:\n" +
- " 03: 6,101,196,568\n" +
- " 12: 145,166,335\n" +
- " 21: 7,836,753\n" +
- " 30: 479,818\n";
-
- expectedOutput(parameters(12, "undirected", "hash"), expected);
- }
-
- @Test
- public void testPrintWithSmallDirectedRMatGraph() throws Exception {
+ public void testPrintWithDirectedRMatGraph() throws Exception {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- expectedOutputChecksum(parameters(7, "directed", "print"), new Checksum(22932, 0x00002ce58634cf58L));
+ expectedOutputChecksum(parameters(8, "directed", "print"), new Checksum(61410, 0x000077d967722c8aL));
}
@Test
- public void testPrintWithSmallUndirectedRMatGraph() throws Exception {
+ public void testPrintWithUndirectedRMatGraph() throws Exception {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- expectedOutputChecksum(parameters(7, "undirected", "print"), new Checksum(22932, 0x00002d208304e7c4L));
+ expectedOutputChecksum(parameters(8, "undirected", "print"), new Checksum(61410, 0x0000780ffcb6838eL));
}
}
[3/5] flink git commit: [FLINK-6648] [gelly] Transforms for Gelly
examples
Posted by gr...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/TriangleListingBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/TriangleListingBase.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/TriangleListingBase.java
index 9e26dfe..180c46f 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/TriangleListingBase.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/TriangleListingBase.java
@@ -35,9 +35,25 @@ public abstract class TriangleListingBase<K, VV, EV, R>
extends GraphAlgorithmWrappingDataSet<K, VV, EV, R> {
// Optional configuration
+ protected boolean permuteResults;
+
protected OptionalBoolean sortTriangleVertices = new OptionalBoolean(false, true);
/**
+ * By default only one result is output for each triangle, whether vertices
+ * are sorted or unsorted. When permutation is enabled a result is instead
+ * output for each of the six permutations of the three vertex IDs.
+ *
+ * @param permuteResults whether output results should be permuted
+ * @return this
+ */
+ public TriangleListingBase<K, VV, EV, R> setPermuteResults(boolean permuteResults) {
+ this.permuteResults = permuteResults;
+
+ return this;
+ }
+
+ /**
* Normalize the triangle listing such that for each result (K0, K1, K2)
* the vertex IDs are sorted K0 < K1 < K2.
*
@@ -51,6 +67,17 @@ extends GraphAlgorithmWrappingDataSet<K, VV, EV, R> {
}
@Override
+ protected boolean canMergeConfigurationWith(GraphAlgorithmWrappingBase other) {
+ if (!super.canMergeConfigurationWith(other)) {
+ return false;
+ }
+
+ TriangleListingBase rhs = (TriangleListingBase) other;
+
+ return permuteResults == rhs.permuteResults;
+ }
+
+ @Override
protected final void mergeConfiguration(GraphAlgorithmWrappingBase other) {
super.mergeConfiguration(other);
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/TriangleListing.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/TriangleListing.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/TriangleListing.java
index beddd24..00b2210 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/TriangleListing.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/TriangleListing.java
@@ -18,6 +18,7 @@
package org.apache.flink.graph.library.clustering.directed;
+import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
@@ -121,7 +122,11 @@ extends TriangleListingBase<K, VV, EV, Result<K>> {
.with(new ProjectTriangles<K>())
.name("Triangle listing");
- if (sortTriangleVertices.get()) {
+ if (permuteResults) {
+ triangles = triangles
+ .flatMap(new PermuteResult<K>())
+ .name("Permute triangle vertices");
+ } else if (sortTriangleVertices.get()) {
triangles = triangles
.map(new SortTriangleVertices<K>())
.name("Sort triangle vertices");
@@ -309,6 +314,98 @@ extends TriangleListingBase<K, VV, EV, Result<K>> {
}
/**
+ * Output each input and an additional result for each of the five
+ * permutations of the three vertex IDs.
+ *
+ * @param <T> ID type
+ */
+ private static class PermuteResult<T>
+ implements FlatMapFunction<Result<T>, Result<T>> {
+ @Override
+ public void flatMap(Result<T> value, Collector<Result<T>> out)
+ throws Exception {
+ T tmp;
+
+ int f0f1, f0f2, f1f2;
+
+ byte bitmask = value.getBitmask().getValue();
+
+ // 0, 1, 2
+ out.collect(value);
+
+ tmp = value.getVertexId0();
+ value.setVertexId0(value.getVertexId1());
+ value.setVertexId1(tmp);
+
+ f0f1 = ((bitmask & 0b100000) >>> 1) | ((bitmask & 0b010000) << 1);
+ f0f2 = (bitmask & 0b001100) >>> 2;
+ f1f2 = (bitmask & 0b000011) << 2;
+
+ bitmask = (byte) (f0f1 | f0f2 | f1f2);
+ value.setBitmask(bitmask);
+
+ // 1, 0, 2
+ out.collect(value);
+
+ tmp = value.getVertexId1();
+ value.setVertexId1(value.getVertexId2());
+ value.setVertexId2(tmp);
+
+ f0f1 = (bitmask & 0b110000) >>> 2;
+ f0f2 = (bitmask & 0b001100) << 2;
+ f1f2 = ((bitmask & 0b000010) >>> 1) | ((bitmask & 0b000001) << 1);
+
+ bitmask = (byte) (f0f1 | f0f2 | f1f2);
+ value.setBitmask(bitmask);
+
+ // 1, 2, 0
+ out.collect(value);
+
+ tmp = value.getVertexId0();
+ value.setVertexId0(value.getVertexId2());
+ value.setVertexId2(tmp);
+
+ f0f1 = ((bitmask & 0b100000) >>> 5) | ((bitmask & 0b010000) >>> 3);
+ f0f2 = ((bitmask & 0b001000) >>> 1) | ((bitmask & 0b000100) << 1);
+ f1f2 = ((bitmask & 0b000010) << 3) | ((bitmask & 0b000001) << 5);
+
+ bitmask = (byte) (f0f1 | f0f2 | f1f2);
+ value.setBitmask(bitmask);
+
+ // 0, 2, 1
+ out.collect(value);
+
+ tmp = value.getVertexId0();
+ value.setVertexId0(value.getVertexId1());
+ value.setVertexId1(tmp);
+
+ f0f1 = ((bitmask & 0b100000) >>> 1) | ((bitmask & 0b010000) << 1);
+ f0f2 = (bitmask & 0b001100) >>> 2;
+ f1f2 = (bitmask & 0b000011) << 2;
+
+ bitmask = (byte) (f0f1 | f0f2 | f1f2);
+ value.setBitmask(bitmask);
+
+ // 2, 0, 1
+ out.collect(value);
+
+ tmp = value.getVertexId1();
+ value.setVertexId1(value.getVertexId2());
+ value.setVertexId2(tmp);
+
+ f0f1 = (bitmask & 0b110000) >>> 2;
+ f0f2 = (bitmask & 0b001100) << 2;
+ f1f2 = ((bitmask & 0b000010) >>> 1) | ((bitmask & 0b000001) << 1);
+
+ bitmask = (byte) (f0f1 | f0f2 | f1f2);
+ value.setBitmask(bitmask);
+
+ // 2, 1, 0
+ out.collect(value);
+ }
+ }
+
+ /**
* Reorders the vertices of each emitted triangle (K0, K1, K2, bitmask)
* into sorted order such that K0 < K1 < K2.
*
@@ -384,7 +481,7 @@ extends TriangleListingBase<K, VV, EV, Result<K>> {
this.bitmask = bitmask;
}
- private void setBitmask(byte bitmask) {
+ public void setBitmask(byte bitmask) {
this.bitmask.setValue(bitmask);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/TriangleListing.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/TriangleListing.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/TriangleListing.java
index c714bed..2472744 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/TriangleListing.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/TriangleListing.java
@@ -113,7 +113,11 @@ extends TriangleListingBase<K, VV, EV, Result<K>> {
.with(new ProjectTriangles<K>())
.name("Triangle listing");
- if (sortTriangleVertices.get()) {
+ if (permuteResults) {
+ triangles = triangles
+ .flatMap(new PermuteResult<K>())
+ .name("Permute triangle vertices");
+ } else if (sortTriangleVertices.get()) {
triangles = triangles
.map(new SortTriangleVertices<K>())
.name("Sort triangle vertices");
@@ -250,6 +254,59 @@ extends TriangleListingBase<K, VV, EV, Result<K>> {
}
/**
+ * Output each input and an additional result for each of the five
+ * permutations of the three vertex IDs.
+ *
+ * @param <T> ID type
+ */
+ private static class PermuteResult<T>
+ implements FlatMapFunction<Result<T>, Result<T>> {
+ @Override
+ public void flatMap(Result<T> value, Collector<Result<T>> out)
+ throws Exception {
+ T tmp;
+
+ // 0, 1, 2
+ out.collect(value);
+
+ tmp = value.getVertexId0();
+ value.setVertexId0(value.getVertexId1());
+ value.setVertexId1(tmp);
+
+ // 1, 0, 2
+ out.collect(value);
+
+ tmp = value.getVertexId1();
+ value.setVertexId1(value.getVertexId2());
+ value.setVertexId2(tmp);
+
+ // 1, 2, 0
+ out.collect(value);
+
+ tmp = value.getVertexId0();
+ value.setVertexId0(value.getVertexId2());
+ value.setVertexId2(tmp);
+
+ // 0, 2, 1
+ out.collect(value);
+
+ tmp = value.getVertexId0();
+ value.setVertexId0(value.getVertexId1());
+ value.setVertexId1(tmp);
+
+ // 2, 0, 1
+ out.collect(value);
+
+ tmp = value.getVertexId1();
+ value.setVertexId1(value.getVertexId2());
+ value.setVertexId2(tmp);
+
+ // 2, 1, 0
+ out.collect(value);
+ }
+ }
+
+ /**
* Reorders the vertices of each emitted triangle (K0, K1, K2)
* into sorted order such that K0 < K1 < K2.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/similarity/AdamicAdar.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/similarity/AdamicAdar.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/similarity/AdamicAdar.java
index 78dff0b..d761f60 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/similarity/AdamicAdar.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/similarity/AdamicAdar.java
@@ -34,6 +34,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
import org.apache.flink.graph.asm.degree.annotate.undirected.VertexDegree;
+import org.apache.flink.graph.asm.result.BinaryResult.MirrorResult;
import org.apache.flink.graph.asm.result.BinaryResultBase;
import org.apache.flink.graph.asm.result.PrintableResult;
import org.apache.flink.graph.library.similarity.AdamicAdar.Result;
@@ -82,6 +83,8 @@ extends GraphAlgorithmWrappingDataSet<K, VV, EV, Result<K>> {
private float minimumRatio = 0.0f;
+ private boolean mirrorResults;
+
/**
* Filter out Adamic-Adar scores less than the given minimum.
*
@@ -110,6 +113,20 @@ extends GraphAlgorithmWrappingDataSet<K, VV, EV, Result<K>> {
return this;
}
+ /**
+ * By default only one result is output for each pair of vertices. When
+ * mirroring a second result with the vertex order flipped is output for
+ * each pair of vertices.
+ *
+ * @param mirrorResults whether output results should be mirrored
+ * @return this
+ */
+ public AdamicAdar<K, VV, EV> setMirrorResults(boolean mirrorResults) {
+ this.mirrorResults = mirrorResults;
+
+ return this;
+ }
+
@Override
protected boolean canMergeConfigurationWith(GraphAlgorithmWrappingBase other) {
if (!super.canMergeConfigurationWith(other)) {
@@ -194,7 +211,13 @@ extends GraphAlgorithmWrappingDataSet<K, VV, EV, Result<K>> {
.withBroadcastSet(sumOfScoresAndNumberOfNeighborPairs, SUM_OF_SCORES_AND_NUMBER_OF_NEIGHBOR_PAIRS);
}
- return scores;
+ if (mirrorResults) {
+ return scores
+ .flatMap(new MirrorResult<K, Result<K>>())
+ .name("Mirror results");
+ } else {
+ return scores;
+ }
}
/**
@@ -411,7 +434,7 @@ extends GraphAlgorithmWrappingDataSet<K, VV, EV, Result<K>> {
*/
public static class Result<T>
extends BinaryResultBase<T>
- implements PrintableResult {
+ implements PrintableResult, Comparable<Result<T>> {
private FloatValue adamicAdarScore = new FloatValue();
/**
@@ -453,6 +476,11 @@ extends GraphAlgorithmWrappingDataSet<K, VV, EV, Result<K>> {
+ "), adamic-adar score: " + adamicAdarScore;
}
+ @Override
+ public int compareTo(Result<T> o) {
+ return Float.compare(adamicAdarScore.getValue(), o.adamicAdarScore.getValue());
+ }
+
// ----------------------------------------------------------------------------------------
public static final int HASH_SEED = 0xe405f6d1;
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/directed/TriangleListingTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/directed/TriangleListingTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/directed/TriangleListingTest.java
index b002956..e8f8659 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/directed/TriangleListingTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/directed/TriangleListingTest.java
@@ -31,6 +31,7 @@ import org.apache.flink.types.NullValue;
import org.apache.commons.math3.util.CombinatoricsUtils;
import org.junit.Test;
+import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -42,7 +43,7 @@ public class TriangleListingTest
extends AsmTestBase {
@Test
- public void testSimpleGraph()
+ public void testSimpleGraphSorted()
throws Exception {
DataSet<Result<IntValue>> tl = directedSimpleGraph
.run(new TriangleListing<IntValue, NullValue, NullValue>()
@@ -56,6 +57,38 @@ extends AsmTestBase {
}
@Test
+ public void testSimpleGraphPermuted()
+ throws Exception {
+ DataSet<Result<IntValue>> tl = directedSimpleGraph
+ .run(new TriangleListing<IntValue, NullValue, NullValue>()
+ .setPermuteResults(true));
+
+ String expectedResult =
+ // permutation of (0,1,2,41)
+ "1st vertex ID: 0, 2nd vertex ID: 1, 3rd vertex ID: 2, edge directions: 0->1, 0->2, 1<-2\n" +
+ "1st vertex ID: 0, 2nd vertex ID: 2, 3rd vertex ID: 1, edge directions: 0->2, 0->1, 2->1\n" +
+ "1st vertex ID: 1, 2nd vertex ID: 0, 3rd vertex ID: 2, edge directions: 1<-0, 1<-2, 0->2\n" +
+ "1st vertex ID: 1, 2nd vertex ID: 2, 3rd vertex ID: 0, edge directions: 1<-2, 1<-0, 2<-0\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 0, 3rd vertex ID: 1, edge directions: 2<-0, 2->1, 0->1\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 1, 3rd vertex ID: 0, edge directions: 2->1, 2<-0, 1<-0\n" +
+ // permutation of (1,2,3,22)
+ "1st vertex ID: 1, 2nd vertex ID: 2, 3rd vertex ID: 3, edge directions: 1<-2, 1<-3, 2->3\n" +
+ "1st vertex ID: 1, 2nd vertex ID: 3, 3rd vertex ID: 2, edge directions: 1<-3, 1<-2, 3<-2\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 1, 3rd vertex ID: 3, edge directions: 2->1, 2->3, 1<-3\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 3, 3rd vertex ID: 1, edge directions: 2->3, 2->1, 3->1\n" +
+ "1st vertex ID: 3, 2nd vertex ID: 1, 3rd vertex ID: 2, edge directions: 3->1, 3<-2, 1<-2\n" +
+ "1st vertex ID: 3, 2nd vertex ID: 2, 3rd vertex ID: 1, edge directions: 3<-2, 3->1, 2->1";
+
+ List<String> printableStrings = new ArrayList<>();
+
+ for (Result<IntValue> result : tl.collect()) {
+ printableStrings.add(result.toPrintableString());
+ }
+
+ TestBaseUtils.compareResultAsText(printableStrings, expectedResult);
+ }
+
+ @Test
public void testCompleteGraph()
throws Exception {
long expectedDegree = completeGraphVertexCount - 1;
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/undirected/TriangleListingTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/undirected/TriangleListingTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/undirected/TriangleListingTest.java
index ab99faa..6af1b01 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/undirected/TriangleListingTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/clustering/undirected/TriangleListingTest.java
@@ -31,6 +31,9 @@ import org.apache.flink.types.NullValue;
import org.apache.commons.math3.util.CombinatoricsUtils;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.List;
+
import static org.junit.Assert.assertEquals;
/**
@@ -40,7 +43,7 @@ public class TriangleListingTest
extends AsmTestBase {
@Test
- public void testSimpleGraph()
+ public void testSimpleGraphSorted()
throws Exception {
DataSet<Result<IntValue>> tl = undirectedSimpleGraph
.run(new TriangleListing<IntValue, NullValue, NullValue>()
@@ -54,6 +57,38 @@ extends AsmTestBase {
}
@Test
+ public void testSimpleGraphPermuted()
+ throws Exception {
+ DataSet<Result<IntValue>> tl = undirectedSimpleGraph
+ .run(new TriangleListing<IntValue, NullValue, NullValue>()
+ .setPermuteResults(true));
+
+ String expectedResult =
+ // permutation of (0,1,2)
+ "1st vertex ID: 0, 2nd vertex ID: 1, 3rd vertex ID: 2\n" +
+ "1st vertex ID: 0, 2nd vertex ID: 2, 3rd vertex ID: 1\n" +
+ "1st vertex ID: 1, 2nd vertex ID: 0, 3rd vertex ID: 2\n" +
+ "1st vertex ID: 1, 2nd vertex ID: 2, 3rd vertex ID: 0\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 0, 3rd vertex ID: 1\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 1, 3rd vertex ID: 0\n" +
+ // permutation of (1,2,3)
+ "1st vertex ID: 1, 2nd vertex ID: 2, 3rd vertex ID: 3\n" +
+ "1st vertex ID: 1, 2nd vertex ID: 3, 3rd vertex ID: 2\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 1, 3rd vertex ID: 3\n" +
+ "1st vertex ID: 2, 2nd vertex ID: 3, 3rd vertex ID: 1\n" +
+ "1st vertex ID: 3, 2nd vertex ID: 1, 3rd vertex ID: 2\n" +
+ "1st vertex ID: 3, 2nd vertex ID: 2, 3rd vertex ID: 1";
+
+ List<String> printableStrings = new ArrayList<>();
+
+ for (Result<IntValue> result : tl.collect()) {
+ printableStrings.add(result.toPrintableString());
+ }
+
+ TestBaseUtils.compareResultAsText(printableStrings, expectedResult);
+ }
+
+ @Test
public void testCompleteGraph()
throws Exception {
long expectedDegree = completeGraphVertexCount - 1;
[4/5] flink git commit: [FLINK-6648] [gelly] Transforms for Gelly
examples
Posted by gr...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transform.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transform.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transform.java
new file mode 100644
index 0000000..7d65cbb
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transform.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.drivers.transform;
+
+import org.apache.flink.graph.drivers.parameter.Parameterized;
+
+/**
+ * A transform is executed both before and after algorithm execution. Multiple
+ * transforms may be applied and the order of execution is as listed by the
+ * input then algorithm components. The input is transformed in given order and
+ * the algorithm result is transformed in reverse order.
+ *
+ * @param <II> input transformation input type
+ * @param <IO> input transformation output type
+ * @param <RI> result transformation input type
+ * @param <RO> input transformation output type
+ */
+public interface Transform<II, IO, RI, RO>
+extends Parameterized {
+
+ /**
+ * A human-readable identifier summarizing the transform and configuration.
+ *
+ * @return the transform identifier
+ */
+ String getIdentity();
+
+ /**
+ * Apply the forward transformation to the input graph.
+ *
+ * @param input transformation input
+ * @return transformation output
+ * @throws Exception
+ */
+ IO transformInput(II input) throws Exception;
+
+ /**
+ * Apply the reverse transformation to the algorithm result.
+ *
+ * @param result transformation input
+ * @return transformation output
+ * @throws Exception
+ */
+ RO transformResult(RI result) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transformable.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transformable.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transformable.java
new file mode 100644
index 0000000..d60f043
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/Transformable.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.flink.graph.drivers.transform;
+
+import java.util.List;
+
+/**
+ * A transformable input or algorithm returns a list of {@link Transform} to be
+ * applied before and after execution of the algorithm.
+ */
+public interface Transformable {
+
+ /**
+ * Get the list of transforms.
+ *
+ * @return the list of transforms
+ */
+ List<Transform> getTransformers();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
index c36dcb5..b048f08 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/AdamicAdarITCase.java
@@ -39,7 +39,7 @@ extends CopyableValueDriverBaseITCase {
private String[] parameters(int scale, String output, String... additionalParameters) {
String[] parameters = new String[] {
- "--algorithm", "AdamicAdar",
+ "--algorithm", "AdamicAdar", "--mirror_results",
"--input", "RMatGraph", "--scale", Integer.toString(scale), "--type", idType, "--simplify", "undirected",
"--output", output};
@@ -63,6 +63,6 @@ extends CopyableValueDriverBaseITCase {
expectedCount(
parameters(7, "print"),
- 5694);
+ 11388);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
index ce8cb3b..6bcab6b 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ClusteringCoefficientITCase.java
@@ -56,29 +56,8 @@ extends CopyableValueDriverBaseITCase {
@Test
public void testHashWithSmallDirectedRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "short":
- case "char":
- case "integer":
- checksum = 0x0000003621c62ca1L;
- break;
-
- case "long":
- checksum = 0x0000003b74c6719bL;
- break;
-
- case "string":
- checksum = 0x0000003ab67abea8L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
String expected = "\n" +
- new Checksum(117, checksum) + "\n" +
+ new Checksum(117, 0x0000003621c62ca1L) + "\n\n" +
"triplet count: 29286, triangle count: 11466, global clustering coefficient: 0.39151813[0-9]+\n" +
"vertex count: 117, average clustering coefficient: 0.45125697[0-9]+\n";
@@ -87,39 +66,14 @@ extends CopyableValueDriverBaseITCase {
@Test
public void testHashWithSmallUndirectedRMatGraph() throws Exception {
- long directedChecksum;
- long undirectedChecksum;
- switch (idType) {
- case "byte":
- case "short":
- case "char":
- case "integer":
- directedChecksum = 0x0000003875b38c43L;
- undirectedChecksum = 0x0000003c20344c75L;
- break;
-
- case "long":
- directedChecksum = 0x0000003671970c59L;
- undirectedChecksum = 0x0000003939645d8cL;
- break;
-
- case "string":
- directedChecksum = 0x0000003be109a770L;
- undirectedChecksum = 0x0000003b8c98d14aL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- String expected = "\n" +
+ String expected = "\n\n" +
"triplet count: 29286, triangle count: 11466, global clustering coefficient: 0.39151813[0-9]+\n" +
"vertex count: 117, average clustering coefficient: 0.57438679[0-9]+\n";
expectedOutput(parameters(7, "directed", "undirected", "hash"),
- "\n" + new Checksum(117, directedChecksum) + expected);
+ "\n" + new Checksum(117, 0x0000003875b38c43L) + expected);
expectedOutput(parameters(7, "undirected", "undirected", "hash"),
- "\n" + new Checksum(117, undirectedChecksum) + expected);
+ "\n" + new Checksum(117, 0x0000003c20344c75L) + expected);
}
@Test
@@ -127,31 +81,11 @@ extends CopyableValueDriverBaseITCase {
// computation is too large for collection mode
Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
- long checksum;
- switch (idType) {
- case "byte":
- return;
-
- case "short":
- case "char":
- case "integer":
- checksum = 0x0000067a9d18e7f3L;
- break;
-
- case "long":
- checksum = 0x00000694a90ee6d4L;
- break;
-
- case "string":
- checksum = 0x000006893e3b314fL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
+ // skip 'byte' which cannot store vertex IDs for scale > 8
+ Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
String expected = "\n" +
- new Checksum(3349, checksum) + "\n" +
+ new Checksum(3349, 0x0000067a9d18e7f3L) + "\n\n" +
"triplet count: 9276207, triangle count: 1439454, global clustering coefficient: 0.15517700[0-9]+\n" +
"vertex count: 3349, average clustering coefficient: 0.24571815[0-9]+\n";
@@ -163,40 +97,16 @@ extends CopyableValueDriverBaseITCase {
// computation is too large for collection mode
Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
- long directedChecksum;
- long undirectedChecksum;
- switch (idType) {
- case "byte":
- return;
-
- case "short":
- case "char":
- case "integer":
- directedChecksum = 0x00000681fad1587eL;
- undirectedChecksum = 0x0000068713b3b7f1L;
- break;
-
- case "long":
- directedChecksum = 0x000006928a6301b1L;
- undirectedChecksum = 0x000006a399edf0e6L;
- break;
-
- case "string":
- directedChecksum = 0x000006749670a2f7L;
- undirectedChecksum = 0x0000067f19c6c4d5L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
+ // skip 'byte' which cannot store vertex IDs for scale > 8
+ Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
- String expected = "\n" +
+ String expected = "\n\n" +
"triplet count: 9276207, triangle count: 1439454, global clustering coefficient: 0.15517700[0-9]+\n" +
"vertex count: 3349, average clustering coefficient: 0.33029442[0-9]+\n";
expectedOutput(parameters(12, "directed", "undirected", "hash"),
- "\n" + new Checksum(3349, directedChecksum) + expected);
+ "\n" + new Checksum(3349, 0x00000681fad1587eL) + expected);
expectedOutput(parameters(12, "undirected", "undirected", "hash"),
- "\n" + new Checksum(3349, undirectedChecksum) + expected);
+ "\n" + new Checksum(3349, 0x0000068713b3b7f1L) + expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
index cbb5c9d..d6913c4 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/ConnectedComponentsITCase.java
@@ -31,7 +31,7 @@ import org.junit.runners.Parameterized;
*/
@RunWith(Parameterized.class)
public class ConnectedComponentsITCase
-extends DriverBaseITCase {
+extends NonTransformableDriverBaseITCase {
public ConnectedComponentsITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -57,34 +57,7 @@ extends DriverBaseITCase {
@Test
public void testHashWithSmallRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x0000000000033e88L;
- break;
-
- case "long":
- checksum = 0x0000000000057848L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x000000000254a4c3L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- expectedChecksum(parameters(7, "hash"), 106, checksum);
+ expectedChecksum(parameters(7, "hash"), 106, 0x0000000000033e88L);
}
@Test
@@ -92,36 +65,10 @@ extends DriverBaseITCase {
// computation is too large for collection mode
Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- return;
-
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x00000003094ffba2L;
- break;
-
- case "long":
- checksum = 0x000000030b68e522L;
- break;
+ // skip 'byte' which cannot store vertex IDs for scale > 8
+ Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
- case "string":
- case "nativeString":
- checksum = 0x00001839ad14edb1L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- expectedChecksum(parameters(15, "hash"), 25572, checksum);
+ expectedChecksum(parameters(15, "hash"), 25572, 0x00000003094ffba2L);
}
@Test
@@ -129,28 +76,6 @@ extends DriverBaseITCase {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "integer":
- case "nativeInteger":
- case "long":
- case "nativeLong":
- checksum = 0x00000024edd0568dL;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x000000232d8bf58dL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- expectedOutputChecksum(parameters(7, "print"), new Checksum(106, checksum));
+ expectedOutputChecksum(parameters(7, "print"), new Checksum(106, 0x00000024edd0568dL));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/CopyableValueDriverBaseITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/CopyableValueDriverBaseITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/CopyableValueDriverBaseITCase.java
index bc09820..b01fd88 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/CopyableValueDriverBaseITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/CopyableValueDriverBaseITCase.java
@@ -41,7 +41,7 @@ extends DriverBaseITCase {
public static Collection<Object[]> executionModes() {
List<Object[]> executionModes = new ArrayList<>();
- for (String idType : new String[] {"byte", "short", "char", "integer", "long", "string"}) {
+ for (String idType : new String[] {"byte", "short", "char", "integer", "long", "float", "double", "string"}) {
for (TestExecutionMode executionMode : TestExecutionMode.values()) {
executionModes.add(new Object[] {idType, executionMode});
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/DriverBaseITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/DriverBaseITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/DriverBaseITCase.java
index 670968c..749998d 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/DriverBaseITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/DriverBaseITCase.java
@@ -61,7 +61,8 @@ extends MultipleProgramsTestBase {
List<Object[]> executionModes = new ArrayList<>();
for (String idType : new String[] {"byte", "nativeByte", "short", "nativeShort", "char", "nativeChar",
- "integer", "nativeInteger", "long", "nativeLong", "string", "nativeString"}) {
+ "integer", "nativeInteger", "long", "nativeLong", "float", "nativeFloat", "double", "nativeDouble",
+ "string", "nativeString"}) {
for (TestExecutionMode executionMode : TestExecutionMode.values()) {
executionModes.add(new Object[] {idType, executionMode});
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
index a56a19b..2f8eb0d 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/EdgeListITCase.java
@@ -32,7 +32,7 @@ import org.junit.runners.Parameterized;
*/
@RunWith(Parameterized.class)
public class EdgeListITCase
-extends DriverBaseITCase {
+extends NonTransformableDriverBaseITCase {
public EdgeListITCase(String idType, TestExecutionMode mode) {
super(idType, mode);
@@ -48,37 +48,20 @@ extends DriverBaseITCase {
}
@Test
- public void testHashWithCirculantGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x000000000001ae80L;
- break;
-
- case "long":
- checksum = 0x0000000000053580L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x0000000000656880L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
+ public void testLongDescription() throws Exception {
+ String expected = regexSubstring(new EdgeList().getLongDescription());
+
+ expectedOutputFromException(
+ new String[]{"--algorithm", "EdgeList"},
+ expected,
+ ProgramParametrizationException.class);
+ }
+ @Test
+ public void testHashWithCirculantGraph() throws Exception {
expectedChecksum(
parameters("CirculantGraph", "hash", "--vertex_count", "42", "--range0", "13:4"),
- 168, checksum);
+ 168, 0x000000000001ae80);
}
@Test
@@ -92,47 +75,10 @@ extends DriverBaseITCase {
}
@Test
- public void testLongDescription() throws Exception {
- String expected = regexSubstring(new EdgeList().getLongDescription());
-
- expectedOutputFromException(
- new String[]{"--algorithm", "EdgeList"},
- expected,
- ProgramParametrizationException.class);
- }
-
- @Test
public void testHashWithCompleteGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x0000000000113ca0L;
- break;
-
- case "long":
- checksum = 0x0000000000356460L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x00000000040f6f20L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("CompleteGraph", "hash", "--vertex_count", "42"),
- 1722, checksum);
+ 1722, 0x0000000000113ca0L);
}
@Test
@@ -147,36 +93,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithCycleGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x000000000000d740L;
- break;
-
- case "long":
- checksum = 0x0000000000029ac0L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x000000000032b440L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("CycleGraph", "hash", "--vertex_count", "42"),
- 84, checksum);
+ 84, 0x000000000000d740L);
}
@Test
@@ -191,36 +110,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithEchoGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x0000000000057720L;
- break;
-
- case "long":
- checksum = 0x000000000010ede0L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x00000000014993a0L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("EchoGraph", "hash", "--vertex_count", "42", "--vertex_degree", "13"),
- 546, checksum);
+ 546, 0x0000000000057720L);
}
@Test
@@ -242,36 +134,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithGridGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x000000000000eba0L;
- break;
-
- case "long":
- checksum = 0x000000000003a660L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x0000000000430ee0L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("GridGraph", "hash", "--dim0", "2:true", "--dim1", "3:false", "--dim2", "5:true"),
- 130, checksum);
+ 130, 0x000000000000eba0L);
}
@Test
@@ -286,36 +151,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithHypercubeGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x00000000001bc800L;
- break;
-
- case "long":
- checksum = 0x00000000002e9800L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x00000000143c1500L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("HypercubeGraph", "hash", "--dimensions", "7"),
- 896, checksum);
+ 896, 0x00000000001bc800L);
}
@Test
@@ -330,36 +168,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithPathGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x000000000000d220L;
- break;
-
- case "long":
- checksum = 0x0000000000028ae0L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x000000000031dea0L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("PathGraph", "hash", "--vertex_count", "42"),
- 82, checksum);
+ 82, 0x000000000000d220L);
}
@Test
@@ -374,36 +185,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x00000000001ee529L;
- break;
-
- case "long":
- checksum = 0x000000000049e529L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x000000000b8c9aa3L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("RMatGraph", "hash", "--scale", "7"),
- 2048, checksum);
+ 2048, 0x00000000001ee529);
}
@Test
@@ -418,36 +202,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithDirectedRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x00000000001579bdL;
- break;
-
- case "long":
- checksum = 0x00000000002dffbdL;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x0000000009213f1fL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("RMatGraph", "hash", "--scale", "7", "--simplify", "directed"),
- 1168, checksum);
+ 1168, 0x00000000001579bdL);
}
@Test
@@ -462,36 +219,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithUndirectedRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x0000000000242920L;
- break;
-
- case "long":
- checksum = 0x00000000004b1660L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x000000000fcbc080L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("RMatGraph", "hash", "--scale", "7", "--simplify", "undirected"),
- 1854, checksum);
+ 1854, 0x0000000000242920L);
}
@Test
@@ -506,36 +236,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithSingletonEdgeGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x000000000001b3c0L;
- break;
-
- case "long":
- checksum = 0x0000000000037740L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x00000000003ca2c0L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("SingletonEdgeGraph", "hash", "--vertex_pair_count", "42"),
- 84, checksum);
+ 84, 0x000000000001b3c0L);
}
@Test
@@ -550,36 +253,9 @@ extends DriverBaseITCase {
@Test
public void testHashWithStarGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "nativeByte":
- case "short":
- case "nativeShort":
- case "char":
- case "nativeChar":
- case "integer":
- case "nativeInteger":
- case "nativeLong":
- checksum = 0x0000000000006ba0L;
- break;
-
- case "long":
- checksum = 0x0000000000022460L;
- break;
-
- case "string":
- case "nativeString":
- checksum = 0x00000000001a4a20L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
expectedChecksum(
parameters("StarGraph", "hash", "--vertex_count", "42"),
- 82, checksum);
+ 82, 0x0000000000006ba0L);
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
index a8b0111..25eec22 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/JaccardIndexITCase.java
@@ -59,28 +59,7 @@ extends CopyableValueDriverBaseITCase {
@Test
public void testHashWithSmallRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "short":
- case "char":
- case "integer":
- checksum = 0x0000164757052eebL;
- break;
-
- case "long":
- checksum = 0x000016337a6a7270L;
- break;
-
- case "string":
- checksum = 0x00001622a522a290L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- expectedChecksum(parameters(7, "hash"), 11388, checksum);
+ expectedChecksum(parameters(7, "hash"), 11388, 0x0000164757052eebL);
}
@Test
@@ -88,30 +67,10 @@ extends CopyableValueDriverBaseITCase {
// computation is too large for collection mode
Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
- long checksum;
- switch (idType) {
- case "byte":
- return;
-
- case "short":
- case "char":
- case "integer":
- checksum = 0x0021ce158d811c4eL;
- break;
-
- case "long":
- checksum = 0x0021d20fb3904720L;
- break;
-
- case "string":
- checksum = 0x0021cd8fafec1524L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
+ // skip 'byte' which cannot store vertex IDs for scale > 8
+ Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
- expectedChecksum(parameters(12, "hash"), 4432058, checksum);
+ expectedChecksum(parameters(12, "hash"), 4432058, 0x0021ce158d811c4eL);
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/NonTransformableDriverBaseITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/NonTransformableDriverBaseITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/NonTransformableDriverBaseITCase.java
new file mode 100644
index 0000000..2238a35
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/NonTransformableDriverBaseITCase.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.flink.graph.drivers;
+
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Base class for drivers requiring the key ID to hash to the same value
+ * without transformation of the algorithm result to a common output type. This
+ * class overrides {@link DriverBaseITCase} to restrict the tested ID types.
+ */
+public abstract class NonTransformableDriverBaseITCase
+extends DriverBaseITCase {
+
+ protected NonTransformableDriverBaseITCase(String idType, TestExecutionMode mode) {
+ super(idType, mode);
+ }
+
+ // limit tests to types using a proper and consistent hashCode
+ @Parameterized.Parameters(name = "ID type = {0}, Execution mode = {1}")
+ public static Collection<Object[]> executionModes() {
+ List<Object[]> executionModes = new ArrayList<>();
+
+ for (String idType : new String[] {"byte", "nativeByte", "short", "nativeShort", "char", "nativeChar",
+ "integer", "nativeInteger", "nativeLong"}) {
+ for (TestExecutionMode executionMode : TestExecutionMode.values()) {
+ executionModes.add(new Object[] {idType, executionMode});
+ }
+ }
+
+ return executionModes;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
index c691922..687610e 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/TriangleListingITCase.java
@@ -40,7 +40,7 @@ extends CopyableValueDriverBaseITCase {
private String[] parameters(int scale, String order, String output) {
String[] parameters = new String[] {
- "--algorithm", "TriangleListing", "--order", order,
+ "--algorithm", "TriangleListing", "--order", order, "--permute_results",
"--input", "RMatGraph", "--scale", Integer.toString(scale), "--type", idType, "--simplify", order,
"--output", output};
@@ -63,29 +63,8 @@ extends CopyableValueDriverBaseITCase {
@Test
public void testHashWithSmallDirectedRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "short":
- case "char":
- case "integer":
- checksum = 0x00000784d2c336cdL;
- break;
-
- case "long":
- checksum = 0x0000078e5ebf2927L;
- break;
-
- case "string":
- checksum = 0x0000077eddf67481L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
String expected = "\n" +
- new Checksum(3822, checksum) + "\n" +
+ new Checksum(22932, 0x00002cdfb573650dL) + "\n\n" +
"Triadic census:\n" +
" 003: 178,989\n" +
" 012: 47,736\n" +
@@ -109,29 +88,8 @@ extends CopyableValueDriverBaseITCase {
@Test
public void testHashWithSmallUndirectedRMatGraph() throws Exception {
- long checksum;
- switch (idType) {
- case "byte":
- case "short":
- case "char":
- case "integer":
- checksum = 0x0000075b6b9a0ad0L;
- break;
-
- case "long":
- checksum = 0x00000761619e7f3cL;
- break;
-
- case "string":
- checksum = 0x0000079b15eb30acL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
String expected = "\n" +
- new Checksum(3822, checksum) + "\n" +
+ new Checksum(22932, 0x00002c7fb95ec78eL) + "\n\n" +
"Triadic census:\n" +
" 03: 178,989\n" +
" 12: 59,499\n" +
@@ -146,31 +104,11 @@ extends CopyableValueDriverBaseITCase {
// computation is too large for collection mode
Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
- long checksum;
- switch (idType) {
- case "byte":
- return;
-
- case "short":
- case "char":
- case "integer":
- checksum = 0x0003a986d6bedc53L;
- break;
-
- case "long":
- checksum = 0x0003a8d91c92b884L;
- break;
-
- case "string":
- checksum = 0x0003a88fffc33f27L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
+ // skip 'byte' which cannot store vertex IDs for scale > 8
+ Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
String expected = "\n" +
- new Checksum(479818, checksum) + "\n" +
+ new Checksum(2878908, 0x0015f77b8984c7caL) + "\n\n" +
"Triadic census:\n" +
" 003: 6,101,196,568\n" +
" 012: 132,051,207\n" +
@@ -197,31 +135,11 @@ extends CopyableValueDriverBaseITCase {
// computation is too large for collection mode
Assume.assumeFalse(mode == TestExecutionMode.COLLECTION);
- long checksum;
- switch (idType) {
- case "byte":
- return;
-
- case "short":
- case "char":
- case "integer":
- checksum = 0x0003a95630aae344L;
- break;
-
- case "long":
- checksum = 0x0003a9b1e055d59dL;
- break;
-
- case "string":
- checksum = 0x0003aa1e3d8f2c6bL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
+ // skip 'byte' which cannot store vertex IDs for scale > 8
+ Assume.assumeFalse(idType.equals("byte") || idType.equals("nativeByte"));
String expected = "\n" +
- new Checksum(479818, checksum) + "\n" +
+ new Checksum(2878908, 0x0015f51b1336df61L) + "\n\n" +
"Triadic census:\n" +
" 03: 6,101,196,568\n" +
" 12: 145,166,335\n" +
@@ -236,24 +154,7 @@ extends CopyableValueDriverBaseITCase {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- long checksum;
- switch (idType) {
- case "byte":
- case "short":
- case "integer":
- case "long":
- checksum = 0x00000764227995aaL;
- break;
-
- case "string":
- checksum = 0x000007643d93c30aL;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- expectedOutputChecksum(parameters(7, "directed", "print"), new Checksum(3822, checksum));
+ expectedOutputChecksum(parameters(7, "directed", "print"), new Checksum(22932, 0x00002ce58634cf58L));
}
@Test
@@ -261,23 +162,6 @@ extends CopyableValueDriverBaseITCase {
// skip 'char' since it is not printed as a number
Assume.assumeFalse(idType.equals("char") || idType.equals("nativeChar"));
- long checksum;
- switch (idType) {
- case "byte":
- case "short":
- case "integer":
- case "long":
- checksum = 0x0000077d1582e206L;
- break;
-
- case "string":
- checksum = 0x0000077a49cb5268L;
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + idType);
- }
-
- expectedOutputChecksum(parameters(7, "undirected", "print"), new Checksum(3822, checksum));
+ expectedOutputChecksum(parameters(7, "undirected", "print"), new Checksum(22932, 0x00002d208304e7c4L));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/input/GeneratedGraphTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/input/GeneratedGraphTest.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/input/GeneratedGraphTest.java
deleted file mode 100644
index 1608f95..0000000
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/input/GeneratedGraphTest.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.drivers.input;
-
-import org.apache.flink.graph.asm.translate.TranslateFunction;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToChar;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToCharValue;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToLong;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToString;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToUnsignedByte;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToUnsignedByteValue;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToUnsignedInt;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToUnsignedShort;
-import org.apache.flink.graph.drivers.input.GeneratedGraph.LongValueToUnsignedShortValue;
-import org.apache.flink.types.ByteValue;
-import org.apache.flink.types.CharValue;
-import org.apache.flink.types.LongValue;
-import org.apache.flink.types.ShortValue;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Tests for {@link GeneratedGraph}.
- */
-public class GeneratedGraphTest {
-
- private TranslateFunction<LongValue, ByteValue> byteValueTranslator = new LongValueToUnsignedByteValue();
- private TranslateFunction<LongValue, Byte> byteTranslator = new LongValueToUnsignedByte();
- private TranslateFunction<LongValue, ShortValue> shortValueTranslator = new LongValueToUnsignedShortValue();
- private TranslateFunction<LongValue, Short> shortTranslator = new LongValueToUnsignedShort();
- private TranslateFunction<LongValue, CharValue> charValueTranslator = new LongValueToCharValue();
- private TranslateFunction<LongValue, Character> charTranslator = new LongValueToChar();
- private TranslateFunction<LongValue, Integer> intTranslator = new LongValueToUnsignedInt();
- private TranslateFunction<LongValue, Long> longTranslator = new LongValueToLong();
- private TranslateFunction<LongValue, String> stringTranslator = new LongValueToString();
-
- private ByteValue byteValue = new ByteValue();
- private ShortValue shortValue = new ShortValue();
- private CharValue charValue = new CharValue();
-
- // ByteValue
-
- @Test
- public void testByteValueTranslation() throws Exception {
- assertEquals(new ByteValue((byte) 0), byteValueTranslator.translate(new LongValue(0L), byteValue));
- assertEquals(new ByteValue(Byte.MIN_VALUE), byteValueTranslator.translate(new LongValue((long) Byte.MAX_VALUE + 1), byteValue));
- assertEquals(new ByteValue((byte) -1), byteValueTranslator.translate(new LongValue((1L << 8) - 1), byteValue));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testByteValueTranslationUpperOutOfRange() throws Exception {
- byteValueTranslator.translate(new LongValue(1L << 8), byteValue);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testByteValueTranslationLowerOutOfRange() throws Exception {
- byteValueTranslator.translate(new LongValue(-1), byteValue);
- }
-
- // Byte
-
- @Test
- public void testByteTranslation() throws Exception {
- assertEquals(Byte.valueOf((byte) 0), byteTranslator.translate(new LongValue(0L), null));
- assertEquals(Byte.valueOf(Byte.MIN_VALUE), byteTranslator.translate(new LongValue((long) Byte.MAX_VALUE + 1), null));
- assertEquals(Byte.valueOf((byte) -1), byteTranslator.translate(new LongValue((1L << 8) - 1), null));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testByteTranslationUpperOutOfRange() throws Exception {
- byteTranslator.translate(new LongValue(1L << 8), null);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testByteTranslationLowerOutOfRange() throws Exception {
- byteTranslator.translate(new LongValue(-1), null);
- }
-
- // ShortValue
-
- @Test
- public void testShortValueTranslation() throws Exception {
- assertEquals(new ShortValue((short) 0), shortValueTranslator.translate(new LongValue(0L), shortValue));
- assertEquals(new ShortValue(Short.MIN_VALUE), shortValueTranslator.translate(new LongValue((long) Short.MAX_VALUE + 1), shortValue));
- assertEquals(new ShortValue((short) -1), shortValueTranslator.translate(new LongValue((1L << 16) - 1), shortValue));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testShortValueTranslationUpperOutOfRange() throws Exception {
- shortValueTranslator.translate(new LongValue(1L << 16), shortValue);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testShortValueTranslationLowerOutOfRange() throws Exception {
- shortValueTranslator.translate(new LongValue(-1), shortValue);
- }
-
- // Short
-
- @Test
- public void testShortTranslation() throws Exception {
- assertEquals(Short.valueOf((short) 0), shortTranslator.translate(new LongValue(0L), null));
- assertEquals(Short.valueOf(Short.MIN_VALUE), shortTranslator.translate(new LongValue((long) Short.MAX_VALUE + 1), null));
- assertEquals(Short.valueOf((short) -1), shortTranslator.translate(new LongValue((1L << 16) - 1), null));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testShortTranslationUpperOutOfRange() throws Exception {
- shortTranslator.translate(new LongValue(1L << 16), null);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testShortTranslationLowerOutOfRange() throws Exception {
- shortTranslator.translate(new LongValue(-1), null);
- }
-
- // CharValue
-
- @Test
- public void testCharValueTranslation() throws Exception {
- assertEquals(new CharValue((char) 0), charValueTranslator.translate(new LongValue(0L), charValue));
- assertEquals(new CharValue(Character.MAX_VALUE), charValueTranslator.translate(new LongValue((long) Character.MAX_VALUE), charValue));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testCharValueTranslationUpperOutOfRange() throws Exception {
- charValueTranslator.translate(new LongValue(1L << 16), charValue);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testCharValueTranslationLowerOutOfRange() throws Exception {
- charValueTranslator.translate(new LongValue(-1), charValue);
- }
-
- // Character
-
- @Test
- public void testCharacterTranslation() throws Exception {
- assertEquals(Character.valueOf((char) 0), charTranslator.translate(new LongValue(0L), null));
- assertEquals(Character.valueOf(Character.MAX_VALUE), charTranslator.translate(new LongValue((long) Character.MAX_VALUE), null));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testCharacterTranslationUpperOutOfRange() throws Exception {
- charTranslator.translate(new LongValue(1L << 16), null);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testCharacterTranslationLowerOutOfRange() throws Exception {
- charTranslator.translate(new LongValue(-1), null);
- }
-
- // Integer
-
- @Test
- public void testIntegerTranslation() throws Exception {
- assertEquals(Integer.valueOf(0), intTranslator.translate(new LongValue(0L), null));
- assertEquals(Integer.valueOf(Integer.MIN_VALUE), intTranslator.translate(new LongValue((long) Integer.MAX_VALUE + 1), null));
- assertEquals(Integer.valueOf(-1), intTranslator.translate(new LongValue((1L << 32) - 1), null));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testIntegerTranslationUpperOutOfRange() throws Exception {
- intTranslator.translate(new LongValue(1L << 32), null);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testIntegerTranslationLowerOutOfRange() throws Exception {
- intTranslator.translate(new LongValue(-1), null);
- }
-
- // Long
-
- @Test
- public void testLongTranslation() throws Exception {
- assertEquals(Long.valueOf(0L), longTranslator.translate(new LongValue(0L), null));
- assertEquals(Long.valueOf(Long.MIN_VALUE), longTranslator.translate(new LongValue(Long.MIN_VALUE), null));
- assertEquals(Long.valueOf(Long.MAX_VALUE), longTranslator.translate(new LongValue(Long.MAX_VALUE), null));
- }
-
- // String
-
- @Test
- public void testStringTranslation() throws Exception {
- assertEquals("0", stringTranslator.translate(new LongValue(0L), null));
- assertEquals("-9223372036854775808", stringTranslator.translate(new LongValue(Long.MIN_VALUE), null));
- assertEquals("9223372036854775807", stringTranslator.translate(new LongValue(Long.MAX_VALUE), null));
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransformTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransformTest.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransformTest.java
new file mode 100644
index 0000000..2a93911
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransformTest.java
@@ -0,0 +1,534 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.flink.graph.drivers.transform;
+
+import org.apache.flink.graph.asm.translate.TranslateFunction;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.DoubleToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.DoubleValueToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToChar;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToCharValue;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToDouble;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToDoubleValue;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToLong;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToString;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedByte;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedByteValue;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedFloat;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedFloatValue;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedInt;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedShort;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.LongValueToUnsignedShortValue;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.StringToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.StringValueToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedByteToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedByteValueToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedFloatToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedFloatValueToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedIntToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedShortToLongValueWithProperHashCode;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform.UnsignedShortValueToLongValueWithProperHashCode;
+import org.apache.flink.types.ByteValue;
+import org.apache.flink.types.CharValue;
+import org.apache.flink.types.DoubleValue;
+import org.apache.flink.types.FloatValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.ShortValue;
+import org.apache.flink.types.StringValue;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Tests for {@link GraphKeyTypeTransform}.
+ */
+public class GraphKeyTypeTransformTest {
+
+ private ByteValue byteValue = new ByteValue();
+ private ShortValue shortValue = new ShortValue();
+ private CharValue charValue = new CharValue();
+ private FloatValue floatValue = new FloatValue();
+ private DoubleValue doubleValue = new DoubleValue();
+ private LongValueWithProperHashCode longValueWithProperHashCode = new LongValueWithProperHashCode();
+
+ // ByteValue
+
+ @Test
+ public void testToByteValue() throws Exception {
+ TranslateFunction<LongValue, ByteValue> translator = new LongValueToUnsignedByteValue();
+
+ Assert.assertEquals(new ByteValue((byte) 0),
+ translator.translate(new LongValue(0L), byteValue));
+
+ Assert.assertEquals(new ByteValue(Byte.MIN_VALUE),
+ translator.translate(new LongValue(Byte.MAX_VALUE + 1), byteValue));
+
+ Assert.assertEquals(new ByteValue((byte) -1),
+ translator.translate(new LongValue(LongValueToUnsignedByteValue.MAX_VERTEX_COUNT - 1), byteValue));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToByteValueUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedByteValue().translate(new LongValue(LongValueToUnsignedByteValue.MAX_VERTEX_COUNT), byteValue);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToByteValueLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedByteValue().translate(new LongValue(-1), byteValue);
+ }
+
+ @Test
+ public void testFromByteValue() throws Exception {
+ TranslateFunction<ByteValue, LongValueWithProperHashCode> translator = new UnsignedByteValueToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(new ByteValue((byte) 0), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Byte.MAX_VALUE + 1),
+ translator.translate(new ByteValue(Byte.MIN_VALUE), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedByteValue.MAX_VERTEX_COUNT - 1),
+ translator.translate(new ByteValue((byte) -1), longValueWithProperHashCode));
+ }
+
+ // Byte
+
+ @Test
+ public void testToByte() throws Exception {
+ TranslateFunction<LongValue, Byte> translator = new LongValueToUnsignedByte();
+
+ Assert.assertEquals(Byte.valueOf((byte) 0),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Byte.valueOf(Byte.MIN_VALUE),
+ translator.translate(new LongValue((long) Byte.MAX_VALUE + 1), null));
+
+ Assert.assertEquals(Byte.valueOf((byte) -1),
+ translator.translate(new LongValue(LongValueToUnsignedByte.MAX_VERTEX_COUNT - 1), null));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToByteUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedByte().translate(new LongValue(LongValueToUnsignedByte.MAX_VERTEX_COUNT), null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToByteLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedByte().translate(new LongValue(-1), null);
+ }
+
+ @Test
+ public void testFromByte() throws Exception {
+ TranslateFunction<Byte, LongValueWithProperHashCode> translator = new UnsignedByteToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate((byte) 0, longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Byte.MAX_VALUE + 1),
+ translator.translate(Byte.MIN_VALUE, longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedByte.MAX_VERTEX_COUNT - 1),
+ translator.translate((byte) -1, longValueWithProperHashCode));
+ }
+
+ // ShortValue
+
+ @Test
+ public void testToShortValue() throws Exception {
+ TranslateFunction<LongValue, ShortValue> translator = new LongValueToUnsignedShortValue();
+
+ Assert.assertEquals(new ShortValue((short) 0),
+ translator.translate(new LongValue(0L), shortValue));
+
+ Assert.assertEquals(new ShortValue(Short.MIN_VALUE),
+ translator.translate(new LongValue((long) Short.MAX_VALUE + 1), shortValue));
+
+ Assert.assertEquals(new ShortValue((short) -1),
+ translator.translate(new LongValue(LongValueToUnsignedShortValue.MAX_VERTEX_COUNT - 1), shortValue));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToShortValueUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedShortValue().translate(new LongValue(LongValueToUnsignedShortValue.MAX_VERTEX_COUNT), shortValue);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToShortValueLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedShortValue().translate(new LongValue(-1), shortValue);
+ }
+
+ @Test
+ public void testFromShortValue() throws Exception {
+ TranslateFunction<ShortValue, LongValueWithProperHashCode> translator = new UnsignedShortValueToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(new ShortValue((short) 0), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Short.MAX_VALUE + 1),
+ translator.translate(new ShortValue(Short.MIN_VALUE), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedShortValue.MAX_VERTEX_COUNT - 1),
+ translator.translate(new ShortValue((short) -1), longValueWithProperHashCode));
+ }
+
+ // Short
+
+ @Test
+ public void testToShort() throws Exception {
+ TranslateFunction<LongValue, Short> translator = new LongValueToUnsignedShort();
+
+ Assert.assertEquals(Short.valueOf((short) 0),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Short.valueOf(Short.MIN_VALUE),
+ translator.translate(new LongValue((long) Short.MAX_VALUE + 1), null));
+
+ Assert.assertEquals(Short.valueOf((short) -1),
+ translator.translate(new LongValue(LongValueToUnsignedShort.MAX_VERTEX_COUNT - 1), null));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToShortUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedShort().translate(new LongValue(LongValueToUnsignedShort.MAX_VERTEX_COUNT), null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToShortLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedShort().translate(new LongValue(-1), null);
+ }
+
+ @Test
+ public void testFromShort() throws Exception {
+ TranslateFunction<Short, LongValueWithProperHashCode> translator = new UnsignedShortToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate((short) 0, longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Short.MAX_VALUE + 1),
+ translator.translate(Short.MIN_VALUE, longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedShort.MAX_VERTEX_COUNT - 1),
+ translator.translate((short) -1, longValueWithProperHashCode));
+ }
+
+ // CharValue
+
+ @Test
+ public void testToCharValue() throws Exception {
+ TranslateFunction<LongValue, CharValue> translator = new LongValueToCharValue();
+
+ Assert.assertEquals(new CharValue((char) 0),
+ translator.translate(new LongValue(0L), charValue));
+
+ Assert.assertEquals(new CharValue(Character.MAX_VALUE),
+ translator.translate(new LongValue(LongValueToCharValue.MAX_VERTEX_COUNT - 1), charValue));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToCharValueUpperOutOfRange() throws Exception {
+ new LongValueToCharValue().translate(new LongValue(LongValueToCharValue.MAX_VERTEX_COUNT), charValue);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToCharValueLowerOutOfRange() throws Exception {
+ new LongValueToCharValue().translate(new LongValue(-1), charValue);
+ }
+
+ // Character
+
+ @Test
+ public void testToCharacter() throws Exception {
+ TranslateFunction<LongValue, Character> translator = new LongValueToChar();
+
+ Assert.assertEquals(Character.valueOf((char) 0),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Character.valueOf(Character.MAX_VALUE),
+ translator.translate(new LongValue(LongValueToChar.MAX_VERTEX_COUNT - 1), null));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToCharacterUpperOutOfRange() throws Exception {
+ new LongValueToChar().translate(new LongValue(LongValueToChar.MAX_VERTEX_COUNT), null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToCharacterLowerOutOfRange() throws Exception {
+ new LongValueToChar().translate(new LongValue(-1), null);
+ }
+
+ // Integer
+
+ @Test
+ public void testToInt() throws Exception {
+ TranslateFunction<LongValue, Integer> translator = new LongValueToUnsignedInt();
+
+ Assert.assertEquals(Integer.valueOf(0),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Integer.valueOf(Integer.MIN_VALUE),
+ translator.translate(new LongValue((long) Integer.MAX_VALUE + 1), null));
+
+ Assert.assertEquals(Integer.valueOf(-1),
+ translator.translate(new LongValue(LongValueToUnsignedInt.MAX_VERTEX_COUNT - 1), null));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToIntUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedInt().translate(new LongValue(LongValueToUnsignedInt.MAX_VERTEX_COUNT), null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToIntLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedInt().translate(new LongValue(-1), null);
+ }
+
+ @Test
+ public void testFromInt() throws Exception {
+ TranslateFunction<Integer, LongValueWithProperHashCode> translator = new UnsignedIntToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(0, longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode((long) Integer.MAX_VALUE + 1),
+ translator.translate(Integer.MIN_VALUE, longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedInt.MAX_VERTEX_COUNT - 1),
+ translator.translate(-1, longValueWithProperHashCode));
+ }
+
+ // LongValue
+
+ @Test
+ public void testFromLongValue() throws Exception {
+ TranslateFunction<LongValue, LongValueWithProperHashCode> translator = new LongValueToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(new LongValue(0), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MIN_VALUE),
+ translator.translate(new LongValue(Long.MIN_VALUE), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MAX_VALUE),
+ translator.translate(new LongValue(Long.MAX_VALUE), longValueWithProperHashCode));
+ }
+
+ // Long
+
+ @Test
+ public void testLongValueToLongTranslation() throws Exception {
+ TranslateFunction<LongValue, Long> translator = new LongValueToLong();
+
+ Assert.assertEquals(Long.valueOf(0L),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Long.valueOf(Long.MIN_VALUE),
+ translator.translate(new LongValue(Long.MIN_VALUE), null));
+
+ Assert.assertEquals(Long.valueOf(Long.MAX_VALUE),
+ translator.translate(new LongValue(Long.MAX_VALUE), null));
+ }
+
+ // FloatValue
+
+ @Test
+ public void testToFloatValue() throws Exception {
+ TranslateFunction<LongValue, FloatValue> translator = new LongValueToUnsignedFloatValue();
+
+ Assert.assertEquals(new FloatValue(Float.intBitsToFloat(0)),
+ translator.translate(new LongValue(0L), floatValue));
+
+ Assert.assertEquals(new FloatValue(Float.intBitsToFloat(Integer.MIN_VALUE)),
+ translator.translate(new LongValue((long) Integer.MAX_VALUE + 1), floatValue));
+
+ Assert.assertEquals(new FloatValue(Float.intBitsToFloat(-1)),
+ translator.translate(new LongValue(LongValueToUnsignedFloatValue.MAX_VERTEX_COUNT - 1), floatValue));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToFloatValueUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedFloatValue().translate(new LongValue(LongValueToUnsignedFloatValue.MAX_VERTEX_COUNT), floatValue);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToFloatValueLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedFloatValue().translate(new LongValue(-1), floatValue);
+ }
+
+ @Test
+ public void testFromFloatValue() throws Exception {
+ TranslateFunction<FloatValue, LongValueWithProperHashCode> translator = new UnsignedFloatValueToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(new FloatValue(Float.intBitsToFloat(0)), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode((long) Integer.MAX_VALUE + 1),
+ translator.translate(new FloatValue(Float.intBitsToFloat(Integer.MIN_VALUE)), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedFloatValue.MAX_VERTEX_COUNT - 1),
+ translator.translate(new FloatValue(Float.intBitsToFloat(-1)), longValueWithProperHashCode));
+ }
+
+ // Float
+
+ @Test
+ public void testToFloat() throws Exception {
+ TranslateFunction<LongValue, Float> translator = new LongValueToUnsignedFloat();
+
+ Assert.assertEquals(Float.valueOf(Float.intBitsToFloat(0)),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Float.valueOf(Float.intBitsToFloat(Integer.MIN_VALUE)),
+ translator.translate(new LongValue((long) Integer.MAX_VALUE + 1), null));
+
+ Assert.assertEquals(Float.valueOf(Float.intBitsToFloat(-1)),
+ translator.translate(new LongValue(LongValueToUnsignedFloat.MAX_VERTEX_COUNT - 1), null));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToFloatUpperOutOfRange() throws Exception {
+ new LongValueToUnsignedFloat().translate(new LongValue(LongValueToUnsignedFloat.MAX_VERTEX_COUNT), null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testToFloatLowerOutOfRange() throws Exception {
+ new LongValueToUnsignedFloat().translate(new LongValue(-1), null);
+ }
+
+ @Test
+ public void testFromFloat() throws Exception {
+ TranslateFunction<Float, LongValueWithProperHashCode> translator = new UnsignedFloatToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(Float.intBitsToFloat(0), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode((long) Integer.MAX_VALUE + 1),
+ translator.translate(Float.intBitsToFloat(Integer.MIN_VALUE), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(LongValueToUnsignedFloat.MAX_VERTEX_COUNT - 1),
+ translator.translate(Float.intBitsToFloat(-1), longValueWithProperHashCode));
+ }
+
+ // DoubleValue
+
+ @Test
+ public void testToDoubleValue() throws Exception {
+ TranslateFunction<LongValue, DoubleValue> translator = new LongValueToDoubleValue();
+
+ Assert.assertEquals(new DoubleValue(Double.longBitsToDouble(0L)),
+ translator.translate(new LongValue(0L), doubleValue));
+
+ Assert.assertEquals(new DoubleValue(Double.longBitsToDouble(Long.MIN_VALUE)),
+ translator.translate(new LongValue(Long.MIN_VALUE), doubleValue));
+
+ Assert.assertEquals(new DoubleValue(Double.longBitsToDouble(Long.MAX_VALUE)),
+ translator.translate(new LongValue(Long.MAX_VALUE), doubleValue));
+ }
+
+ @Test
+ public void testFromDoubleValue() throws Exception {
+ TranslateFunction<DoubleValue, LongValueWithProperHashCode> translator = new DoubleValueToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(new DoubleValue(Double.longBitsToDouble(0L)), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MIN_VALUE),
+ translator.translate(new DoubleValue(Double.longBitsToDouble(Long.MIN_VALUE)), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MAX_VALUE),
+ translator.translate(new DoubleValue(Double.longBitsToDouble(Long.MAX_VALUE)), longValueWithProperHashCode));
+ }
+
+ // Double
+
+ @Test
+ public void testToDouble() throws Exception {
+ TranslateFunction<LongValue, Double> translator = new LongValueToDouble();
+
+ Assert.assertEquals(Double.valueOf(Double.longBitsToDouble(0L)),
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals(Double.valueOf(Double.longBitsToDouble(Long.MIN_VALUE)),
+ translator.translate(new LongValue(Long.MIN_VALUE), null));
+
+ Assert.assertEquals(Double.valueOf(Double.longBitsToDouble(Long.MAX_VALUE)),
+ translator.translate(new LongValue(Long.MAX_VALUE), null));
+ }
+
+ @Test
+ public void testFromDouble() throws Exception {
+ TranslateFunction<Double, LongValueWithProperHashCode> translator = new DoubleToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(Double.longBitsToDouble(0L), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MIN_VALUE),
+ translator.translate(Double.longBitsToDouble(Long.MIN_VALUE), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MAX_VALUE),
+ translator.translate(Double.longBitsToDouble(Long.MAX_VALUE), longValueWithProperHashCode));
+ }
+
+ // StringValue
+
+ @Test
+ public void testFromStringValue() throws Exception {
+ TranslateFunction<StringValue, LongValueWithProperHashCode> translator = new StringValueToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate(new StringValue("0"), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MIN_VALUE),
+ translator.translate(new StringValue("-9223372036854775808"), longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MAX_VALUE),
+ translator.translate(new StringValue("9223372036854775807"), longValueWithProperHashCode));
+ }
+
+ // String
+
+ @Test
+ public void testLongValueToStringTranslation() throws Exception {
+ TranslateFunction<LongValue, String> translator = new LongValueToString();
+
+ Assert.assertEquals("0",
+ translator.translate(new LongValue(0L), null));
+
+ Assert.assertEquals("-9223372036854775808",
+ translator.translate(new LongValue(Long.MIN_VALUE), null));
+
+ Assert.assertEquals("9223372036854775807",
+ translator.translate(new LongValue(Long.MAX_VALUE), null));
+ }
+
+ @Test
+ public void testFromString() throws Exception {
+ TranslateFunction<String, LongValueWithProperHashCode> translator = new StringToLongValueWithProperHashCode();
+
+ Assert.assertEquals(new LongValueWithProperHashCode(0L),
+ translator.translate("0", longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MIN_VALUE),
+ translator.translate("-9223372036854775808", longValueWithProperHashCode));
+
+ Assert.assertEquals(new LongValueWithProperHashCode(Long.MAX_VALUE),
+ translator.translate("9223372036854775807", longValueWithProperHashCode));
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResult.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResult.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResult.java
index f41268c..21a226c 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResult.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResult.java
@@ -26,8 +26,10 @@ import java.io.Serializable;
/**
* A {@link GraphAlgorithm} result for a pair vertices.
+ *
+ * @param <K> graph ID type
*/
-public interface BinaryResult<T>
+public interface BinaryResult<K>
extends Serializable {
/**
@@ -35,33 +37,33 @@ extends Serializable {
*
* @return first vertex ID
*/
- T getVertexId0();
+ K getVertexId0();
/**
* Set the first vertex ID.
*
- * @param value new vertex ID
+ * @param vertexId0 new vertex ID
*/
- void setVertexId0(T value);
+ void setVertexId0(K vertexId0);
/**
* Get the second vertex ID.
*
* @return second vertex ID
*/
- T getVertexId1();
+ K getVertexId1();
/**
* Set the second vertex ID.
*
- * @param value new vertex ID
+ * @param vertexId1 new vertex ID
*/
- void setVertexId1(T value);
+ void setVertexId1(K vertexId1);
/**
* Output each input and a second result with the vertex order flipped.
*
- * @param <T> ID type
+ * @param <T> graph ID type
* @param <RT> result type
*/
class MirrorResult<T, RT extends BinaryResult<T>>
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResultBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResultBase.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResultBase.java
index 45791ee..1275075 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResultBase.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/BinaryResultBase.java
@@ -18,6 +18,9 @@
package org.apache.flink.graph.asm.result;
+import org.apache.flink.graph.asm.translate.TranslateFunction;
+import org.apache.flink.util.Collector;
+
/**
* Base class for algorithm results for a pair of vertices.
*
@@ -25,7 +28,7 @@ package org.apache.flink.graph.asm.result;
*/
public abstract class BinaryResultBase<K>
extends ResultBase
-implements BinaryResult<K> {
+implements BinaryResult<K>, TranslatableResult<K> {
private K vertexId0;
@@ -50,4 +53,43 @@ implements BinaryResult<K> {
public void setVertexId1(K vertexId1) {
this.vertexId1 = vertexId1;
}
+
+ @Override
+ public <T> TranslatableResult<T> translate(TranslateFunction<K, T> translator, TranslatableResult<T> reuse, Collector<TranslatableResult<T>> out)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new BasicBinaryResult<>();
+ }
+
+ K vertexId0 = this.getVertexId0();
+ K vertexId1 = this.getVertexId1();
+
+ BinaryResult<T> translatable = (BinaryResult<T>) reuse;
+ BinaryResult<T> translated = (BinaryResult<T>) this;
+
+ translated.setVertexId0(translator.translate(this.getVertexId0(), translatable.getVertexId0()));
+ translated.setVertexId1(translator.translate(this.getVertexId1(), translatable.getVertexId1()));
+
+ out.collect((TranslatableResult<T>) translated);
+
+ this.setVertexId0(vertexId0);
+ this.setVertexId1(vertexId1);
+
+ return reuse;
+ }
+
+ /**
+ * Simple override of {@code BinaryResultBase}. This holds no additional
+ * values but is used by {@link BinaryResultBase#translate} as the reuse
+ * object for translating vertex IDs.
+ *
+ * @param <U> result ID type
+ */
+ private static class BasicBinaryResult<U>
+ extends BinaryResultBase<U> {
+ @Override
+ public String toString() {
+ return "(" + getVertexId0() + "," + getVertexId1() + ")";
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResult.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResult.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResult.java
index a546387..61e1fb9 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResult.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResult.java
@@ -24,8 +24,10 @@ import java.io.Serializable;
/**
* A {@link GraphAlgorithm} result for three vertices.
+ *
+ * @param <K> graph ID type
*/
-public interface TertiaryResult<T>
+public interface TertiaryResult<K>
extends Serializable {
/**
@@ -33,40 +35,40 @@ extends Serializable {
*
* @return first vertex ID
*/
- T getVertexId0();
+ K getVertexId0();
/**
* Set the first vertex ID.
*
- * @param value new vertex ID
+ * @param vertexId0 new vertex ID
*/
- void setVertexId0(T value);
+ void setVertexId0(K vertexId0);
/**
* Get the second vertex ID.
*
* @return second vertex ID
*/
- T getVertexId1();
+ K getVertexId1();
/**
* Set the second vertex ID.
*
- * @param value new vertex ID
+ * @param vertexId1 new vertex ID
*/
- void setVertexId1(T value);
+ void setVertexId1(K vertexId1);
/**
* Get the third vertex ID.
*
* @return third vertex ID
*/
- T getVertexId2();
+ K getVertexId2();
/**
* Set the third vertex ID.
*
- * @param value new vertex ID
+ * @param vertexId2 new vertex ID
*/
- void setVertexId2(T value);
+ void setVertexId2(K vertexId2);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResultBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResultBase.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResultBase.java
index eb505b8..a63cf90 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResultBase.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TertiaryResultBase.java
@@ -19,6 +19,9 @@
package org.apache.flink.graph.asm.result;
+import org.apache.flink.graph.asm.translate.TranslateFunction;
+import org.apache.flink.util.Collector;
+
/**
* Base class for algorithm results for three vertices.
*
@@ -26,7 +29,7 @@ package org.apache.flink.graph.asm.result;
*/
public abstract class TertiaryResultBase<K>
extends ResultBase
-implements TertiaryResult<K> {
+implements TertiaryResult<K>, TranslatableResult<K> {
private K vertexId0;
@@ -63,4 +66,46 @@ implements TertiaryResult<K> {
public void setVertexId2(K vertexId2) {
this.vertexId2 = vertexId2;
}
+
+ @Override
+ public <T> TranslatableResult<T> translate(TranslateFunction<K, T> translator, TranslatableResult<T> reuse, Collector<TranslatableResult<T>> out)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new BasicTertiaryResult<>();
+ }
+
+ K vertexId0 = this.getVertexId0();
+ K vertexId1 = this.getVertexId1();
+ K vertexId2 = this.getVertexId2();
+
+ TertiaryResult<T> translatable = (TertiaryResult<T>) reuse;
+ TertiaryResult<T> translated = (TertiaryResult<T>) this;
+
+ translated.setVertexId0(translator.translate(this.getVertexId0(), translatable.getVertexId0()));
+ translated.setVertexId1(translator.translate(this.getVertexId1(), translatable.getVertexId1()));
+ translated.setVertexId2(translator.translate(this.getVertexId2(), translatable.getVertexId2()));
+
+ out.collect((TranslatableResult<T>) translated);
+
+ this.setVertexId0(vertexId0);
+ this.setVertexId1(vertexId1);
+ this.setVertexId2(vertexId2);
+
+ return reuse;
+ }
+
+ /**
+ * Simple override of {@code TertiaryResultBase}. This holds no additional
+ * values but is used by {@link TertiaryResultBase#translate} as the reuse
+ * object for translating vertex IDs.
+ *
+ * @param <U> result ID type
+ */
+ private static class BasicTertiaryResult<U>
+ extends TertiaryResultBase<U> {
+ @Override
+ public String toString() {
+ return "(" + getVertexId0() + "," + getVertexId1() + "," + getVertexId2() + ")";
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TranslatableResult.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TranslatableResult.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TranslatableResult.java
new file mode 100644
index 0000000..3bd2a1e
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/TranslatableResult.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.asm.result;
+
+import org.apache.flink.graph.asm.translate.TranslateFunction;
+import org.apache.flink.util.Collector;
+
+/**
+ * A transformable result can transform its ID type.
+ *
+ * @param <K> graph ID type
+ */
+public interface TranslatableResult<K> {
+
+ /**
+ * Output the result after transforming the vertex ID type.
+ *
+ * @param translator translates type {@code K} to type {@code T}
+ * @param reuse reusable value
+ * @param out output collector
+ * @return reusable result
+ * @throws Exception on error
+ *
+ * @param <T> ID output type
+ */
+ <T> TranslatableResult<T> translate(TranslateFunction<K, T> translator, TranslatableResult<T> reuse, Collector<TranslatableResult<T>> out) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResult.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResult.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResult.java
index d861b43..88e6c00 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResult.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResult.java
@@ -24,8 +24,10 @@ import java.io.Serializable;
/**
* A {@link GraphAlgorithm} result for a single vertex.
+ *
+ * @param <K> graph ID type
*/
-public interface UnaryResult<T>
+public interface UnaryResult<K>
extends Serializable {
/**
@@ -33,12 +35,12 @@ extends Serializable {
*
* @return first vertex ID
*/
- T getVertexId0();
+ K getVertexId0();
/**
* Set the first vertex ID.
*
- * @param value new vertex ID
+ * @param vertexId0 new vertex ID
*/
- void setVertexId0(T value);
+ void setVertexId0(K vertexId0);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResultBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResultBase.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResultBase.java
index 5bd3229..727f82c 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResultBase.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/result/UnaryResultBase.java
@@ -19,6 +19,9 @@
package org.apache.flink.graph.asm.result;
+import org.apache.flink.graph.asm.translate.TranslateFunction;
+import org.apache.flink.util.Collector;
+
/**
* Base class for algorithm results for a single vertex.
*
@@ -26,7 +29,7 @@ package org.apache.flink.graph.asm.result;
*/
public abstract class UnaryResultBase<K>
extends ResultBase
-implements UnaryResult<K> {
+implements UnaryResult<K>, TranslatableResult<K> {
private K vertexId0;
@@ -39,4 +42,40 @@ implements UnaryResult<K> {
public void setVertexId0(K vertexId0) {
this.vertexId0 = vertexId0;
}
+
+ @Override
+ public <T> TranslatableResult<T> translate(TranslateFunction<K, T> translator, TranslatableResult<T> reuse, Collector<TranslatableResult<T>> out)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new BasicUnaryResult<>();
+ }
+
+ K vertexId0 = this.getVertexId0();
+
+ UnaryResult<T> translatable = (UnaryResult<T>) reuse;
+ UnaryResult<T> translated = (UnaryResult<T>) this;
+
+ translated.setVertexId0(translator.translate(this.getVertexId0(), translatable.getVertexId0()));
+
+ out.collect((TranslatableResult<T>) translated);
+
+ this.setVertexId0(vertexId0);
+
+ return reuse;
+ }
+
+ /**
+ * Simple override of {@code UnaryResultBase}. This holds no additional
+ * values but is used by {@link UnaryResultBase#translate} as the reuse
+ * object for translating vertex IDs.
+ *
+ * @param <U> result ID type
+ */
+ private static class BasicUnaryResult<U>
+ extends UnaryResultBase<U> {
+ @Override
+ public String toString() {
+ return "(" + getVertexId0() + ")";
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/translate/translators/LongValueToUnsignedIntValue.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/translate/translators/LongValueToUnsignedIntValue.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/translate/translators/LongValueToUnsignedIntValue.java
index d7896d3..c495b25 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/translate/translators/LongValueToUnsignedIntValue.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/translate/translators/LongValueToUnsignedIntValue.java
@@ -43,10 +43,9 @@ implements TranslateFunction<LongValue, IntValue> {
if (l < 0 || l >= MAX_VERTEX_COUNT) {
throw new IllegalArgumentException("Cannot cast long value " + value + " to integer.");
- } else {
- reuse.setValue((int) (l & (MAX_VERTEX_COUNT - 1)));
}
+ reuse.setValue((int) (l & (MAX_VERTEX_COUNT - 1)));
return reuse;
}
}
[5/5] flink git commit: [FLINK-6648] [gelly] Transforms for Gelly
examples
Posted by gr...@apache.org.
[FLINK-6648] [gelly] Transforms for Gelly examples
Replaces GeneratedGraph class (which was extended by inputs) with the
GraphKeyTypeTransform which can also transform the algorithm result to
obtain consistent hash codes. This allows for the removal of the case
statements in the driver checksum tests.
Float and double are now supported types.
This closes #4304
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8695a210
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8695a210
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8695a210
Branch: refs/heads/master
Commit: 8695a21098c06cd1cd727d24bca51db9971cb146
Parents: 3e12673
Author: Greg Hogan <co...@greghogan.com>
Authored: Wed Jul 5 16:58:06 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed Jul 26 10:31:20 2017 -0400
----------------------------------------------------------------------
.../java/org/apache/flink/graph/Runner.java | 102 ++-
.../apache/flink/graph/drivers/AdamicAdar.java | 4 +
.../apache/flink/graph/drivers/DriverBase.java | 5 -
.../flink/graph/drivers/TriangleListing.java | 4 +
.../graph/drivers/input/CirculantGraph.java | 4 +-
.../graph/drivers/input/CompleteGraph.java | 6 +-
.../flink/graph/drivers/input/CycleGraph.java | 6 +-
.../flink/graph/drivers/input/EchoGraph.java | 4 +-
.../flink/graph/drivers/input/EmptyGraph.java | 6 +-
.../graph/drivers/input/GeneratedGraph.java | 342 +-------
.../drivers/input/GeneratedMultiGraph.java | 9 +-
.../flink/graph/drivers/input/GridGraph.java | 6 +-
.../graph/drivers/input/HypercubeGraph.java | 6 +-
.../apache/flink/graph/drivers/input/Input.java | 2 +-
.../flink/graph/drivers/input/InputBase.java | 5 -
.../flink/graph/drivers/input/PathGraph.java | 6 +-
.../flink/graph/drivers/input/RMatGraph.java | 5 +-
.../graph/drivers/input/SingletonEdgeGraph.java | 6 +-
.../flink/graph/drivers/input/StarGraph.java | 6 +-
.../apache/flink/graph/drivers/output/Hash.java | 4 +-
.../flink/graph/drivers/output/OutputBase.java | 5 -
.../flink/graph/drivers/output/Print.java | 9 +-
.../drivers/parameter/ChoiceParameter.java | 5 +
.../drivers/parameter/ParameterizedBase.java | 5 +
.../transform/GraphKeyTypeTransform.java | 841 +++++++++++++++++++
.../transform/LongValueWithProperHashCode.java | 229 +++++
.../graph/drivers/transform/Transform.java | 61 ++
.../graph/drivers/transform/Transformable.java | 36 +
.../flink/graph/drivers/AdamicAdarITCase.java | 4 +-
.../drivers/ClusteringCoefficientITCase.java | 114 +--
.../drivers/ConnectedComponentsITCase.java | 87 +-
.../drivers/CopyableValueDriverBaseITCase.java | 2 +-
.../flink/graph/drivers/DriverBaseITCase.java | 3 +-
.../flink/graph/drivers/EdgeListITCase.java | 370 +-------
.../flink/graph/drivers/JaccardIndexITCase.java | 49 +-
.../NonTransformableDriverBaseITCase.java | 54 ++
.../graph/drivers/TriangleListingITCase.java | 138 +--
.../graph/drivers/input/GeneratedGraphTest.java | 207 -----
.../transform/GraphKeyTypeTransformTest.java | 534 ++++++++++++
.../flink/graph/asm/result/BinaryResult.java | 18 +-
.../graph/asm/result/BinaryResultBase.java | 44 +-
.../flink/graph/asm/result/TertiaryResult.java | 22 +-
.../graph/asm/result/TertiaryResultBase.java | 47 +-
.../graph/asm/result/TranslatableResult.java | 43 +
.../flink/graph/asm/result/UnaryResult.java | 10 +-
.../flink/graph/asm/result/UnaryResultBase.java | 41 +-
.../LongValueToUnsignedIntValue.java | 3 +-
.../library/clustering/TriangleListingBase.java | 27 +
.../clustering/directed/TriangleListing.java | 101 ++-
.../clustering/undirected/TriangleListing.java | 59 +-
.../graph/library/similarity/AdamicAdar.java | 32 +-
.../directed/TriangleListingTest.java | 35 +-
.../undirected/TriangleListingTest.java | 37 +-
53 files changed, 2461 insertions(+), 1349 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Runner.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Runner.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Runner.java
index d0e6a92..2a8b2df 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Runner.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Runner.java
@@ -53,6 +53,8 @@ import org.apache.flink.graph.drivers.parameter.BooleanParameter;
import org.apache.flink.graph.drivers.parameter.Parameterized;
import org.apache.flink.graph.drivers.parameter.ParameterizedBase;
import org.apache.flink.graph.drivers.parameter.StringParameter;
+import org.apache.flink.graph.drivers.transform.Transform;
+import org.apache.flink.graph.drivers.transform.Transformable;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.util.InstantiationUtil;
@@ -64,6 +66,7 @@ import org.apache.commons.lang3.text.StrBuilder;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -127,6 +130,9 @@ extends ParameterizedBase {
private final StringParameter jobDetailsPath = new StringParameter(this, "__job_details_path")
.setDefaultValue(null);
+ private StringParameter jobName = new StringParameter(this, "__job_name")
+ .setDefaultValue(null);
+
/**
* Create an algorithm runner from the given arguments.
*
@@ -225,6 +231,21 @@ extends ParameterizedBase {
.toString();
}
+ /**
+ * Configure a runtime component. Catch {@link RuntimeException} and
+ * re-throw with a Flink internal exception which is processed by
+ * CliFrontend for display to the user.
+ *
+ * @param parameterized the component to be configured
+ */
+ private void parameterize(Parameterized parameterized) {
+ try {
+ parameterized.configure(parameters);
+ } catch (RuntimeException ex) {
+ throw new ProgramParametrizationException(ex.getMessage());
+ }
+ }
+
public void run() throws Exception {
// Set up the execution environment
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -235,6 +256,7 @@ extends ParameterizedBase {
config.disableForceKryo();
config.setGlobalJobParameters(parameters);
+ parameterize(this);
// configure local parameters and throw proper exception on error
try {
@@ -275,11 +297,7 @@ extends ParameterizedBase {
throw new ProgramParametrizationException("No input given");
}
- try {
- algorithm.configure(parameters);
- } catch (RuntimeException ex) {
- throw new ProgramParametrizationException(ex.getMessage());
- }
+ parameterize(algorithm);
String inputName = parameters.get(INPUT);
Input input = inputFactory.get(inputName);
@@ -288,11 +306,7 @@ extends ParameterizedBase {
throw new ProgramParametrizationException("Unknown input type: " + inputName);
}
- try {
- input.configure(parameters);
- } catch (RuntimeException ex) {
- throw new ProgramParametrizationException(ex.getMessage());
- }
+ parameterize(input);
// output and usage
if (!parameters.has(OUTPUT)) {
@@ -306,10 +320,29 @@ extends ParameterizedBase {
throw new ProgramParametrizationException("Unknown output type: " + outputName);
}
- try {
- output.configure(parameters);
- } catch (RuntimeException ex) {
- throw new ProgramParametrizationException(ex.getMessage());
+ parameterize(output);
+
+ // ----------------------------------------------------------------------------------------
+ // Create list of input and algorithm transforms
+ // ----------------------------------------------------------------------------------------
+
+ List<Transform> transforms = new ArrayList<>();
+
+ if (input instanceof Transformable) {
+ transforms.addAll(((Transformable) input).getTransformers());
+ }
+
+ if (algorithm instanceof Transformable) {
+ transforms.addAll(((Transformable) algorithm).getTransformers());
+ }
+
+ for (Transform transform : transforms) {
+ parameterize(transform);
+ }
+
+ // unused parameters
+ if (parameters.getUnrequestedParameters().size() > 0) {
+ throw new ProgramParametrizationException("Unrequested parameters: " + parameters.getUnrequestedParameters());
}
// ----------------------------------------------------------------------------------------
@@ -319,20 +352,55 @@ extends ParameterizedBase {
// Create input
Graph graph = input.create(env);
+ // Transform input
+ for (Transform transform : transforms) {
+ graph = (Graph) transform.transformInput(graph);
+ }
+
// Run algorithm
DataSet results = algorithm.plan(graph);
// Output
- String executionName = input.getIdentity() + " ⇨ " + algorithmName + " ⇨ " + output.getName();
+ String executionName = jobName.getValue() != null ? jobName.getValue() + ": " : "";
- System.out.println();
+ executionName += input.getIdentity() + " ⇨ " + algorithmName + " ⇨ " + output.getName();
+
+ if (transforms.size() > 0) {
+ // append identifiers to job name
+ StringBuffer buffer = new StringBuffer(executionName).append(" [");
+
+ for (Transform transform : transforms) {
+ buffer.append(transform.getIdentity());
+ }
+
+ executionName = buffer.append("]").toString();
+ }
+
+ if (output == null) {
+ throw new ProgramParametrizationException("Unknown output type: " + outputName);
+ }
+
+ try {
+ output.configure(parameters);
+ } catch (RuntimeException ex) {
+ throw new ProgramParametrizationException(ex.getMessage());
+ }
if (results == null) {
env.execute(executionName);
} else {
- output.write(executionName, System.out, results);
+ // Transform output if algorithm returned result DataSet
+ if (transforms.size() > 0) {
+ Collections.reverse(transforms);
+ for (Transform transform : transforms) {
+ results = (DataSet) transform.transformResult(results);
+ }
+ }
+
+ output.write(executionName.toString(), System.out, results);
}
+ System.out.println();
algorithm.printAnalytics(System.out);
if (jobDetailsPath.getValue() != null) {
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/AdamicAdar.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/AdamicAdar.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/AdamicAdar.java
index 5dd8ea1..7f58cf5 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/AdamicAdar.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/AdamicAdar.java
@@ -20,6 +20,7 @@ package org.apache.flink.graph.drivers;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.drivers.parameter.BooleanParameter;
import org.apache.flink.graph.drivers.parameter.DoubleParameter;
import org.apache.flink.types.CopyableValue;
@@ -40,6 +41,8 @@ extends DriverBase<K, VV, EV> {
.setDefaultValue(0.0)
.setMinimumValue(0.0, true);
+ private BooleanParameter mirrorResults = new BooleanParameter(this, "mirror_results");
+
@Override
public String getShortDescription() {
return "similarity score weighted by centerpoint degree";
@@ -62,6 +65,7 @@ extends DriverBase<K, VV, EV> {
.run(new org.apache.flink.graph.library.similarity.AdamicAdar<K, VV, EV>()
.setMinimumRatio(minRatio.getValue().floatValue())
.setMinimumScore(minScore.getValue().floatValue())
+ .setMirrorResults(mirrorResults.getValue())
.setParallelism(parallelism.getValue().intValue()));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/DriverBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/DriverBase.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/DriverBase.java
index 5b5a684..4337a52 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/DriverBase.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/DriverBase.java
@@ -40,11 +40,6 @@ implements Driver<K, VV, EV> {
.setDefaultValue(PARALLELISM_DEFAULT);
@Override
- public String getName() {
- return this.getClass().getSimpleName();
- }
-
- @Override
public void printAnalytics(PrintStream out) {
// analytics are optionally executed by drivers overriding this method
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java
index 4a7d230..41dc1dc 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java
@@ -53,6 +53,8 @@ extends DriverBase<K, VV, EV> {
private BooleanParameter computeTriadicCensus = new BooleanParameter(this, "triadic_census");
+ private BooleanParameter permuteResults = new BooleanParameter(this, "permute_results");
+
private GraphAnalytic<K, VV, EV, ? extends PrintableResult> triadicCensus;
@Override
@@ -86,6 +88,7 @@ extends DriverBase<K, VV, EV> {
@SuppressWarnings("unchecked")
DataSet<PrintableResult> directedResult = (DataSet<PrintableResult>) (DataSet<?>) graph
.run(new org.apache.flink.graph.library.clustering.directed.TriangleListing<K, VV, EV>()
+ .setPermuteResults(permuteResults.getValue())
.setSortTriangleVertices(sortTriangleVertices.getValue())
.setParallelism(parallelism));
return directedResult;
@@ -100,6 +103,7 @@ extends DriverBase<K, VV, EV> {
@SuppressWarnings("unchecked")
DataSet<PrintableResult> undirectedResult = (DataSet<PrintableResult>) (DataSet<?>) graph
.run(new org.apache.flink.graph.library.clustering.undirected.TriangleListing<K, VV, EV>()
+ .setPermuteResults(permuteResults.getValue())
.setSortTriangleVertices(sortTriangleVertices.getValue())
.setParallelism(parallelism));
return undirectedResult;
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CirculantGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CirculantGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CirculantGraph.java
index f7364ce..edd7384 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CirculantGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CirculantGraph.java
@@ -96,7 +96,7 @@ extends GeneratedGraph<LongValue> {
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + offsetRanges + ")";
+ return getName() + " (" + offsetRanges + ")";
}
@Override
@@ -105,7 +105,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
org.apache.flink.graph.generator.CirculantGraph graph = new org.apache.flink.graph.generator.CirculantGraph(env,
vertexCount.getValue());
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CompleteGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CompleteGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CompleteGraph.java
index d3a7e49..59470a7 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CompleteGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CompleteGraph.java
@@ -30,14 +30,14 @@ import static org.apache.flink.graph.generator.CompleteGraph.MINIMUM_VERTEX_COUN
* Generate a {@link org.apache.flink.graph.generator.CompleteGraph}.
*/
public class CompleteGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter vertexCount = new LongParameter(this, "vertex_count")
.setMinimumValue(MINIMUM_VERTEX_COUNT);
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexCount.getValue() + ")";
+ return getName() + " (" + vertexCount.getValue() + ")";
}
@Override
@@ -46,7 +46,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- protected Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) throws Exception {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) throws Exception {
return new org.apache.flink.graph.generator.CompleteGraph(env, vertexCount.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CycleGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CycleGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CycleGraph.java
index 632ad77..6200e05 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CycleGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/CycleGraph.java
@@ -30,14 +30,14 @@ import static org.apache.flink.graph.generator.CycleGraph.MINIMUM_VERTEX_COUNT;
* Generate a {@link org.apache.flink.graph.generator.CycleGraph}.
*/
public class CycleGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter vertexCount = new LongParameter(this, "vertex_count")
.setMinimumValue(MINIMUM_VERTEX_COUNT);
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexCount + ")";
+ return getName() + " (" + vertexCount + ")";
}
@Override
@@ -46,7 +46,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
return new org.apache.flink.graph.generator.CycleGraph(env, vertexCount.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EchoGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EchoGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EchoGraph.java
index 1eb0dd9..c568c49 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EchoGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EchoGraph.java
@@ -41,7 +41,7 @@ extends GeneratedGraph<LongValue> {
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexCount.getValue() + ":" + vertexDegree.getValue() + ")";
+ return getName() + " (" + vertexCount.getValue() + ":" + vertexDegree.getValue() + ")";
}
@Override
@@ -50,7 +50,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- protected Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) throws Exception {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) throws Exception {
return new org.apache.flink.graph.generator.EchoGraph(env, vertexCount.getValue(), vertexDegree.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EmptyGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EmptyGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EmptyGraph.java
index f54fcba..e318b9f 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EmptyGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/EmptyGraph.java
@@ -30,14 +30,14 @@ import static org.apache.flink.graph.generator.EmptyGraph.MINIMUM_VERTEX_COUNT;
* Generate an {@link org.apache.flink.graph.generator.EmptyGraph}.
*/
public class EmptyGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter vertexCount = new LongParameter(this, "vertex_count")
.setMinimumValue(MINIMUM_VERTEX_COUNT);
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexCount + ")";
+ return getName() + " (" + vertexCount + ")";
}
@Override
@@ -46,7 +46,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
return new org.apache.flink.graph.generator.EmptyGraph(env, vertexCount.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedGraph.java
index f6d8ae9..cb4ff6e 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedGraph.java
@@ -18,22 +18,14 @@
package org.apache.flink.graph.drivers.input;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.client.program.ProgramParametrizationException;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.asm.translate.TranslateFunction;
-import org.apache.flink.graph.asm.translate.TranslateGraphIds;
-import org.apache.flink.graph.asm.translate.translators.LongValueToStringValue;
-import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue;
-import org.apache.flink.graph.drivers.parameter.ChoiceParameter;
import org.apache.flink.graph.drivers.parameter.LongParameter;
-import org.apache.flink.types.ByteValue;
-import org.apache.flink.types.CharValue;
-import org.apache.flink.types.LongValue;
+import org.apache.flink.graph.drivers.transform.GraphKeyTypeTransform;
+import org.apache.flink.graph.drivers.transform.Transform;
+import org.apache.flink.graph.drivers.transform.Transformable;
import org.apache.flink.types.NullValue;
-import org.apache.flink.types.ShortValue;
-import org.apache.commons.lang3.text.WordUtils;
+import java.util.Arrays;
+import java.util.List;
import static org.apache.flink.api.common.ExecutionConfig.PARALLELISM_DEFAULT;
@@ -43,30 +35,13 @@ import static org.apache.flink.api.common.ExecutionConfig.PARALLELISM_DEFAULT;
* @param <K> graph ID type
*/
public abstract class GeneratedGraph<K>
-extends InputBase<K, NullValue, NullValue> {
+extends InputBase<K, NullValue, NullValue>
+implements Transformable {
- private static final String BYTE = "byte";
- private static final String NATIVE_BYTE = "nativeByte";
-
- private static final String SHORT = "short";
- private static final String NATIVE_SHORT = "nativeShort";
-
- private static final String CHAR = "char";
- private static final String NATIVE_CHAR = "nativeChar";
-
- private static final String INTEGER = "integer";
- private static final String NATIVE_INTEGER = "nativeInteger";
-
- private static final String LONG = "long";
- private static final String NATIVE_LONG = "nativeLong";
-
- private static final String STRING = "string";
- private static final String NATIVE_STRING = "nativeString";
-
- private ChoiceParameter type = new ChoiceParameter(this, "type")
- .setDefaultValue(INTEGER)
- .addChoices(LONG, STRING)
- .addHiddenChoices(BYTE, NATIVE_BYTE, SHORT, NATIVE_SHORT, CHAR, NATIVE_CHAR, NATIVE_INTEGER, NATIVE_LONG, NATIVE_STRING);
+ @Override
+ public List<Transform> getTransformers() {
+ return Arrays.<Transform>asList(new GraphKeyTypeTransform(vertexCount()));
+ }
protected LongParameter parallelism = new LongParameter(this, "__parallelism")
.setDefaultValue(PARALLELISM_DEFAULT);
@@ -79,299 +54,4 @@ extends InputBase<K, NullValue, NullValue> {
* @return number of vertices configured for the graph
*/
protected abstract long vertexCount();
-
- /**
- * Generate the graph as configured.
- *
- * @param env Flink execution environment
- * @return generated graph
- * @throws Exception on error
- */
- protected abstract Graph<K, NullValue, NullValue> generate(ExecutionEnvironment env) throws Exception;
-
- /**
- * Get the name of the type.
- *
- * @return name of the type
- */
- protected String getTypeName() {
- return WordUtils.capitalize(type.getValue());
- }
-
- @Override
- public Graph<K, NullValue, NullValue> create(ExecutionEnvironment env)
- throws Exception {
- long maxVertexCount = Long.MAX_VALUE;
- TranslateFunction translator = null;
-
- switch (type.getValue()) {
- case BYTE:
- maxVertexCount = LongValueToUnsignedByteValue.MAX_VERTEX_COUNT;
- translator = new LongValueToUnsignedByteValue();
- break;
-
- case NATIVE_BYTE:
- maxVertexCount = LongValueToUnsignedByte.MAX_VERTEX_COUNT;
- translator = new LongValueToUnsignedByte();
- break;
-
- case SHORT:
- maxVertexCount = LongValueToUnsignedShortValue.MAX_VERTEX_COUNT;
- translator = new LongValueToUnsignedShortValue();
- break;
-
- case NATIVE_SHORT:
- maxVertexCount = LongValueToUnsignedShort.MAX_VERTEX_COUNT;
- translator = new LongValueToUnsignedShort();
- break;
-
- case CHAR:
- maxVertexCount = LongValueToCharValue.MAX_VERTEX_COUNT;
- translator = new LongValueToCharValue();
- break;
-
- case NATIVE_CHAR:
- maxVertexCount = LongValueToChar.MAX_VERTEX_COUNT;
- translator = new LongValueToChar();
- break;
-
- case INTEGER:
- maxVertexCount = LongValueToUnsignedIntValue.MAX_VERTEX_COUNT;
- translator = new LongValueToUnsignedIntValue();
- break;
-
- case NATIVE_INTEGER:
- maxVertexCount = LongValueToUnsignedInt.MAX_VERTEX_COUNT;
- translator = new LongValueToUnsignedInt();
- break;
-
- case LONG:
- break;
-
- case NATIVE_LONG:
- translator = new LongValueToLong();
- break;
-
- case STRING:
- translator = new LongValueToStringValue();
- break;
-
- case NATIVE_STRING:
- translator = new LongValueToString();
- break;
-
- default:
- throw new ProgramParametrizationException("Unknown type '" + type.getValue() + "'");
- }
-
- long vertexCount = vertexCount();
- if (vertexCount > maxVertexCount) {
- throw new ProgramParametrizationException("Vertex count '" + vertexCount +
- "' must be no greater than " + maxVertexCount + " for type '" + type.getValue() + "'.");
- }
-
- Graph<K, NullValue, NullValue> graph = generate(env);
-
- if (translator != null) {
- graph = (Graph<K, NullValue, NullValue>) graph.run(new TranslateGraphIds(translator));
- }
-
- return graph;
- }
-
- /**
- * Translate {@link LongValue} to {@link ByteValue}.
- *
- * <p>Throws {@link RuntimeException} for byte overflow.
- */
- static class LongValueToUnsignedByteValue
- implements TranslateFunction<LongValue, ByteValue> {
- public static final long MAX_VERTEX_COUNT = 1L << 8;
-
- @Override
- public ByteValue translate(LongValue value, ByteValue reuse)
- throws Exception {
- if (reuse == null) {
- reuse = new ByteValue();
- }
-
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to byte.");
- } else {
- reuse.setValue((byte) (l & (MAX_VERTEX_COUNT - 1)));
- }
-
- return reuse;
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link Byte}.
- *
- * <p>Throws {@link RuntimeException} for byte overflow.
- */
- static class LongValueToUnsignedByte
- implements TranslateFunction<LongValue, Byte> {
- public static final long MAX_VERTEX_COUNT = 1L << 8;
-
- @Override
- public Byte translate(LongValue value, Byte reuse)
- throws Exception {
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to byte.");
- }
-
- return (byte) (l & (MAX_VERTEX_COUNT - 1));
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link ShortValue}.
- *
- * <p>Throws {@link RuntimeException} for short overflow.
- */
- static class LongValueToUnsignedShortValue
- implements TranslateFunction<LongValue, ShortValue> {
- public static final long MAX_VERTEX_COUNT = 1L << 16;
-
- @Override
- public ShortValue translate(LongValue value, ShortValue reuse)
- throws Exception {
- if (reuse == null) {
- reuse = new ShortValue();
- }
-
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to short.");
- } else {
- reuse.setValue((short) (l & (MAX_VERTEX_COUNT - 1)));
- }
-
- return reuse;
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link Short}.
- *
- * <p>Throws {@link RuntimeException} for short overflow.
- */
- static class LongValueToUnsignedShort
- implements TranslateFunction<LongValue, Short> {
- public static final long MAX_VERTEX_COUNT = 1L << 16;
-
- @Override
- public Short translate(LongValue value, Short reuse)
- throws Exception {
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to short.");
- }
-
- return (short) (l & (MAX_VERTEX_COUNT - 1));
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link CharValue}.
- *
- * <p>Throws {@link RuntimeException} for char overflow.
- */
- static class LongValueToCharValue
- implements TranslateFunction<LongValue, CharValue> {
- public static final long MAX_VERTEX_COUNT = 1L << 16;
-
- @Override
- public CharValue translate(LongValue value, CharValue reuse)
- throws Exception {
- if (reuse == null) {
- reuse = new CharValue();
- }
-
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to char.");
- } else {
- reuse.setValue((char) (l & (MAX_VERTEX_COUNT - 1)));
- }
-
- return reuse;
- }
- }
-
- /**
- * Translate {@link LongValue} to {@code Character}.
- *
- * <p>Throws {@link RuntimeException} for char overflow.
- */
- static class LongValueToChar
- implements TranslateFunction<LongValue, Character> {
- public static final long MAX_VERTEX_COUNT = 1L << 16;
-
- @Override
- public Character translate(LongValue value, Character reuse)
- throws Exception {
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to char.");
- }
-
- return (char) (l & (MAX_VERTEX_COUNT - 1));
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link Integer}.
- *
- * <p>Throws {@link RuntimeException} for integer overflow.
- */
- static class LongValueToUnsignedInt
- implements TranslateFunction<LongValue, Integer> {
- public static final long MAX_VERTEX_COUNT = 1L << 32;
-
- @Override
- public Integer translate(LongValue value, Integer reuse)
- throws Exception {
- long l = value.getValue();
-
- if (l < 0 || l >= MAX_VERTEX_COUNT) {
- throw new IllegalArgumentException("Cannot cast long value " + value + " to integer.");
- }
-
- return (int) (l & (MAX_VERTEX_COUNT - 1));
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link Long}.
- */
- static class LongValueToLong
- implements TranslateFunction<LongValue, Long> {
- @Override
- public Long translate(LongValue value, Long reuse)
- throws Exception {
- return value.getValue();
- }
- }
-
- /**
- * Translate {@link LongValue} to {@link String}.
- */
- static class LongValueToString
- implements TranslateFunction<LongValue, String> {
- @Override
- public String translate(LongValue value, String reuse)
- throws Exception {
- return Long.toString(value.getValue());
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedMultiGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedMultiGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedMultiGraph.java
index 02c8e58..c4dccb5 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedMultiGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GeneratedMultiGraph.java
@@ -21,6 +21,7 @@ package org.apache.flink.graph.drivers.input;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.graph.Graph;
import org.apache.flink.graph.drivers.parameter.Simplify;
+import org.apache.flink.types.LongValue;
import org.apache.flink.types.NullValue;
/**
@@ -29,7 +30,7 @@ import org.apache.flink.types.NullValue;
* @param <K> graph ID type
*/
public abstract class GeneratedMultiGraph<K extends Comparable<K>>
-extends GeneratedGraph<K> {
+extends GeneratedGraph<LongValue> {
private Simplify simplify = new Simplify(this);
@@ -43,12 +44,14 @@ extends GeneratedGraph<K> {
}
@Override
- public Graph<K, NullValue, NullValue> create(ExecutionEnvironment env)
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env)
throws Exception {
- Graph<K, NullValue, NullValue> graph = super.create(env);
+ Graph<LongValue, NullValue, NullValue> graph = generate(env);
// simplify after the translation to improve the performance of the
// simplify operators by processing smaller data types
return simplify.simplify(graph);
}
+
+ public abstract Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GridGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GridGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GridGraph.java
index b92b175..1421b1d 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GridGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/GridGraph.java
@@ -35,7 +35,7 @@ import java.util.TreeMap;
* Generate a {@link org.apache.flink.graph.generator.GridGraph}.
*/
public class GridGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private static final String PREFIX = "dim";
@@ -71,7 +71,7 @@ extends GeneratedGraph<LongValue> {
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + dimensions + ")";
+ return getName() + " (" + dimensions + ")";
}
@Override
@@ -91,7 +91,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
org.apache.flink.graph.generator.GridGraph graph = new org.apache.flink.graph.generator.GridGraph(env);
for (Dimension dimension : dimensions) {
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/HypercubeGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/HypercubeGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/HypercubeGraph.java
index 00b40c7..8c71916 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/HypercubeGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/HypercubeGraph.java
@@ -30,7 +30,7 @@ import static org.apache.flink.graph.generator.HypercubeGraph.MINIMUM_DIMENSIONS
* Generate a {@link org.apache.flink.graph.generator.HypercubeGraph}.
*/
public class HypercubeGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter dimensions = new LongParameter(this, "dimensions")
.setMinimumValue(MINIMUM_DIMENSIONS)
@@ -38,7 +38,7 @@ extends GeneratedGraph<LongValue> {
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + dimensions + ")";
+ return getName() + " (" + dimensions + ")";
}
@Override
@@ -47,7 +47,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
return new org.apache.flink.graph.generator.HypercubeGraph(env, dimensions.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/Input.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/Input.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/Input.java
index d647dd6..c80e6ab 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/Input.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/Input.java
@@ -37,7 +37,7 @@ extends Parameterized {
/**
* A human-readable identifier summarizing the input and configuration.
*
- * @return the unique identifier
+ * @return the input identifier
*/
String getIdentity();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/InputBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/InputBase.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/InputBase.java
index 60b5217..9e5a1b8 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/InputBase.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/InputBase.java
@@ -31,9 +31,4 @@ import org.apache.flink.graph.drivers.parameter.ParameterizedBase;
public abstract class InputBase<K, VV, EV>
extends ParameterizedBase
implements Input<K, VV, EV> {
-
- @Override
- public String getName() {
- return this.getClass().getSimpleName();
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/PathGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/PathGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/PathGraph.java
index aa0c18a..7312adf 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/PathGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/PathGraph.java
@@ -30,14 +30,14 @@ import static org.apache.flink.graph.generator.PathGraph.MINIMUM_VERTEX_COUNT;
* Generate a {@link org.apache.flink.graph.generator.PathGraph}.
*/
public class PathGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter vertexCount = new LongParameter(this, "vertex_count")
.setMinimumValue(MINIMUM_VERTEX_COUNT);
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexCount + ")";
+ return getName() + " (" + vertexCount + ")";
}
@Override
@@ -46,7 +46,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
return new org.apache.flink.graph.generator.PathGraph(env, vertexCount.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/RMatGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/RMatGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/RMatGraph.java
index f24b99f..0dac4f4 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/RMatGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/RMatGraph.java
@@ -39,7 +39,7 @@ import org.apache.commons.math3.random.JDKRandomGenerator;
* @see org.apache.flink.graph.generator.RMatGraph
*/
public class RMatGraph
-extends GeneratedMultiGraph<LongValue> {
+extends GeneratedMultiGraph {
// generate graph with 2^scale vertices
private LongParameter scale = new LongParameter(this, "scale")
@@ -79,8 +79,7 @@ extends GeneratedMultiGraph<LongValue> {
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() +
- " (s" + scale + "e" + edgeFactor + getSimplifyShortString() + ")";
+ return getName() + " (s" + scale + "e" + edgeFactor + getSimplifyShortString() + ")";
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/SingletonEdgeGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/SingletonEdgeGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/SingletonEdgeGraph.java
index b0c7463..80273c3 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/SingletonEdgeGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/SingletonEdgeGraph.java
@@ -30,7 +30,7 @@ import static org.apache.flink.graph.generator.PathGraph.MINIMUM_VERTEX_COUNT;
* Generate a {@link org.apache.flink.graph.generator.SingletonEdgeGraph}.
*/
public class SingletonEdgeGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter vertexPairCount = new LongParameter(this, "vertex_pair_count")
.setMinimumValue(MINIMUM_VERTEX_COUNT)
@@ -38,7 +38,7 @@ extends GeneratedGraph<LongValue> {
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexPairCount + ")";
+ return getName() + " (" + vertexPairCount + ")";
}
@Override
@@ -47,7 +47,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
return new org.apache.flink.graph.generator.SingletonEdgeGraph(env, vertexPairCount.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/StarGraph.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/StarGraph.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/StarGraph.java
index cde25ce..e954df5 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/StarGraph.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/input/StarGraph.java
@@ -30,14 +30,14 @@ import static org.apache.flink.graph.generator.StarGraph.MINIMUM_VERTEX_COUNT;
* Generate a {@link org.apache.flink.graph.generator.StarGraph}.
*/
public class StarGraph
-extends GeneratedGraph<LongValue> {
+extends GeneratedGraph {
private LongParameter vertexCount = new LongParameter(this, "vertex_count")
.setMinimumValue(MINIMUM_VERTEX_COUNT);
@Override
public String getIdentity() {
- return getTypeName() + " " + getName() + " (" + vertexCount + ")";
+ return getName() + " (" + vertexCount + ")";
}
@Override
@@ -46,7 +46,7 @@ extends GeneratedGraph<LongValue> {
}
@Override
- public Graph<LongValue, NullValue, NullValue> generate(ExecutionEnvironment env) {
+ public Graph<LongValue, NullValue, NullValue> create(ExecutionEnvironment env) {
return new org.apache.flink.graph.generator.StarGraph(env, vertexCount.getValue())
.setParallelism(parallelism.getValue().intValue())
.generate();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Hash.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Hash.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Hash.java
index a853339..54367cc 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Hash.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Hash.java
@@ -39,12 +39,14 @@ extends OutputBase<T> {
ChecksumHashCode<T> checksumHashCode = new ChecksumHashCode<T>().run(data);
if (printExecutionPlan.getValue()) {
- System.out.println(data.getExecutionEnvironment().getExecutionPlan());
+ out.println();
+ out.println(data.getExecutionEnvironment().getExecutionPlan());
}
ChecksumHashCode.Checksum checksum = checksumHashCode
.execute(executionName);
+ out.println();
out.println(checksum);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/OutputBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/OutputBase.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/OutputBase.java
index cbeff30..097f2c9 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/OutputBase.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/OutputBase.java
@@ -29,9 +29,4 @@ import org.apache.flink.graph.drivers.parameter.ParameterizedBase;
public abstract class OutputBase<T>
extends ParameterizedBase
implements Output<T> {
-
- @Override
- public String getName() {
- return this.getClass().getSimpleName();
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Print.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Print.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Print.java
index 623fed4..79f2afd 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Print.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/output/Print.java
@@ -41,7 +41,8 @@ extends OutputBase<T> {
Collect<T> collector = new Collect<T>().run(data);
if (printExecutionPlan.getValue()) {
- System.out.println(data.getExecutionEnvironment().getExecutionPlan());
+ out.println();
+ out.println(data.getExecutionEnvironment().getExecutionPlan());
}
List<T> results = collector.execute(executionName);
@@ -50,13 +51,15 @@ extends OutputBase<T> {
return;
}
+ out.println();
+
if (results.get(0) instanceof PrintableResult) {
for (Object result : results) {
- System.out.println(((PrintableResult) result).toPrintableString());
+ out.println(((PrintableResult) result).toPrintableString());
}
} else {
for (Object result : results) {
- System.out.println(result);
+ out.println(result);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ChoiceParameter.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ChoiceParameter.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ChoiceParameter.java
index 2c4544f..7280269 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ChoiceParameter.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ChoiceParameter.java
@@ -127,4 +127,9 @@ extends SimpleParameter<String> {
throw new ProgramParametrizationException(
"Selection '" + selected + "' for option '" + name + "' is not in choices '[" + StringUtils.join(choices, ", ") + "]'");
}
+
+ @Override
+ public String toString() {
+ return this.value;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ParameterizedBase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ParameterizedBase.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ParameterizedBase.java
index ba6f6af..6131677 100644
--- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ParameterizedBase.java
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/parameter/ParameterizedBase.java
@@ -46,6 +46,11 @@ implements Parameterized {
}
@Override
+ public String getName() {
+ return this.getClass().getSimpleName();
+ }
+
+ @Override
public String getUsage() {
StrBuilder strBuilder = new StrBuilder();
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransform.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransform.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransform.java
new file mode 100644
index 0000000..b98d820
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/GraphKeyTypeTransform.java
@@ -0,0 +1,841 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.drivers.transform;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.client.program.ProgramParametrizationException;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.asm.result.TranslatableResult;
+import org.apache.flink.graph.asm.translate.TranslateFunction;
+import org.apache.flink.graph.asm.translate.TranslateGraphIds;
+import org.apache.flink.graph.asm.translate.translators.LongValueToStringValue;
+import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue;
+import org.apache.flink.graph.drivers.parameter.BooleanParameter;
+import org.apache.flink.graph.drivers.parameter.ChoiceParameter;
+import org.apache.flink.graph.drivers.parameter.ParameterizedBase;
+import org.apache.flink.types.ByteValue;
+import org.apache.flink.types.CharValue;
+import org.apache.flink.types.DoubleValue;
+import org.apache.flink.types.FloatValue;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.ShortValue;
+import org.apache.flink.types.StringValue;
+import org.apache.flink.util.Collector;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+
+/**
+ * Transform the ID type of graphs from {@link LongValue} to byte, short,
+ * char, integer, float, double, or string. Both boxed Java types and Flink
+ * {@link org.apache.flink.types.Value} types are supported. The result
+ * is transformed when necessary to standardize {@code hashCode}.
+ */
+public class GraphKeyTypeTransform<VV, EV>
+extends ParameterizedBase
+implements Transform<Graph<LongValue, VV, EV>, Graph<?, VV, EV>, DataSet<TranslatableResult<?>>, DataSet<TranslatableResult<?>>> {
+
+ private static final String BYTE = "byte";
+ private static final String NATIVE_BYTE = "nativeByte";
+
+ private static final String SHORT = "short";
+ private static final String NATIVE_SHORT = "nativeShort";
+
+ private static final String CHAR = "char";
+ private static final String NATIVE_CHAR = "nativeChar";
+
+ private static final String INTEGER = "integer";
+ private static final String NATIVE_INTEGER = "nativeInteger";
+
+ private static final String LONG = "long";
+ private static final String NATIVE_LONG = "nativeLong";
+
+ private static final String FLOAT = "float";
+ private static final String NATIVE_FLOAT = "nativeFloat";
+
+ private static final String DOUBLE = "double";
+ private static final String NATIVE_DOUBLE = "nativeDouble";
+
+ private static final String STRING = "string";
+ private static final String NATIVE_STRING = "nativeString";
+
+ private ChoiceParameter type = new ChoiceParameter(this, "type")
+ .setDefaultValue(INTEGER)
+ .addChoices(LONG, STRING)
+ .addHiddenChoices(BYTE, NATIVE_BYTE, SHORT, NATIVE_SHORT, CHAR, NATIVE_CHAR, NATIVE_INTEGER, NATIVE_LONG,
+ FLOAT, NATIVE_FLOAT, DOUBLE, NATIVE_DOUBLE, NATIVE_STRING);
+
+ private final BooleanParameter disableTypeReversal = new BooleanParameter(this, "__disable_type_reversal");
+
+ private final long vertexCount;
+
+ /**
+ * Create a key type transform. The type is configured by the command-line
+ * arguments but because many types reduce the size of the vertex ID space
+ * the maximum extent must be set explicitly by the input in order to
+ * perform bounds checking. An n-bit type can store up to 2^n - 1 vertex
+ * identifiers.
+ *
+ * @param vertexCount the size of the vertex set of the input graph
+ */
+ public GraphKeyTypeTransform(long vertexCount) {
+ this.vertexCount = vertexCount;
+ }
+
+ @Override
+ public String getIdentity() {
+ return type.toString();
+ }
+
+ @Override
+ public Graph<?, VV, EV> transformInput(Graph<LongValue, VV, EV> input)
+ throws Exception {
+ // Long.MAX_VALUE is much larger than the number of atoms in the Earth
+ // and considered sufficient though representing 63 instead of 64 bits
+ long maxVertexCount = Long.MAX_VALUE;
+ TranslateFunction<LongValue, ?> translator = null;
+
+ switch (type.getValue()) {
+ case BYTE:
+ maxVertexCount = LongValueToUnsignedByteValue.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedByteValue();
+ break;
+
+ case NATIVE_BYTE:
+ maxVertexCount = LongValueToUnsignedByte.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedByte();
+ break;
+
+ case SHORT:
+ maxVertexCount = LongValueToUnsignedShortValue.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedShortValue();
+ break;
+
+ case NATIVE_SHORT:
+ maxVertexCount = LongValueToUnsignedShort.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedShort();
+ break;
+
+ case CHAR:
+ maxVertexCount = LongValueToCharValue.MAX_VERTEX_COUNT;
+ translator = new LongValueToCharValue();
+ break;
+
+ case NATIVE_CHAR:
+ maxVertexCount = LongValueToChar.MAX_VERTEX_COUNT;
+ translator = new LongValueToChar();
+ break;
+
+ case INTEGER:
+ maxVertexCount = LongValueToUnsignedIntValue.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedIntValue();
+ break;
+
+ case NATIVE_INTEGER:
+ maxVertexCount = LongValueToUnsignedInt.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedInt();
+ break;
+
+ case LONG:
+ break;
+
+ case NATIVE_LONG:
+ translator = new LongValueToLong();
+ break;
+
+ case FLOAT:
+ maxVertexCount = LongValueToUnsignedFloatValue.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedFloatValue();
+ break;
+
+ case NATIVE_FLOAT:
+ maxVertexCount = LongValueToUnsignedFloat.MAX_VERTEX_COUNT;
+ translator = new LongValueToUnsignedFloat();
+ break;
+
+ case DOUBLE:
+ translator = new LongValueToDoubleValue();
+ break;
+
+ case NATIVE_DOUBLE:
+ translator = new LongValueToDouble();
+ break;
+
+ case STRING:
+ translator = new LongValueToStringValue();
+ break;
+
+ case NATIVE_STRING:
+ translator = new LongValueToString();
+ break;
+
+ default:
+ throw new ProgramParametrizationException("Unknown type '" + type.getValue() + "'");
+ }
+
+ if (vertexCount > maxVertexCount) {
+ throw new ProgramParametrizationException("Vertex count '" + vertexCount +
+ "' must be no greater than " + maxVertexCount + " for type '" + type.getValue() + "'.");
+ }
+
+ if (translator == null) {
+ return input;
+ } else {
+ return (Graph<?, VV, EV>) input.run(new TranslateGraphIds(translator));
+ }
+ }
+
+ @Override
+ public DataSet<TranslatableResult<?>> transformResult(final DataSet<TranslatableResult<?>> result) throws Exception {
+ if (disableTypeReversal.getValue()) {
+ return result;
+ }
+
+ // the minimum vertex count is used for signed types for which results
+ // must be translated to a wider type when overflowing to negative
+ // values in order to preserve consistent checksums
+ long minVertexCount = 0;
+ TranslateFunction<?, LongValueWithProperHashCode> translator = null;
+
+ switch (type.getValue()) {
+ case BYTE:
+ minVertexCount = UnsignedByteValueToLongValueWithProperHashCode.MIN_VERTEX_COUNT;
+ translator = new UnsignedByteValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_BYTE:
+ minVertexCount = UnsignedByteToLongValueWithProperHashCode.MIN_VERTEX_COUNT;
+ translator = new UnsignedByteToLongValueWithProperHashCode();
+ break;
+
+ case SHORT:
+ minVertexCount = UnsignedShortValueToLongValueWithProperHashCode.MIN_VERTEX_COUNT;
+ translator = new UnsignedShortValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_SHORT:
+ minVertexCount = UnsignedShortToLongValueWithProperHashCode.MIN_VERTEX_COUNT;
+ translator = new UnsignedShortToLongValueWithProperHashCode();
+ break;
+
+ case CHAR:
+ break;
+
+ case NATIVE_CHAR:
+ break;
+
+ case INTEGER:
+ minVertexCount = UnsignedIntValueToLongValueWithProperHashCode.MIN_VERTEX_COUNT;
+ translator = new UnsignedIntValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_INTEGER:
+ minVertexCount = UnsignedIntToLongValueWithProperHashCode.MIN_VERTEX_COUNT;
+ translator = new UnsignedIntToLongValueWithProperHashCode();
+ break;
+
+ case LONG:
+ translator = new LongValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_LONG:
+ break;
+
+ case FLOAT:
+ translator = new UnsignedFloatValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_FLOAT:
+ translator = new UnsignedFloatToLongValueWithProperHashCode();
+ break;
+
+ case DOUBLE:
+ translator = new DoubleValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_DOUBLE:
+ translator = new DoubleToLongValueWithProperHashCode();
+ break;
+
+ case STRING:
+ translator = new StringValueToLongValueWithProperHashCode();
+ break;
+
+ case NATIVE_STRING:
+ translator = new StringToLongValueWithProperHashCode();
+ break;
+
+ default:
+ throw new ProgramParametrizationException("Unknown type '" + type.getValue() + "'");
+ }
+
+ if (translator == null || vertexCount < minVertexCount) {
+ return result;
+ }
+
+ TypeInformation<?> typeInfo = TypeExtractor.createTypeInfo(new ParameterizedType() {
+ @Override
+ public Type[] getActualTypeArguments() {
+ return new Type[]{LongValueWithProperHashCode.class};
+ }
+
+ @Override
+ public Type getRawType() {
+ return result.getType().getTypeClass();
+ }
+
+ @Override
+ public Type getOwnerType() {
+ return null;
+ }
+ });
+
+ return result
+ .flatMap(new TranslateResultIDs(translator))
+ .returns(typeInfo)
+ .name("Translate results IDs");
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link ByteValue}.
+ *
+ * <p>Throws {@link RuntimeException} for byte overflow.
+ */
+ static class LongValueToUnsignedByteValue
+ implements TranslateFunction<LongValue, ByteValue> {
+ public static final long MAX_VERTEX_COUNT = 1L << 8;
+
+ @Override
+ public ByteValue translate(LongValue value, ByteValue reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new ByteValue();
+ }
+
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to byte.");
+ }
+
+ reuse.setValue((byte) (l & (MAX_VERTEX_COUNT - 1)));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link Byte}.
+ *
+ * <p>Throws {@link RuntimeException} for byte overflow.
+ */
+ static class LongValueToUnsignedByte
+ implements TranslateFunction<LongValue, Byte> {
+ public static final long MAX_VERTEX_COUNT = 1L << 8;
+
+ @Override
+ public Byte translate(LongValue value, Byte reuse)
+ throws Exception {
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to byte.");
+ }
+
+ return (byte) (l & (MAX_VERTEX_COUNT - 1));
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link ShortValue}.
+ *
+ * <p>Throws {@link RuntimeException} for short overflow.
+ */
+ static class LongValueToUnsignedShortValue
+ implements TranslateFunction<LongValue, ShortValue> {
+ public static final long MAX_VERTEX_COUNT = 1L << 16;
+
+ @Override
+ public ShortValue translate(LongValue value, ShortValue reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new ShortValue();
+ }
+
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to short.");
+ }
+
+ reuse.setValue((short) (l & (MAX_VERTEX_COUNT - 1)));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link Short}.
+ *
+ * <p>Throws {@link RuntimeException} for short overflow.
+ */
+ static class LongValueToUnsignedShort
+ implements TranslateFunction<LongValue, Short> {
+ public static final long MAX_VERTEX_COUNT = 1L << 16;
+
+ @Override
+ public Short translate(LongValue value, Short reuse)
+ throws Exception {
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to short.");
+ }
+
+ return (short) (l & (MAX_VERTEX_COUNT - 1));
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to {@link CharValue}.
+ *
+ * <p>Throws {@link RuntimeException} for char overflow.
+ */
+ static class LongValueToCharValue
+ implements TranslateFunction<LongValue, CharValue> {
+ public static final long MAX_VERTEX_COUNT = 1L << 16;
+
+ @Override
+ public CharValue translate(LongValue value, CharValue reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new CharValue();
+ }
+
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to char.");
+ }
+
+ reuse.setValue((char) (l & (MAX_VERTEX_COUNT - 1)));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to {@code Character}.
+ *
+ * <p>Throws {@link RuntimeException} for char overflow.
+ */
+ static class LongValueToChar
+ implements TranslateFunction<LongValue, Character> {
+ public static final long MAX_VERTEX_COUNT = 1L << 16;
+
+ @Override
+ public Character translate(LongValue value, Character reuse)
+ throws Exception {
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to char.");
+ }
+
+ return (char) (l & (MAX_VERTEX_COUNT - 1));
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link Integer}.
+ *
+ * <p>Throws {@link RuntimeException} for integer overflow.
+ */
+ static class LongValueToUnsignedInt
+ implements TranslateFunction<LongValue, Integer> {
+ public static final long MAX_VERTEX_COUNT = 1L << 32;
+
+ @Override
+ public Integer translate(LongValue value, Integer reuse)
+ throws Exception {
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to integer.");
+ }
+
+ return (int) (l & (MAX_VERTEX_COUNT - 1));
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to {@link Long}.
+ */
+ static class LongValueToLong
+ implements TranslateFunction<LongValue, Long> {
+ @Override
+ public Long translate(LongValue value, Long reuse)
+ throws Exception {
+ return value.getValue();
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link FloatValue} bit string.
+ */
+ static class LongValueToUnsignedFloatValue
+ implements TranslateFunction<LongValue, FloatValue> {
+ public static final long MAX_VERTEX_COUNT = 1L << 32;
+
+ @Override
+ public FloatValue translate(LongValue value, FloatValue reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new FloatValue();
+ }
+
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to float.");
+ }
+
+ reuse.setValue(Float.intBitsToFloat((int) (l & (MAX_VERTEX_COUNT - 1))));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to an unsigned {@link Float} bit string.
+ */
+ static class LongValueToUnsignedFloat
+ implements TranslateFunction<LongValue, Float> {
+ public static final long MAX_VERTEX_COUNT = 1L << 32;
+
+ @Override
+ public Float translate(LongValue value, Float reuse)
+ throws Exception {
+ long l = value.getValue();
+
+ if (l < 0 || l >= MAX_VERTEX_COUNT) {
+ throw new IllegalArgumentException("Cannot cast long value " + value + " to float.");
+ }
+
+ return Float.intBitsToFloat((int) (l & (MAX_VERTEX_COUNT - 1)));
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to a {@link DoubleValue} bit string.
+ */
+ static class LongValueToDoubleValue
+ implements TranslateFunction<LongValue, DoubleValue> {
+ @Override
+ public DoubleValue translate(LongValue value, DoubleValue reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new DoubleValue();
+ }
+
+ reuse.setValue(Double.longBitsToDouble(value.getValue()));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to a {@link Double} bit string.
+ */
+ static class LongValueToDouble
+ implements TranslateFunction<LongValue, Double> {
+ @Override
+ public Double translate(LongValue value, Double reuse)
+ throws Exception {
+ return Double.longBitsToDouble(value.getValue());
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to {@link String}.
+ */
+ static class LongValueToString
+ implements TranslateFunction<LongValue, String> {
+ @Override
+ public String translate(LongValue value, String reuse)
+ throws Exception {
+ return Long.toString(value.getValue());
+ }
+ }
+
+ /**
+ * Translate a {@link TranslatableResult}.
+ *
+ * @param <T> ID type to translate from
+ * @param <U> ID type to translate to
+ */
+ private static class TranslateResultIDs<T, U>
+ implements FlatMapFunction<TranslatableResult<T>, TranslatableResult<U>> {
+ private final TranslateFunction<T, U> translator;
+
+ private transient TranslatableResult<U> reuse = null;
+
+ public TranslateResultIDs(TranslateFunction<T, U> translator) {
+ this.translator = translator;
+ }
+
+ @Override
+ public void flatMap(TranslatableResult<T> value, Collector<TranslatableResult<U>> out)
+ throws Exception {
+ reuse = value.translate(translator, reuse, out);
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link ByteValue} to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedByteValueToLongValueWithProperHashCode
+ implements TranslateFunction<ByteValue, LongValueWithProperHashCode> {
+ public static final long MIN_VERTEX_COUNT = Byte.MAX_VALUE + 2;
+
+ @Override
+ public LongValueWithProperHashCode translate(ByteValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(value.getValue() & 0xff);
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link Byte} to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedByteToLongValueWithProperHashCode
+ implements TranslateFunction<Byte, LongValueWithProperHashCode> {
+ public static final long MIN_VERTEX_COUNT = Byte.MAX_VALUE + 2;
+
+ @Override
+ public LongValueWithProperHashCode translate(Byte value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(value & 0xff);
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link ShortValue} to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedShortValueToLongValueWithProperHashCode
+ implements TranslateFunction<ShortValue, LongValueWithProperHashCode> {
+ public static final long MIN_VERTEX_COUNT = Short.MAX_VALUE + 2;
+
+ @Override
+ public LongValueWithProperHashCode translate(ShortValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(value.getValue() & 0xffff);
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link Short} to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedShortToLongValueWithProperHashCode
+ implements TranslateFunction<Short, LongValueWithProperHashCode> {
+ public static final long MIN_VERTEX_COUNT = Short.MAX_VALUE + 2;
+
+ @Override
+ public LongValueWithProperHashCode translate(Short value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(value & 0xffff);
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link IntValue} to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedIntValueToLongValueWithProperHashCode
+ implements TranslateFunction<IntValue, LongValueWithProperHashCode> {
+ public static final long MIN_VERTEX_COUNT = Integer.MAX_VALUE + 2L;
+
+ @Override
+ public LongValueWithProperHashCode translate(IntValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(0xffffffffL & (long) value.getValue());
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link Integer} to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedIntToLongValueWithProperHashCode
+ implements TranslateFunction<Integer, LongValueWithProperHashCode> {
+ public static final long MIN_VERTEX_COUNT = Integer.MAX_VALUE + 2L;
+
+ @Override
+ public LongValueWithProperHashCode translate(Integer value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(0xffffffffL & (long) value);
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link LongValue} to {@link LongValueWithProperHashCode}.
+ */
+ static class LongValueToLongValueWithProperHashCode
+ implements TranslateFunction<LongValue, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(LongValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(value.getValue());
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link FloatValue} bit string to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedFloatValueToLongValueWithProperHashCode
+ implements TranslateFunction<FloatValue, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(FloatValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(0xffffffffL & Float.floatToRawIntBits(value.getValue()));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate an unsigned {@link Float} bit string to {@link LongValueWithProperHashCode}.
+ */
+ static class UnsignedFloatToLongValueWithProperHashCode
+ implements TranslateFunction<Float, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(Float value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(0xffffffffL & Float.floatToRawIntBits(value));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate a {@link DoubleValue} bit string to {@link LongValueWithProperHashCode}.
+ */
+ static class DoubleValueToLongValueWithProperHashCode
+ implements TranslateFunction<DoubleValue, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(DoubleValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(Double.doubleToRawLongBits(value.getValue()));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate a {@link Double} bit string to {@link LongValueWithProperHashCode}.
+ */
+ static class DoubleToLongValueWithProperHashCode
+ implements TranslateFunction<Double, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(Double value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(Double.doubleToRawLongBits(value));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link StringValue} to {@link LongValueWithProperHashCode}.
+ */
+ static class StringValueToLongValueWithProperHashCode
+ implements TranslateFunction<StringValue, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(StringValue value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(Long.parseLong(value.getValue()));
+ return reuse;
+ }
+ }
+
+ /**
+ * Translate {@link String} to {@link LongValueWithProperHashCode}.
+ */
+ static class StringToLongValueWithProperHashCode
+ implements TranslateFunction<String, LongValueWithProperHashCode> {
+ @Override
+ public LongValueWithProperHashCode translate(String value, LongValueWithProperHashCode reuse)
+ throws Exception {
+ if (reuse == null) {
+ reuse = new LongValueWithProperHashCode();
+ }
+
+ reuse.setValue(Long.valueOf(value));
+ return reuse;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8695a210/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java
new file mode 100644
index 0000000..22a681c
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/transform/LongValueWithProperHashCode.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.flink.graph.drivers.transform;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.AtomicType;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.LongValueComparator;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.types.LongValue;
+
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Fix for {@link LongValue#hashCode()}.
+ */
+@TypeInfo(LongValueWithProperHashCode.LongValueWithProperHashCodeTypeInfoFactory.class)
+public class LongValueWithProperHashCode
+extends LongValue {
+
+ @Override
+ public int hashCode() {
+ return (int) (this.getValue() ^ this.getValue() >>> 32);
+ }
+
+ /**
+ * Initializes the encapsulated long with 0.
+ */
+ public LongValueWithProperHashCode() {
+ super();
+ }
+
+ /**
+ * Initializes the encapsulated long with the specified value.
+ *
+ * @param value Initial value of the encapsulated long.
+ */
+ public LongValueWithProperHashCode(final long value) {
+ super(value);
+ }
+
+ /**
+ * Used by {@link TypeExtractor} to create a {@link TypeInformation} for
+ * {@link LongValueWithProperHashCode}.
+ */
+ public static class LongValueWithProperHashCodeTypeInfoFactory extends TypeInfoFactory<LongValueWithProperHashCode> {
+ @Override
+ public TypeInformation<LongValueWithProperHashCode> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParameters) {
+ @SuppressWarnings("unchecked")
+ TypeInformation<LongValueWithProperHashCode> typeInfo = new LongValueWithProperHashCodeTypeInfo();
+
+ return typeInfo;
+ }
+ }
+
+ /**
+ * A {@link TypeInformation} for the {@link LongValueWithProperHashCode} type.
+ *
+ * @see org.apache.flink.api.java.typeutils.ValueTypeInfo#LONG_VALUE_TYPE_INFO
+ */
+ public static class LongValueWithProperHashCodeTypeInfo extends TypeInformation<LongValueWithProperHashCode> implements AtomicType<LongValueWithProperHashCode> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public int getArity() {
+ return 1;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 1;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Class<LongValueWithProperHashCode> getTypeClass() {
+ return LongValueWithProperHashCode.class;
+ }
+
+ @Override
+ public boolean isBasicType() {
+ return false;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return false;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return Comparable.class.isAssignableFrom(LongValueWithProperHashCode.class);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public TypeSerializer<LongValueWithProperHashCode> createSerializer(ExecutionConfig executionConfig) {
+ return new LongValueWithProperHashCodeSerializer();
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Override
+ public TypeComparator<LongValueWithProperHashCode> createComparator(boolean sortOrderAscending, ExecutionConfig executionConfig) {
+ return (TypeComparator<LongValueWithProperHashCode>) (TypeComparator<?>) new LongValueComparator(sortOrderAscending);
+ }
+
+ @Override
+ public Map<String, TypeInformation<?>> getGenericParameters() {
+ return Collections.emptyMap();
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof LongValueWithProperHashCodeTypeInfo;
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof LongValueWithProperHashCodeTypeInfo;
+ }
+
+ @Override
+ public String toString() {
+ return LongValueWithProperHashCodeTypeInfo.class.getSimpleName();
+ }
+ }
+
+ /**
+ * Serializer for {@link LongValueWithProperHashCode}.
+ *
+ * @see org.apache.flink.api.common.typeutils.base.LongValueSerializer
+ */
+ public static final class LongValueWithProperHashCodeSerializer extends TypeSerializerSingleton<LongValueWithProperHashCode> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public LongValueWithProperHashCode createInstance() {
+ return new LongValueWithProperHashCode();
+ }
+
+ @Override
+ public LongValueWithProperHashCode copy(LongValueWithProperHashCode from) {
+ return copy(from, new LongValueWithProperHashCode());
+ }
+
+ @Override
+ public LongValueWithProperHashCode copy(LongValueWithProperHashCode from, LongValueWithProperHashCode reuse) {
+ reuse.setValue(from.getValue());
+ return reuse;
+ }
+
+ @Override
+ public int getLength() {
+ return 8;
+ }
+
+ @Override
+ public void serialize(LongValueWithProperHashCode record, DataOutputView target) throws IOException {
+ record.write(target);
+ }
+
+ @Override
+ public LongValueWithProperHashCode deserialize(DataInputView source) throws IOException {
+ return deserialize(new LongValueWithProperHashCode(), source);
+ }
+
+ @Override
+ public LongValueWithProperHashCode deserialize(LongValueWithProperHashCode reuse, DataInputView source) throws IOException {
+ reuse.read(source);
+ return reuse;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ target.writeLong(source.readLong());
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof LongValueWithProperHashCodeSerializer;
+ }
+
+ @Override
+ protected boolean isCompatibleSerializationFormatIdentifier(String identifier) {
+ return super.isCompatibleSerializationFormatIdentifier(identifier)
+ || identifier.equals(LongSerializer.class.getCanonicalName());
+ }
+ }
+}
[2/5] flink git commit: [FLINK-7234] [docs] Fix CombineHint
documentation
Posted by gr...@apache.org.
[FLINK-7234] [docs] Fix CombineHint documentation
The CombineHint documentation applies to DataSet#reduce not
DataSet#reduceGroup and should also be noted for DataSet#distinct. Also
correct the usage where the CombineHint is set with setCombineHint
rather than alongside the user-defined function parameter.
This closes #4372
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4a88f658
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4a88f658
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4a88f658
Branch: refs/heads/master
Commit: 4a88f6587fdfadd5749188a76e6b38a3585cd31b
Parents: 8695a21
Author: Greg Hogan <co...@greghogan.com>
Authored: Wed Jul 19 15:24:20 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed Jul 26 10:31:20 2017 -0400
----------------------------------------------------------------------
docs/dev/batch/index.md | 24 ++++++++++++--------
.../operators/base/ReduceOperatorBase.java | 3 +--
2 files changed, 16 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4a88f658/docs/dev/batch/index.md
----------------------------------------------------------------------
diff --git a/docs/dev/batch/index.md b/docs/dev/batch/index.md
index 9a8ce22..7fb84e8 100644
--- a/docs/dev/batch/index.md
+++ b/docs/dev/batch/index.md
@@ -205,12 +205,17 @@ data.filter(new FilterFunction<Integer>() {
<td><strong>Reduce</strong></td>
<td>
<p>Combines a group of elements into a single element by repeatedly combining two elements
- into one. Reduce may be applied on a full data set, or on a grouped data set.</p>
+ into one. Reduce may be applied on a full data set or on a grouped data set.</p>
{% highlight java %}
data.reduce(new ReduceFunction<Integer> {
public Integer reduce(Integer a, Integer b) { return a + b; }
});
{% endhighlight %}
+ <p>If the reduce was applied to a grouped data set then you can specify the way that the
+ runtime executes the combine phase of the reduce by supplying a <code>CombineHint</code> to
+ <code>setCombineHint</code>. The hash-based strategy should be faster in most cases,
+ especially if the number of different keys is small compared to the number of input
+ elements (eg. 1/10).</p>
</td>
</tr>
@@ -218,7 +223,7 @@ data.reduce(new ReduceFunction<Integer> {
<td><strong>ReduceGroup</strong></td>
<td>
<p>Combines a group of elements into one or more elements. ReduceGroup may be applied on a
- full data set, or on a grouped data set.</p>
+ full data set or on a grouped data set.</p>
{% highlight java %}
data.reduceGroup(new GroupReduceFunction<Integer, Integer> {
public void reduce(Iterable<Integer> values, Collector<Integer> out) {
@@ -230,10 +235,6 @@ data.reduceGroup(new GroupReduceFunction<Integer, Integer> {
}
});
{% endhighlight %}
- <p>If the reduce was applied to a grouped data set, you can specify the way that the
- runtime executes the combine phase of the reduce via supplying a CombineHint as a second
- parameter. The hash-based strategy should be faster in most cases, especially if the
- number of different keys is small compared to the number of input elements (eg. 1/10).</p>
</td>
</tr>
@@ -260,9 +261,14 @@ DataSet<Tuple3<Integer, String, Double>> output = input.sum(0).andMin(2);
<td>
<p>Returns the distinct elements of a data set. It removes the duplicate entries
from the input DataSet, with respect to all fields of the elements, or a subset of fields.</p>
- {% highlight java %}
- data.distinct();
- {% endhighlight %}
+{% highlight java %}
+data.distinct();
+{% endhighlight %}
+ <p>Distinct is implemented using a reduce function. You can specify the way that the
+ runtime executes the combine phase of the reduce by supplying a <code>CombineHint</code> to
+ <code>setCombineHint</code>. The hash-based strategy should be faster in most cases,
+ especially if the number of different keys is small compared to the number of input
+ elements (eg. 1/10).</p>
</td>
</tr>
http://git-wip-us.apache.org/repos/asf/flink/blob/4a88f658/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
index f97e4d6..f6a7a59 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java
@@ -79,8 +79,7 @@ public class ReduceOperatorBase<T, FT extends ReduceFunction<T>> extends SingleI
HASH,
/**
- * Disable the use of a combiner. This can be faster in cases when the number of different keys
- * is very small compared to the number of input elements (eg. 1/100).
+ * Disable the use of a combiner.
*/
NONE
}