Upgraded to Cassovary 7.1.0

graphjet-adapters/pom.xml

@@ -30,8 +30,8 @@
     </dependency>
     <dependency>
       <groupId>com.twitter</groupId>
-      <artifactId>cassovary-core_2.10</artifactId>
-      <version>6.4.0</version>
+      <artifactId>cassovary-core_2.11</artifactId>
+      <version>7.1.0</version>
     </dependency>
     <dependency>
       <!-- Cassovary depends on v0.0.37, which isn't found on Maven central. Forcing newer version.

graphjet-adapters/src/main/java/com/twitter/graphjet/adapter/cassovary/CassovaryOutIndexedDirectedGraph.java

@@ -16,33 +16,33 @@
 
 package com.twitter.graphjet.adapter.cassovary;
 
-import java.util.Random;
-
 import com.twitter.cassovary.graph.DirectedGraph;
-import com.twitter.cassovary.graph.Node;
 import com.twitter.cassovary.graph.GraphDir;
-import com.twitter.graphjet.directed.api.OutIndexedDirectedGraph;
+import com.twitter.cassovary.graph.Node;
 import com.twitter.graphjet.bipartite.api.EdgeIterator;
-
-import scala.collection.Seq;
+import com.twitter.graphjet.directed.api.OutIndexedDirectedGraph;
 import scala.Option;
+import scala.collection.Seq;
+
+import java.util.Random;
 
 /**
- * A GraphJet wrapper for an out-indexed Cassovary graph. Implements the GraphJet API by delegating methods to the 
- * underlying Cassovary API.
+ * A GraphJet wrapper for an out-indexed Cassovary graph. Implements the GraphJet API by
+ * delegating methods to the underlying Cassovary API.
  */
 public class CassovaryOutIndexedDirectedGraph implements OutIndexedDirectedGraph {
   // This is the Cassovary graph that's being wrapped.
   final private DirectedGraph<Node> graph;
 
   /**
-   * Constructs a GraphJet wrapper for an out-indexed Cassovart graph.
+   * Constructs a GraphJet wrapper for an out-indexed Cassovary graph.
    *
    * @param graph the Cassovary graph
    */
   public CassovaryOutIndexedDirectedGraph(DirectedGraph<Node> graph) {
     if (!graph.isDirStored(GraphDir.OutDir()) || graph.isBiDirectional()) {
-      // If the graph isn't out-indexed or if the graph is bidirectional, we should be using a different wrapper class.
+      // If the graph isn't out-indexed or if the graph is bidirectional, we should be using a
+      // different wrapper class.
       throw new IncompatibleCassovaryGraphException();
     }
     this.graph = graph;
@@ -60,7 +60,9 @@ public EdgeIterator getOutEdges(long node) {
       return new EmptyEdgeIterator();
     }
 
-    return new SeqEdgeIteratorWrapper(opt.get().outboundNodes());
+    // Note that outboundNodes returns a CSeq, whereas randomOutboundNodeSet returns a Seq, so we
+    // need different wrapper classes.
+    return new CSeqEdgeIteratorWrapper(opt.get().outboundNodes());
   }
 
   @Override
@@ -70,6 +72,8 @@ public EdgeIterator getRandomOutEdges(long node, int numSamples, Random random)
       return new EmptyEdgeIterator();
     }
 
+    // Note that randomOutboundNodeSet returns a Seq, whereas outboundNodes returns a CSeq, so we
+    // need different wrapper classes.
     return new SeqEdgeIteratorWrapper(opt.get().randomOutboundNodeSet(
         numSamples, scala.util.Random.javaRandomToRandom(random)));
   }
@@ -107,6 +111,52 @@ public Long next() {
       return (Long) seq.apply(index++);
     }
 
+    @Override
+    public int skip(int n) {
+      if (index + n < seq.length()) {
+        index += n;
+        return n;
+      }
+
+      int skipped = seq.length() - index;
+      index = seq.length();
+      return skipped;
+    }
+  }
+
+  /**
+   * Wrapper for CSeq as an EdgeIterator.
+   */
+  private class CSeqEdgeIteratorWrapper implements EdgeIterator {
+    final private com.twitter.cassovary.collections.CSeq seq;
+    private int index = 0;
+
+    public CSeqEdgeIteratorWrapper(com.twitter.cassovary.collections.CSeq seq) {
+      this.seq = seq;
+      index = 0;
+    }
+
+    @Override
+    public byte currentEdgeType() {
+      // Always return 0 since Cassovary edges aren't typed.
+      return 0;
+    }
+
+    @Override
+    public boolean hasNext() {
+      return index < seq.length();
+    }
+
+    @Override
+    public long nextLong() {
+      return (long) (int) seq.apply(index++);
+    }
+
+    @Override
+    public Long next() {
+      return (Long) seq.apply(index++);
+    }
+
     @Override
     public int skip(int n) {
       if ( index+n < seq.length()) {

graphjet-demo/src/main/java/com/twitter/graphjet/demo/PageRankCassovaryDemo.java

@@ -72,7 +72,7 @@ public static void main(String[] argv) throws Exception {
     }
 
     DirectedGraph<Node> cgraph = ListOfEdgesGraphReader.forIntIds(args.inputDir, args.inputFile,
-        new NodeNumberer.IntIdentity(), false, false, '\t', StoredGraphDir.OnlyOut())
+        new NodeNumberer.IntIdentity(), false, false, '\t', StoredGraphDir.OnlyOut(), true)
           .toSharedArrayBasedDirectedGraph(scala.Option.apply(null));
 
     // Wrap the Cassovary graph.