class SchemaKGroupedTable extends SchemaKGroupedStream {

  @Override
  public SchemaKTable<Struct> aggregate(
      final List<ColumnName> nonAggregateColumns,
      final List<FunctionCall> aggregations,
      final Optional<WindowExpression> windowExpression,
      final ValueFormat valueFormat,
      final QueryContext.Stacker contextStacker
  ) {
    if (windowExpression.isPresent()) {
      throw new KsqlException("Windowing not supported for table aggregations.");
    }

    final List<String> unsupportedFunctionNames = aggregations.stream()
        .map(call -> UdafUtil.resolveAggregateFunction(functionRegistry, call, schema))
        .filter(function -> !(function instanceof TableAggregationFunction))
        .map(KsqlAggregateFunction::name)
        .map(FunctionName::text)
        .distinct()
        .collect(Collectors.toList());

    if (!unsupportedFunctionNames.isEmpty()) {
      final String postfix = unsupportedFunctionNames.size() == 1 ? "" : "s";
      throw new KsqlException("The aggregation function" + postfix + " "
          + GrammaticalJoiner.and().join(unsupportedFunctionNames)
          + " cannot be applied to a table source, only to a stream source."
      );
    }

    final TableAggregate step = ExecutionStepFactory.tableAggregate(
        contextStacker,
        sourceTableStep,
        InternalFormats.of(keyFormat, valueFormat),
        nonAggregateColumns,
        aggregations
    );

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

    SchemaKGroupedTable(
      final ExecutionStep<KGroupedTableHolder> sourceTableStep,
      final LogicalSchema schema,
      final KeyFormat keyFormat,
      final KsqlConfig ksqlConfig,
      final FunctionRegistry functionRegistry
  );

}