public static <K> KStreamHolder<K> build(
      final KStreamHolder<K> stream,
      final StreamSelect<K> step,
      final RuntimeBuildContext buildContext
  ) {
    final QueryContext queryContext = step.getProperties().getQueryContext();

    final LogicalSchema sourceSchema = stream.getSchema();
    final Optional<ImmutableList<ColumnName>> selectedKeys = step.getSelectedKeys();

    final Selection<K> selection = Selection.of(
        sourceSchema,
        step.getKeyColumnNames(),
        selectedKeys,
        step.getSelectExpressions(),
        buildContext.getKsqlConfig(),
        buildContext.getFunctionRegistry()
    );

    final ImmutableList.Builder<Integer> keyIndexBuilder = ImmutableList.builder();
    if (selectedKeys.isPresent()) {
      final ImmutableList<ColumnName> keyNames = sourceSchema.key().stream()
          .map(Column::name)
          .collect(ImmutableList.toImmutableList());

      for (final ColumnName keyName : selectedKeys.get()) {
        keyIndexBuilder.add(keyNames.indexOf(keyName));
      }
    }
    final ImmutableList<Integer> keyIndices = keyIndexBuilder.build();

    final SelectValueMapper<K> selectMapper = selection.getMapper();

    final ProcessingLogger logger = buildContext.getProcessingLogger(queryContext);

    final Named selectName =
        Named.as(StreamsUtil.buildOpName(queryContext));

    if (selectedKeys.isPresent() && !selectedKeys.get().containsAll(
        sourceSchema.key().stream().map(Column::name).collect(ImmutableList.toImmutableList())
    )) {
      return stream.withStream(
          stream.getStream().transform(
            () -> new KsTransformer<>(
                (readOnlyKey, value, ctx) -> {
                  if (keyIndices.isEmpty()) {
                    return null;
                  }

                  if (readOnlyKey instanceof GenericKey) {
                    final GenericKey keys = (GenericKey) readOnlyKey;
                    final Builder resultKeys = GenericKey.builder(keyIndices.size());

                    for (final int keyIndex : keyIndices) {
                      resultKeys.append(keys.get(keyIndex));
                    }

                    return (K) resultKeys.build();
                  } else {
                    throw new UnsupportedOperationException();
                  }
                },
                selectMapper.getTransformer(logger)
            ),
            selectName
          ),
          selection.getSchema()
      );
    } else {
      return stream.withStream(
          stream.getStream().transformValues(
              () -> new KsValueTransformer<>(selectMapper.getTransformer(logger)),
              selectName
          ),
          selection.getSchema()
      );
    }
  }
@Test
  public void shouldReturnCorrectSchema() {
    // When:
    final KStreamHolder<Struct> result = step.build(planBuilder, planInfo);

    // Then:
    assertThat(
        result.getSchema(),
        is(LogicalSchema.builder()
            .keyColumn(SystemColumns.ROWKEY_NAME, SqlTypes.STRING)
            .valueColumn(ColumnName.of("expr1"), SqlTypes.STRING)
            .valueColumn(ColumnName.of("expr2"), SqlTypes.INTEGER)
            .build())
    );
  }