void translate(final LogicalAggregate aggregate, final TranslatorContext context) {
    validateAggregateFunctions(aggregate);

    MessageStream<SamzaSqlRelMessage> inputStream = context.getMessageStream(aggregate.getInput().getId());

    // At this point, the assumption is that only count function is supported.
    SupplierFunction<Long> initialValue = () -> (long) 0;
    FoldLeftFunction<SamzaSqlRelMessage, Long> foldCountFn = (m, c) -> c + 1;

    final ArrayList<String> aggFieldNames = getAggFieldNames(aggregate);

    MessageStream<SamzaSqlRelMessage> outputStream =
        inputStream
            .map(new TranslatorInputMetricsMapFunction(logicalOpId))
            .window(Windows.keyedTumblingWindow(m -> m,
                Duration.ofMillis(context.getExecutionContext().getSamzaSqlApplicationConfig().getWindowDurationMs()),
                initialValue,
                foldCountFn,
                new SamzaSqlRelMessageSerdeFactory.SamzaSqlRelMessageSerde(),
                new LongSerde())
                .setAccumulationMode(
                    AccumulationMode.DISCARDING), changeLogStorePrefix + "_tumblingWindow_" + logicalOpId)
            .map(windowPane -> {
              List<String> fieldNames = windowPane.getKey().getKey().getSamzaSqlRelRecord().getFieldNames();
              List<Object> fieldValues = windowPane.getKey().getKey().getSamzaSqlRelRecord().getFieldValues();
              fieldNames.add(aggFieldNames.get(0));
              fieldValues.add(windowPane.getMessage());
              return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata(0L, 0L));
            });
    context.registerMessageStream(aggregate.getId(), outputStream);
    outputStream.map(new TranslatorOutputMetricsMapFunction(logicalOpId));
  }
Translator to translate the LogicalAggregate node in the relational graph to the corresponding StreamGraph
implementation
samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java