public SchemaKStream<K> selectKey(
      final FormatInfo valueFormat,
      final List<Expression> keyExpression,
      final Optional<KeyFormat> forceInternalKeyFormat,
      final Stacker contextStacker,
      final boolean forceRepartition
  ) {
    final boolean keyFormatChange = forceInternalKeyFormat.isPresent()
        && !forceInternalKeyFormat.get().equals(keyFormat);

    final boolean repartitionNeeded = repartitionNeeded(keyExpression);
    if (!keyFormatChange && !forceRepartition && !repartitionNeeded) {
      return this;
    }

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

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

    final KeyFormat newKeyFormat = forceInternalKeyFormat.orElse(keyFormat);
    return new SchemaKStream<>(
        step,
        resolveSchema(step),
        SerdeFeaturesFactory.sanitizeKeyFormat(
            newKeyFormat,
            toSqlTypes(keyExpression),
            true),
        ksqlConfig,
        functionRegistry
    );
  }
@Test(expected = UnsupportedOperationException.class)
  public void shouldFailRepartitionTable() {
    // Given:
    givenInitialKStreamOf("SELECT * FROM test2 EMIT CHANGES;");

    final UnqualifiedColumnReferenceExp col2 =
        new UnqualifiedColumnReferenceExp(ColumnName.of("COL2"));

    // When:
    schemaKTable.selectKey(
        valueFormat.getFormatInfo(),
        ImmutableList.of(col2),
        Optional.empty(),
        childContextStacker,
        false
    );
  }