public LogicalSchema resolve(final ExecutionStep<?> step, final LogicalSchema schema) {
    return Optional.ofNullable(HANDLERS.get(step.getClass()))
        .map(h -> h.handle(this, schema, step))
        .orElseThrow(() -> new IllegalStateException("Unhandled step class: " + step.getClass()));
  }
@Test
  public void shouldResolveSchemaForStreamWindowedAggregate() {
    // Given:
    givenAggregateFunction("COUNT");
    final StreamWindowedAggregate step = new StreamWindowedAggregate(
        PROPERTIES,
        groupedStreamSource,
        formats,
        ImmutableList.of(ColumnName.of("ORANGE")),
        ImmutableList.of(functionCall("COUNT", "APPLE")),
        new TumblingWindowExpression(new WindowTimeClause(10, TimeUnit.SECONDS))
    );

    // When:
    final LogicalSchema result = resolver.resolve(step, SCHEMA);

    // Then:
    assertThat(result, is(
        LogicalSchema.builder()
            .keyColumn(ColumnName.of("K0"), SqlTypes.INTEGER)
            .valueColumn(ColumnName.of("ORANGE"), SqlTypes.INTEGER)
            .valueColumn(ColumnNames.aggregateColumn(0), SqlTypes.BIGINT)
            .valueColumn(SystemColumns.WINDOWSTART_NAME, SystemColumns.WINDOWBOUND_TYPE)
            .valueColumn(SystemColumns.WINDOWEND_NAME, SystemColumns.WINDOWBOUND_TYPE)
            .build())
    );
  }