GroupByAndAggregate.java 文件源码

@Override
public void notifyDone() {
    List<String> keys = groupByKeys.getSortedKeys();
    for (String groupByKey : keys) {
        String[] variables = groupByKey.isEmpty() ? new String[0] : groupByKey.split(
            GROUP_BY_KEY_DELIMITER);
        Object[] tuple = new Object[variables.length + valueAggregatorPairs.size()];
        System.arraycopy(variables, 0, tuple, 0, variables.length);
        int i = variables.length;
        int index = groupByKeys.mapStringKeyToInt(groupByKey);
        for (Pair<EdgeOrVertexPropertyDescriptor, AbstractAggregator> valueAggregatorPair :
            valueAggregatorPairs) {
            tuple[i++] = valueAggregatorPair.b.getValue(index);
        }
        tuples.addTuple(tuple);
    }
    ((OutputSink) nextOperator).append(tuples);
    this.tuples = new Tuples(columnTypes, columnNames);
    this.vertexIndices = new ArrayList<>();
    this.edgeIndices = new ArrayList<>();
    super.notifyDone();
}