class SchemaKStream {

  @SuppressWarnings("unchecked")
  public SchemaKStream<Struct> selectKey(
      final Expression keyExpression,
      final Stacker contextStacker
  ) {
    if (repartitionNotNeeded(ImmutableList.of(keyExpression))) {
      return (SchemaKStream<Struct>) this;
    }

    if (keyFormat.isWindowed()) {
      throw new KsqlException("Implicit repartitioning of windowed sources is not supported. "
          + "See https://github.com/confluentinc/ksql/issues/4385.");
    }

    final ExecutionStep<KStreamHolder<Struct>> step = ExecutionStepFactory
        .streamSelectKey(contextStacker, sourceStep, keyExpression);

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

    SchemaKStream(
      final ExecutionStep<KStreamHolder<K>> sourceStep,
      final LogicalSchema schema,
      final KeyFormat keyFormat,
      final KsqlConfig ksqlConfig,
      final FunctionRegistry functionRegistry
  );

}