content: /*
   * 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 com.aliyun.odps.examples.graph;
  
  import java.io.IOException;
  import java.util.List;
  
  import org.junit.Assert;
  import org.junit.Test;
  
  import com.aliyun.odps.data.TableInfo;
  import com.aliyun.odps.graph.ComputeContext;
  import com.aliyun.odps.graph.Edge;
  import com.aliyun.odps.graph.GraphJob;
  import com.aliyun.odps.graph.GraphLoader;
  import com.aliyun.odps.graph.MutationContext;
  import com.aliyun.odps.graph.Vertex;
  import com.aliyun.odps.graph.WorkerContext;
  import com.aliyun.odps.io.LongWritable;
  import com.aliyun.odps.io.NullWritable;
  import com.aliyun.odps.io.Tuple;
  import com.aliyun.odps.io.Writable;
  import com.aliyun.odps.io.WritableRecord;
  import com.aliyun.odps.local.common.WareHouse;
  
  /**
   * Compute the number of triangles passing through each vertex.
   *
   * The algorithm can be computed in three supersteps:
   * I. Each vertex sends a message with its ID to all its outgoing
   * neighbors.
   * II. The incoming neighbors and outgoing neighbors are stored and
   * send to outgoing neighbors.
   * III. For each edge compute the intersection of the sets at destination
   * vertex and sum them, then output to table.
   *
   * The triangle count is the sum of output table and divide by three since
   * each triangle is counted three times.
   */
  public class TriangleCount {
  
    public static class TCVertex extends
                                 Vertex<LongWritable, Tuple, NullWritable, Tuple> {
  
      @Override
      public void setup(
          WorkerContext<LongWritable, Tuple, NullWritable, Tuple> context)
          throws IOException {
        // collect the outgoing neighbors
        Tuple t = new Tuple();
        if (this.hasEdges()) {
          for (Edge<LongWritable, NullWritable> edge : this.getEdges()) {
            t.append(edge.getDestVertexId());
          }
        }
  
        this.setValue(t);
      }
  
      @Override
      public void compute(
          ComputeContext<LongWritable, Tuple, NullWritable, Tuple> context,
          Iterable<Tuple> msgs) throws IOException {
  
        if (context.getSuperstep() == 0L) {
          // sends a message with its ID to all its outgoing neighbors
          Tuple t = new Tuple();
          t.append(getId());
          context.sendMessageToNeighbors(this, t);
        } else if (context.getSuperstep() == 1L) {
          // store the incoming neighbors
          for (Tuple msg : msgs) {
            for (Writable item : msg.getAll()) {
              if (!this.getValue().getAll().contains((LongWritable) item)) {
                this.getValue().append((LongWritable) item);
              }
            }
          }
  
          // send both incoming and outgoing neighbors to all outgoing neighbors
          context.sendMessageToNeighbors(this, getValue());
        } else if (context.getSuperstep() == 2L) {
          // count the sum of intersection at each edge
          long count = 0;
          for (Tuple msg : msgs) {
            for (Writable id : msg.getAll()) {
              if (getValue().getAll().contains(id)) {
                count++;
              }
            }
          }
  
          // output to table
          context.write(getId(), new LongWritable(count));
          this.voteToHalt();
        }
      }
    }
  
  
    public static class TCVertexReader extends
                                       GraphLoader<LongWritable, Tuple, NullWritable, Tuple> {
  
      @Override
      public void load(
          LongWritable recordNum,
          WritableRecord record,
          MutationContext<LongWritable, Tuple, NullWritable, Tuple> context)
          throws IOException {
        TCVertex vertex = new TCVertex();
  
        vertex.setId((LongWritable) record.get(0));
  
        String[] edges = record.get(1).toString().split(";");
        for (int i = 0; i < edges.length; i++) {
          try {
            long destID = Long.parseLong(edges[i]);
            vertex.addEdge(new LongWritable(destID), NullWritable.get());
          } catch (NumberFormatException nfe) {
            System.err.println("Ignore " + nfe);
          }
        }
        context.addVertexRequest(vertex);
      }
    }
  
    public static void main(String[] args) throws IOException {
      if (args.length < 2) {
        System.out.println("Usage: <input> <output>");
        System.exit(-1);
      }
  
      GraphJob job = new GraphJob();
      job.setGraphLoaderClass(TCVertexReader.class);
      job.setVertexClass(TCVertex.class);
  
      job.addInput(TableInfo.builder().tableName(args[0]).build());
      job.addOutput(TableInfo.builder().tableName(args[1]).build());
      long startTime = System.currentTimeMillis();
      job.run();
      System.out.println("Job Finished in "
                         + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
    }
  
    @Test
    public void test() throws Exception {
      WareHouse wareHouse = WareHouse.getInstance();
      String project = TestUtils.yzy2;
      String outputTable = "triangle_count_out";
  
      TestUtils.setEnvironment(project);
  
      //If output table exists then delete data (will not delete schema)
      wareHouse.dropTableDataIfExists(project, outputTable, null);
      Assert.assertEquals(true, wareHouse.isTableEmpty(project, outputTable, null));
  
      new TriangleCount().main(new String[]{"connected_components_in", outputTable});
  
      Assert.assertEquals(false, wareHouse.isTableEmpty(project, outputTable, null));
  
      //read output table data
      List<Object[]> result = wareHouse.readData(project, outputTable, null, null, ',');
      Assert.assertEquals(11, result.size());
  
      // Sampling inspection
      Object[] record = result.get(2);
      Assert.assertEquals(2, record.length);
      Assert.assertEquals(true, record[0] instanceof Long);
      Assert.assertEquals(true, record[1] instanceof Long);
      Assert.assertEquals(3L, record[0]);
      Assert.assertEquals(0L, record[1]);
  
    }
  }
  