code: @Override
      public Stream partitionAggregate(Fields inputFields, Aggregator agg, Fields functionFields) {
          projectionValidation(inputFields);
          return _topology.addSourcedNode(this,
                  new ProcessorNode(_topology.getUniqueStreamId(),
                      _name,
                      functionFields,
                      functionFields,
                      new AggregateProcessor(inputFields, agg)));
      }
docs: creates brand new tuples with brand new fields