class SchemaKGroupedStream {

  @SuppressWarnings("unchecked")
  public SchemaKTable<?> aggregate(
      final List<ColumnName> nonAggregateColumns,
      final List<FunctionCall> aggregations,
      final Optional<WindowExpression> windowExpression,
      final ValueFormat valueFormat,
      final QueryContext.Stacker contextStacker
  ) {
    final ExecutionStep<? extends KTableHolder<?>> step;
    final KeyFormat keyFormat;

    if (windowExpression.isPresent()) {
      keyFormat = getKeyFormat(windowExpression.get());
      step = ExecutionStepFactory.streamWindowedAggregate(
          contextStacker,
          sourceStep,
          InternalFormats.of(keyFormat, valueFormat),
          nonAggregateColumns,
          aggregations,
          windowExpression.get().getKsqlWindowExpression()
      );
    } else {
      keyFormat = this.keyFormat;
      step = ExecutionStepFactory.streamAggregate(
          contextStacker,
          sourceStep,
          InternalFormats.of(keyFormat, valueFormat),
          nonAggregateColumns,
          aggregations
      );
    }

    return new SchemaKTable(
        step,
        resolveSchema(step),
        keyFormat,
        ksqlConfig,
        functionRegistry
    );
  }

    SchemaKGroupedStream(
      final ExecutionStep<KGroupedStreamHolder> sourceStep,
      final LogicalSchema schema,
      final KeyFormat keyFormat,
      final KsqlConfig ksqlConfig,
      final FunctionRegistry functionRegistry
  );

}