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

    if (!forceRepartition && !keyFormatChange && !repartitionNeeded) {
      return this;
    }

    if (schema.key().size() > 1) {
      // let's throw a better error message in the case of multi-column tables
      throw new UnsupportedOperationException("Cannot repartition a TABLE source. If this is "
          + "a join, joins on tables with multiple columns is not yet supported.");
    }

    // Table repartitioning is only supported for internal use in enabling joins
    // where we know that the key will be semantically equivalent, but may be serialized
    // differently (thus ensuring all keys are routed to the same partitions)
    if (repartitionNeeded) {
      throw new UnsupportedOperationException("Cannot repartition a TABLE source. "
          + "If this is a join, make sure that the criteria uses the TABLE's key column "
          + Iterables.getOnlyElement(schema.key()).name().text() + " instead of "
          + keyExpression);
    }

    if (keyFormat.isWindowed()) {
      final String errorMsg = "Implicit repartitioning of windowed sources is not supported. "
          + "See https://github.com/confluentinc/ksql/issues/4385.";
      final String additionalMsg = forceRepartition
          ? " As a result, ksqlDB does not support joins on windowed sources with "
          + "Schema-Registry-enabled key formats (AVRO, JSON_SR, PROTOBUF) at this time. "
          + "Please repartition your sources to use a different key format before performing "
          + "the join."
          : "";
      throw new KsqlException(errorMsg + additionalMsg);
    }

    final KeyFormat newKeyFormat = SerdeFeaturesFactory.sanitizeKeyFormat(
        forceInternalKeyFormat.orElse(keyFormat),
        toSqlTypes(keyExpression),
        false // logical schema changes are not supported
    );

    final ExecutionStep<KTableHolder<K>> step = ExecutionStepFactory.tableSelectKey(
        contextStacker,
        sourceTableStep,
        InternalFormats.of(newKeyFormat, valueFormat),
        keyExpression
    );

    return new SchemaKTable<>(
        step,
        resolveSchema(step),
        newKeyFormat,
        ksqlConfig,
        functionRegistry
    );
  }
@Test
  public void shouldFailSelectKeyForceRepartitionOnNonKeyColumn() {
    // Given:
    final String selectQuery = "SELECT col0, col2, col3 FROM test2 WHERE col0 > 100 EMIT CHANGES;";
    final PlanNode logicalPlan = buildLogicalPlan(selectQuery);
    initialSchemaKTable = buildSchemaKTableFromPlan(logicalPlan);

    // When:
    final UnsupportedOperationException e = assertThrows(
        UnsupportedOperationException.class,
        () -> initialSchemaKTable.selectKey(
            valueFormat.getFormatInfo(),
            ImmutableList.of(new UnqualifiedColumnReferenceExp(ColumnName.of("COL1"))),
            Optional.empty(),
            childContextStacker,
            true
    ));

    // Then:
    assertThat(e.getMessage(), containsString("Cannot repartition a TABLE source."));
  }