@Override
  public SchemaKTable<GenericKey> aggregate(
      final List<ColumnName> nonAggregateColumns,
      final List<FunctionCall> aggregations,
      final Optional<WindowExpression> windowExpression,
      final FormatInfo valueFormat,
      final 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, ksqlConfig))
        .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
    );
  }
@Test
  public void shouldFailWindowedTableAggregation() {
    // Given:
    final WindowExpression windowExp = mock(WindowExpression.class);

    final SchemaKGroupedTable groupedTable = buildSchemaKGroupedTable();

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> groupedTable.aggregate(
            NON_AGG_COLUMNS,
            ImmutableList.of(SUM, COUNT),
            Optional.of(windowExp),
            valueFormat.getFormatInfo(),
            queryContext
        )
    );

    // Then:
    assertThat(e.getMessage(), containsString("Windowing not supported for table aggregations."));
  }