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 shouldResolveSchemaForStreamSelectKeyV2() {
    // Given:
    final UnqualifiedColumnReferenceExp keyExpression1 =
        new UnqualifiedColumnReferenceExp(ColumnName.of("ORANGE"));
    final UnqualifiedColumnReferenceExp keyExpression2 =
        new UnqualifiedColumnReferenceExp(ColumnName.of("APPLE"));

    final StreamSelectKey<GenericKey> step = new StreamSelectKey<>(
        PROPERTIES,
        streamSource,
        ImmutableList.of(keyExpression1, keyExpression2)
    );

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

    // Then:
    assertThat(result, is(LogicalSchema.builder()
        .keyColumn(keyExpression1.getColumnName(), SqlTypes.INTEGER)
        .keyColumn(keyExpression2.getColumnName(), SqlTypes.BIGINT)
        .valueColumns(SCHEMA.value())
        .build()
    ));
  }