Creates the operator that represents this vertex-centric graph computation.

@return The operator that represents this vertex-centric graph computation.
184570
@Override
	public DataSet<Tuple2<VertexKey, VertexValue>> createResult() {
		if (this.initialVertices == null) {
			throw new IllegalStateException("The input data set has not been set.");
		}
		
		// prepare some type information
		TypeInformation<Tuple2<VertexKey, VertexValue>> vertexTypes = initialVertices.getType();
		TypeInformation<VertexKey> keyType = ((TupleTypeInfo<?>) initialVertices.getType()).getTypeAt(0);
		TypeInformation<Tuple2<VertexKey, Message>> messageTypeInfo = new TupleTypeInfo<Tuple2<VertexKey,Message>>(keyType, messageType);		
		
		// set up the iteration operator
		final String name = (this.name != null) ? this.name :
			"Vertex-centric iteration (" + updateFunction + " | " + messagingFunction + ")";
		final int[] zeroKeyPos = new int[] {0};
	
		final DeltaIteration<Tuple2<VertexKey, VertexValue>, Tuple2<VertexKey, VertexValue>> iteration =
			this.initialVertices.iterateDelta(this.initialVertices, this.maximumNumberOfIterations, zeroKeyPos);
		iteration.name(name);
		iteration.parallelism(parallelism);
		
		// register all aggregators
		for (Map.Entry<String, Class<? extends Aggregator<?>>> entry : this.aggregators.entrySet()) {
			iteration.registerAggregator(entry.getKey(), entry.getValue());
		}
		
		// build the messaging function (co group)
		CoGroupOperator<?, ?, Tuple2<VertexKey, Message>> messages;
		if (edgesWithoutValue != null) {
			MessagingUdfNoEdgeValues<VertexKey, VertexValue, Message> messenger = new MessagingUdfNoEdgeValues<VertexKey, VertexValue, Message>(messagingFunction, messageTypeInfo);
			messages = this.edgesWithoutValue.coGroup(iteration.getWorkset()).where(0).equalTo(0).with(messenger);
		}
		else {
			MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message, EdgeValue> messenger = new MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message, EdgeValue>(messagingFunction, messageTypeInfo);
			messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(0).equalTo(0).with(messenger);
		}
		
		// configure coGroup message function with name and broadcast variables
		messages = messages.name("Messaging");
		for (Tuple2<String, DataSet<?>> e : this.bcVarsMessaging) {
			messages = messages.withBroadcastSet(e.f1, e.f0);
		}
		
		VertexUpdateUdf<VertexKey, VertexValue, Message> updateUdf = new VertexUpdateUdf<VertexKey, VertexValue, Message>(updateFunction, vertexTypes);
		
		// build the update function (co group)
		CoGroupOperator<?, ?, Tuple2<VertexKey, VertexValue>> updates =
				messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
		
		// configure coGroup update function with name and broadcast variables
		updates = updates.name("Vertex State Updates");
		for (Tuple2<String, DataSet<?>> e : this.bcVarsUpdate) {
			updates = updates.withBroadcastSet(e.f1, e.f0);
		}

		// let the operator know that we preserve the key field
		updates.withConstantSetFirst("0").withConstantSetSecond("0");
		
		return iteration.closeWith(updates, updates);
		
	}
1310310
public static <T> Operation<T> operation(Class<? extends T> type, Operator operator,
                                             Expression<?>... args) {
        return operation(type, operator, ImmutableList.copyOf(args));
    }
805975
@Override
	@SuppressWarnings("unchecked")
	public BinaryOperatorInformation<IN1, IN2, OUT> getOperatorInfo() {
		return (BinaryOperatorInformation<IN1, IN2, OUT>) this.operatorInfo;
	}
741168
public <E extends Edge> double[] compute(Graph<E> g) {
        // Perform a quick test for whether the vertices of g are a contiguous
        // sequence starting at 0, which makes the vertex mapping trivial
        if (!hasContiguousVertices(g))
            throw new IllegalArgumentException(
                "Vertices must be in continugous order");
        
        double[] centralities = new double[g.order()];
        IntIterator vertexIter = g.vertices().iterator();
        while (vertexIter.hasNext()) {
            int s = vertexIter.nextInt();
            Deque<Integer> S = new ArrayDeque<Integer>();

            // Initialize P to an empty list for each vertex
            List<List<Integer>> P = new ArrayList<List<Integer>>(g.order());
            for (int i = 0; i < g.order(); ++i)
                P.add(new ArrayList<Integer>());

            double[] sigma = new double[g.order()];
            sigma[s] = 1;

            double[] d = new double[g.order()];
            Arrays.fill(d, -1);
            d[s] = 0;
            
            Queue<Integer> Q = new ArrayDeque<Integer>();
            Q.add(s);
            while (!Q.isEmpty()) {
                int v = Q.poll();
                S.offer(v);
                IntIterator neighborIter = g.getNeighbors(v).iterator();
                while (neighborIter.hasNext()) { 
                    int w = neighborIter.nextInt();
                    // Check whether this is the first time we've seen vertex w
                    if (d[w] < 0) {
                        Q.offer(w);
                        d[w] = d[v] + 1;
                    }
                    // Check whether the shortest path to w is through v
                    if (d[w] == d[v] + 1) {
                        sigma[w] += sigma[v];
                        P.get(w).add(v);
                    }
                }
            }
            double[] delta = new double[g.order()];
            // S as a stack returns vertices in order of their non-increasing
            // distance from vertex s
            while (!S.isEmpty()) {
                int w = S.pollLast(); // get the top of the stack
                for (int v : P.get(w)) {
                    delta[v] += (sigma[v] / sigma[w]) * (1 + delta[w]);
                }
                if (w != s) {
                    centralities[w] += delta[w];
                }            
            }
        }
        return centralities;
    }
1062365
public static Vertex newB()
    {
        return new Vertex(Predefine.TAG_BIGIN, " ", new CoreDictionary.Attribute(Nature.begin, Predefine.MAX_FREQUENCY / 10), CoreDictionary.getWordID(Predefine.TAG_BIGIN));
    }
795239
public static <T extends Geometry> JTSGeometryExpression<T> geometryOperation(Class<? extends T> type,
                                                                                  Operator op, Expression<?>... args) {
        return new JTSGeometryOperation<T>(type, op, args);
    }
1372692
@Override
    public Iterable<TitanVertex> vertices() {
        GraphCentricQuery query = constructQuery(ElementCategory.VERTEX);
        return Iterables.filter(new QueryProcessor<GraphCentricQuery, TitanElement, JointIndexQuery>(query, tx.elementProcessor), TitanVertex.class);
    }
569715
public Operator operator(final Token tok) throws Exception
    {
        final String token = tok.getValue().toString();
        final Constructor<?> constructor = operatorMap.get(token);
        if (constructor == null)
            return null;
        try
        {
            return (Operator) constructor.newInstance(new Object[] {
                    tok
            });
        } catch (final Exception e)
        {
            throw new Exception("operator construction", e);
        }
    }
564732
@Override
    protected InterproceduralCallGraphEdge allocateEdge(InterproceduralCallGraphVertex source,
            InterproceduralCallGraphVertex target) {
        return new InterproceduralCallGraphEdge(source, target);
    }
511087
public Graph<K, VV, EV> getUndirected() {

		DataSet<Edge<K, EV>> undirectedEdges = edges.
			flatMap(new RegularAndReversedEdgesMap<>()).name("To undirected graph");
		return new Graph<>(vertices, undirectedEdges, this.context);
	}
114060
public static int stencilOperationToGL(
    final JCGLStencilOperation op)
  {
    switch (op) {
      case STENCIL_OP_DECREMENT:
        return GL.GL_DECR;
      case STENCIL_OP_DECREMENT_WRAP:
        return GL.GL_DECR_WRAP;
      case STENCIL_OP_INCREMENT:
        return GL.GL_INCR;
      case STENCIL_OP_INCREMENT_WRAP:
        return GL.GL_INCR_WRAP;
      case STENCIL_OP_INVERT:
        return GL.GL_INVERT;
      case STENCIL_OP_KEEP:
        return GL.GL_KEEP;
      case STENCIL_OP_REPLACE:
        return GL.GL_REPLACE;
      case STENCIL_OP_ZERO:
        return GL.GL_ZERO;
    }

    throw new UnreachableCodeException();
  }
183425
@Override
    public DataBuffer getVectorCoordinates() {
        int idx;
        if (isRowVector()) {
            idx = 1;
        } else if (isColumnVector()) {
            idx = 0;
        } else {
            throw new UnsupportedOperationException();
        }

        // FIXME: int cast
        int[] temp = new int[(int) length()];
        for (int i = 0; i < length(); i++) {
            temp[i] = getUnderlyingIndicesOf(i).getInt(idx);
        }
        return Nd4j.createBuffer(temp);
    }
804495
@Override
    public Iterable<Vertex> vertices() {
        GraphCentricQuery query = constructQuery(ElementCategory.VERTEX);
        return Iterables.filter(new QueryProcessor<>(query, tx.elementProcessor), Vertex.class);
    }
836822
public <NV> Graph<K, NV, EV> mapVertices(final MapFunction<Vertex<K, VV>, NV> mapper, TypeInformation<Vertex<K, NV>> returnType) {
		DataSet<Vertex<K, NV>> mappedVertices = vertices.map(
				new MapFunction<Vertex<K, VV>, Vertex<K, NV>>() {
					private Vertex<K, NV> output = new Vertex<>();

					public Vertex<K, NV> map(Vertex<K, VV> value) throws Exception {
						output.f0 = value.f0;
						output.f1 = mapper.map(value);
						return output;
					}
				})
				.returns(returnType)
				.withForwardedFields("f0")
					.name("Map vertices");

		return new Graph<>(mappedVertices, this.edges, this.context);
	}
234283
public static TerminalOp<Integer, OptionalInt>
    makeInt(IntBinaryOperator operator) {
        Objects.requireNonNull(operator);
        class ReducingSink
                implements AccumulatingSink<Integer, OptionalInt, ReducingSink>, Sink.OfInt {
            private boolean empty;
            private int state;

            public void begin(long size) {
                empty = true;
                state = 0;
            }

            @Override
            public void accept(int t) {
                if (empty) {
                    empty = false;
                    state = t;
                }
                else {
                    state = operator.applyAsInt(state, t);
                }
            }

            @Override
            public OptionalInt get() {
                return empty ? OptionalInt.empty() : OptionalInt.of(state);
            }

            @Override
            public void combine(ReducingSink other) {
                if (!other.empty)
                    accept(other.state);
            }
        }
        return new ReduceOp<Integer, OptionalInt, ReducingSink>(StreamShape.INT_VALUE) {
            @Override
            public ReducingSink makeSink() {
                return new ReducingSink();
            }
        };
    }
563238
public GraphObject getGraphObject(Object obj)
	{
		String key = getKey(obj);
		GraphObject go = objectMap.get(key);

//		if (obj instanceof Conversion && go == null)
//		{
//			go = objectMap.get(key + ConversionWrapper.LEFT_TO_RIGHT);
//			if (go == null)
//				go = objectMap.get(key + ConversionWrapper.RIGHT_TO_LEFT);
//		}

		if (go == null)
		{
			Node node = wrap(obj);

			if (node != null)
			{
				objectMap.put(key, node);
				node.init();
			}
		}

		return objectMap.get(key);
	}
845880
public Graph<T> reduce() {
        Builder<T> builder = new Builder<>();
        nodes.stream()
                .forEach(u -> {
                    builder.addNode(u);
                    edges.get(u).stream()
                         .filter(v -> !pathExists(u, v, false))
                         .forEach(v -> builder.addEdge(u, v));
                });
        return builder.build();
    }