@SuppressWarnings("unchecked")
  public SchemaKTable<?> aggregate(
      final List<ColumnName> nonAggregateColumns,
      final List<FunctionCall> aggregations,
      final Optional<WindowExpression> windowExpression,
      final FormatInfo valueFormat,
      final 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 = SerdeFeaturesFactory.sanitizeKeyFormat(
          this.keyFormat,
          toSqlTypes(schema.key()),
          false
      );
      step = ExecutionStepFactory.streamAggregate(
          contextStacker,
          sourceStep,
          InternalFormats.of(keyFormat, valueFormat),
          nonAggregateColumns,
          aggregations
      );
    }

    return new SchemaKTable(
        step,
        resolveSchema(step),
        keyFormat,
        ksqlConfig,
        functionRegistry
    );
  }
@Test
  public void shouldReturnKTableWithOutputSchema() {
    // When:
    final SchemaKTable result = schemaGroupedStream.aggregate(
        NON_AGGREGATE_COLUMNS,
        ImmutableList.of(AGG),
        Optional.empty(),
        valueFormat.getFormatInfo(),
        queryContext
    );

    // Then:
    assertThat(result.getSchema(), is(OUT_SCHEMA));
  }