/*
Implementation notes:

<p>Use aggregator to replace SumEdgeStats when aggregators are rewritten to use
a hash-combineable hashed-reduce.
423192
@Override
	public EdgeMetrics<K, VV, EV> run(Graph<K, VV, EV> input)
			throws Exception {
		super.run(input);

		// s, t, (d(s), d(t))
		DataSet<Edge<K, Tuple3<EV, LongValue, LongValue>>> edgeDegreePair = input
			.run(new EdgeDegreePair<K, VV, EV>()
				.setReduceOnTargetId(reduceOnTargetId)
				.setParallelism(parallelism));

		// s, d(s), count of (u, v) where deg(u) < deg(v) or (deg(u) == deg(v) and u < v)
		DataSet<Tuple3<K, LongValue, LongValue>> edgeStats = edgeDegreePair
			.map(new EdgeStats<>())
				.setParallelism(parallelism)
				.name("Edge stats")
			.groupBy(0)
			.reduce(new SumEdgeStats<>())
			.setCombineHint(CombineHint.HASH)
				.setParallelism(parallelism)
				.name("Sum edge stats");

		edgeMetricsHelper = new EdgeMetricsHelper<>();

		edgeStats
			.output(edgeMetricsHelper)
				.setParallelism(parallelism)
				.name("Edge metrics");

		return this;
	}
624466
@Override
	public void aggregate(T value) {

		if (value == null) {
			nullCount++;
		}
		else if (isNan(value)) {
			nanCount++;
		}
		else if (isInfinite(value)) {
			infinityCount++;
		}
		else {
			nonMissingCount++;

			min.aggregate(value);
			max.aggregate(value);
			sum.aggregate(value);

			double doubleValue = value.doubleValue();
			double delta = doubleValue - mean.value();
			mean = mean.add(delta / nonMissingCount);
			m2 = m2.add(delta * (doubleValue - mean.value()));
		}
	}
1186292
private AggregationBuilder appendAggregation(CollectionExpression collectionExpression,
            EntityMetadata entityMetadata, AggregationBuilder aggregationBuilder)
    {
        ListIterable<Expression> functionlist = collectionExpression.children();
        for (Expression function : functionlist)
        {
            if (isAggregationExpression(function) && checkIfKeyExists(function.toParsedText()))
            {
                aggregationBuilder.subAggregation(getMetricsAggregation(function, entityMetadata));
            }
        }
        return aggregationBuilder;
    }
680198
function aggregateMetrics(currentTime, metricData) {
    var aggregateKey;

    /**
     * Place aggregate data into the specified slot.  If the current zoom
     * level matches the aggregate level, the data is emitted to keep the
     * display in sync.
     *
     * @param {Number} index
     * The desired slot for the aggregate.
     *
     * @param {Object} data
     * The aggregate data.
     *
     * @returns {void}
     */
    var setAggregateData =
      function setAggregateData(index, data) {
        this._aggregation[aggregateKey].data[index] = data;

        // if this view (current or not) is scrolled, adjust it
        if (this._aggregation[aggregateKey].scrollOffset) {
          this._aggregation[aggregateKey].scrollOffset--;
        }

        // emit to keep the display in sync
        if (this.isCurrentZoom(aggregateKey)) {
          if (this.isScrolled()) {
            this.emit("refreshMetrics");
          } else {
            this.emit("metrics", data);
          }
        }
      }.bind(this);

    /**
     * Given the current time time index, add any missing logical
     * time slots to have a complete picture of data.
     *
     * @param {Number} currentTimeIndex
     * The time index currently being processed.
     *
     * @returns {void}
     */
    var addMissingTimeSlots =
      function addMissingTimeSlots(currentTimeIndex) {
        var aggregateIndex = this._aggregation[aggregateKey].data.length;
        while (aggregateIndex < currentTimeIndex) {
          setAggregateData(aggregateIndex++, this.emptyAverage);
        }
      }.bind(this);

    /**
     * After having detected a new sampling, aggregate the relevant data points
     *
     * @param {Object[]} rows
     * The array reference.
     *
     * @param {Number} startIndex
     * The starting index to derive an average.
     *
     * @param {Number} endIndex
     * The ending index to derive an average.
     *
     * @returns {void}
     */
    var getAveragedAggregate =
      function getAveragedAggregate(rows, startIndex, endIndex) {
        var averagedAggregate = getInitializedAverage(metricData);

        // this is the number of elements we will aggregate
        var aggregateCount = endIndex - startIndex + 1;

        // you can compute an average of a set of numbers two ways
        // first, you can add all the numbers together and then divide by the count
        // second, you call divide each number by the count and add the quotients
        // the first method is more accurate, however you can overflow an accumulator
        // and result with NaN
        // the second method is employed here to ensure no overflows

        for (var dataKey in metricData) {
          for (var dataMetricKey in metricData[dataKey]) {
            for (var rowIndex = startIndex; rowIndex <= endIndex; rowIndex++) {
              averagedAggregate[dataKey][dataMetricKey] +=
                rows[rowIndex][dataKey][dataMetricKey] / aggregateCount;
            }

            // after the average is done, truncate the averages to one decimal point
            averagedAggregate[dataKey][dataMetricKey] =
              +averagedAggregate[dataKey][dataMetricKey].toFixed(1);
          }
        }

        return averagedAggregate;
      };

    /**
     * Process one row of metric data into aggregate.
     *
     * @param {Number} rowIndex
     * The index of the row being processed.
     *
     * @param {Object[]} rows
     * The array reference.
     *
     * @returns {void}
     */
    var processRow = function processRow(rowIndex, rows) {
      var averagedAggregate;
      var lastTimeIndex = this._aggregation[aggregateKey].lastTimeIndex;

      // get the time index of the aggregate
      var currentTimeIndex =
        convertElapsedTimeToTimeIndex(currentTime, this._startTime, +aggregateKey);

      // when the time index changes, average the data for the time band
      if (currentTimeIndex !== lastTimeIndex) {
        // except for the first one
        if (lastTimeIndex !== undefined) {
          // add in any missing logical time slots
          addMissingTimeSlots.call(this, lastTimeIndex);

          // get the average across the discovered time band
          averagedAggregate = getAveragedAggregate(
            rows,
            this._aggregation[aggregateKey].lastAggregateIndex,
            rowIndex - 1
          );

          // place the average
          setAggregateData(lastTimeIndex, averagedAggregate);

          // now we know where the next aggregate begins
          this._aggregation[aggregateKey].lastAggregateIndex = rowIndex;
        }

        // level-break
        this._aggregation[aggregateKey].lastTimeIndex = currentTimeIndex;
      }
    }.bind(this);

    // iterate over the configured aggregation time buckets
    for (aggregateKey in this._aggregation) {
      // iterate through metrics, beginning where we left off
      processRow(this._lastAggregationIndex, this._metrics);
    }

    // remember where we will begin again
    this._lastAggregationIndex++;
  }
1249506
public AggregationHashChain build(TreeNode leaf) throws KSIException {
        Util.notNull(leaf, "TreeNode");
        if (!leaf.isLeaf() || leaf.isRoot() || leaf instanceof MetadataNode) {
            throw new IllegalArgumentException("Aggregation hash chain can be built only from leaf nodes");
        }

        LinkedList<AggregationChainLink> links = new LinkedList<>();
        long levelCorrection = 0L;
        TreeNode node;
        if (leaf.getParent().getRightChildNode() instanceof MetadataNode) {
            links.add(createMetadataChainLink((MetadataNode) leaf.getParent().getRightChildNode(), leaf.getLevel()));
            node = leaf.getParent();
        } else {
            node = leaf;
            levelCorrection = node.getLevel();
        }
        createChainLinks(links, levelCorrection, node);
        LinkedList<Long> chainIndex = new LinkedList<>(singletonList(calculateIndex(links)));
        HashAlgorithm aggregationAlgorithm = new DataHash(leaf.getParent().getValue()).getAlgorithm();
        return SIGNATURE_COMPONENT_FACTORY.createAggregationHashChain(
                new DataHash(leaf.getValue()), new Date(), chainIndex, links, aggregationAlgorithm);
    }
953292
private void createInitialGroupEdges(final HashMap<AbstractJobVertex, ExecutionVertex> vertexMap)
			throws GraphConversionException {

		Iterator<Map.Entry<AbstractJobVertex, ExecutionVertex>> it = vertexMap.entrySet().iterator();

		while (it.hasNext()) {

			final Map.Entry<AbstractJobVertex, ExecutionVertex> entry = it.next();
			final AbstractJobVertex sjv = entry.getKey();
			final ExecutionVertex sev = entry.getValue();
			final ExecutionGroupVertex sgv = sev.getGroupVertex();

			// First compare number of output gates
			if (sjv.getNumberOfForwardConnections() != sgv.getEnvironment().getNumberOfOutputGates()) {
				throw new GraphConversionException("Job and execution vertex " + sjv.getName()
					+ " have different number of outputs");
			}

			if (sjv.getNumberOfBackwardConnections() != sgv.getEnvironment().getNumberOfInputGates()) {
				throw new GraphConversionException("Job and execution vertex " + sjv.getName()
					+ " have different number of inputs");
			}

			// First, build the group edges
			for (int i = 0; i < sjv.getNumberOfForwardConnections(); ++i) {
				final JobEdge edge = sjv.getForwardConnection(i);
				final AbstractJobVertex tjv = edge.getConnectedVertex();

				final ExecutionVertex tev = vertexMap.get(tjv);
				final ExecutionGroupVertex tgv = tev.getGroupVertex();
				// Use NETWORK as default channel type if nothing else is defined by the user
				ChannelType channelType = edge.getChannelType();
				boolean userDefinedChannelType = true;
				if (channelType == null) {
					userDefinedChannelType = false;
					channelType = ChannelType.NETWORK;
				}

				final DistributionPattern distributionPattern = edge.getDistributionPattern();

				// Connect the corresponding group vertices and copy the user settings from the job edge
				final ExecutionGroupEdge groupEdge = sgv.wireTo(tgv, edge.getIndexOfInputGate(), i, channelType,
					userDefinedChannelType,distributionPattern);

				final ExecutionGate outputGate = new ExecutionGate(new GateID(), sev, groupEdge, false);
				sev.insertOutputGate(i, outputGate);
				final ExecutionGate inputGate = new ExecutionGate(new GateID(), tev, groupEdge, true);
				tev.insertInputGate(edge.getIndexOfInputGate(), inputGate);
			}
		}
	}
1179432
void getAffectedEdges(int cluster, AttributeStreamOfInt32 edges) {
		int first_edge = getClusterFirstEdge(cluster);
		if (first_edge == -1)
			return;

		int edge = first_edge;
		do {
			int sweepNode = getEdgeSweepNode(edge);
			if (StridedIndexTypeCollection.isValidElement(sweepNode)) {
				edges.add(edge);
			}
			edge = getNextEdge(edge, cluster);
		} while (edge != first_edge);
	}
406094
public static KamSummary summarizeKamNetwork(Collection<KamEdge> edges,
            int statementCount) {
        KamSummary summary = new KamSummary();

        Set<KamNode> nodes = new HashSet<KamNode>(); //unique set of nodes
        for (KamEdge edge : edges) {
            nodes.add(edge.getSourceNode());
            nodes.add(edge.getTargetNode());
        }

        summary.setNumOfBELStatements(statementCount);
        summary.setNumOfNodes(nodes.size());
        summary.setNumOfEdges(edges.size());
        return summary;
    }
1015088
protected void updateOutgoingEdgesStateM(
			int currS, int currL,
			int prevS, int prevL,
			Emissions emission
	){
		updateOutgoingEdges(currS, currL, States.M, prevS, prevL, States.DL, Transitions.t_DL_to_M, emission);
		updateOutgoingEdges(currS, currL, States.M, prevS, prevL, States.M, Transitions.t_M_in, emission);
		updateOutgoingEdges(currS, currL, States.M, prevS, prevL, States.D, Transitions.t_D_to_M, emission);
		updateOutgoingEdges(currS, currL, States.M, prevS, prevL, States.S, Transitions.t_S_to_M, emission);
	}
242318
@Override
	public void setAggregator(IAggregator aggregator) {
		this.aggregator = aggregator;
		// get the file resource factory
		Iterable<IAggregatorExtension> iter = aggregator.getExtensions(IResourceFactoryExtensionPoint.ID);
		for (IAggregatorExtension ext : iter) {
			if ("file".equals(ext.getAttribute("scheme"))) { //$NON-NLS-1$ //$NON-NLS-2$
				fileResourceFactory = (IResourceFactory)ext.getInstance();
				break;
			}
		}
		if (fileResourceFactory == null) {
			throw new IllegalStateException("Missing file resource factory"); //$NON-NLS-1$
		}
	}
985795
public void processMetricAggregation() {
        if (!isEmpty(metricAggregation)) {
            String[] aggregations = metricAggregation.split(";");
            for (int i = 0; i < aggregations.length; i++) {
                String definition = aggregations[i].trim();
                int index0 = definition.indexOf("=");
                int index1 = definition.indexOf("(");
                int index2 = definition.indexOf(")");
                String aggregatedName = definition.substring(0, index0);
                String operation = definition.substring(index0 + 1, index1);
                String parameters = definition.substring(index1 + 1, index2);
                List<String> metricNames = new ArrayList<>();
                metricNames.add(operation);
                // Case a list of metricNames
                if (parameters.contains(",")) {
                    String[] listMetricNames = parameters.split(",");
                    for (int j = 0; j < listMetricNames.length; j++) {
                        metricNames.add(listMetricNames[j].trim());
                    }
                } else {
                    Pattern pattern = Pattern.compile(parameters);
                    for (String metricName : nameIdMetrics.keySet()) {
                        if (pattern.matcher(metricName).matches()) {
                            metricNames.add(metricName);
                        }
                    }
                }
                aggregatedMetrics.put(aggregatedName, metricNames);
            }
            log.debugf("Gnocchi Aggregated Metrics from %s -> %s", metricAggregation, aggregatedMetrics);
        }
    }
9475
public void accumulate(List<? extends StandardMetricResult> metricValues) {
        if (metricValues == null || metricValues.isEmpty()) return;
        for (StandardMetricResult v : metricValues) {
            if (lastOffset < v.offset) values.put(v.value.intValue());
        }
        lastOffset = ListUtils.last(metricValues).offset;
    }
1068042
private void updateLastSeenEdge(Vertex v) {
    Edge lastSeenEdge = getLastSeenEdge();
    if (lastSeenEdge != null) {
      if (lastSeenEdge.getSourceVertexId() == null) {
        lastSeenEdge.setSourceVertexId(v.getId());
      } else if (lastSeenEdge.getTargetVertexId() == null) {
        lastSeenEdge.setTargetVertexId(v.getId());
      }
    }
  }
1308742
@Private
  @Override
  public void aggregationCompleted(final List<Integer> taskletIds, final TOutput result) {
    try {
      completedTasklets(result, taskletIds);
    } catch (final InterruptedException e) {
      throw new RuntimeException(e);
    }
  }
23216
private <T> QueryRunner<T> withPerSinkMetrics(
      final QueryRunner<T> sinkRunner,
      final QueryToolChest<T, ? extends Query<T>> queryToolChest,
      final SegmentId sinkSegmentId,
      final AtomicLong cpuTimeAccumulator
  )
  {
    // Note: reportSegmentAndCacheTime and reportSegmentTime are effectively the same here. They don't split apart
    // cache vs. non-cache due to the fact that Sinks may be partially cached and partially uncached. Making this
    // better would need to involve another accumulator like the cpuTimeAccumulator that we could share with the
    // sinkRunner.
    String sinkSegmentIdString = sinkSegmentId.toString();
    return CPUTimeMetricQueryRunner.safeBuild(
        new MetricsEmittingQueryRunner<>(
            emitter,
            queryToolChest,
            new MetricsEmittingQueryRunner<>(
                emitter,
                queryToolChest,
                sinkRunner,
                QueryMetrics::reportSegmentTime,
                queryMetrics -> queryMetrics.segment(sinkSegmentIdString)
            ),
            QueryMetrics::reportSegmentAndCacheTime,
            queryMetrics -> queryMetrics.segment(sinkSegmentIdString)
        ).withWaitMeasuredFromNow(),
        queryToolChest,
        emitter,
        cpuTimeAccumulator,
        false
    );
  }
1230148
@NotNull
    public static <Key extends Serializable, ValueIn, MapOutput extends Serializable, ReducerResult extends Serializable> ReducerResult compute(@NotNull AnalyticsContext analyticsContext,
                                                                                                                                                @NotNull MapReduceTask<Key, ValueIn, MapOutput, ReducerResult> mapReduceTask) {

        Timer timer = MetricsFactory.getMetrics().getTimer("analytics.compute");

        ClusterGroup clusterGroup = analyticsContext.getIgnite()
                                                    .cluster()
                                                    .forServers();


        ReducerResult execute = analyticsContext.getIgnite()
                                                .compute(clusterGroup)
                                                .execute(mapReduceTask, null);

        timer.stop();

        LOGGER.debug("Complete compute operation in {}", timer);

        return execute;
    }
771581
public void aggregateValue(TridentTuple tuple) {
        K key = mapper.getKey(tuple);
        V value = mapper.getValue(tuple);
        V currentValue = aggregateValues.get(key);
        V newValue;
        if (currentValue == null) {
            newValue = aggregator.init(tuple);
        } else {
            newValue = aggregator.combine(currentValue, value);
        }
        LOG.debug("Updating state [{}] ==> [{}]", new Object[]{key, newValue});
        aggregateValues.put(key, newValue);
    }