@Override
  public SchemaKTable<K> select(
      final List<ColumnName> keyColumnNames,
      final List<SelectExpression> selectExpressions,
      final Stacker contextStacker,
      final PlanBuildContext buildContext,
      final FormatInfo valueFormat
  ) {
    final TableSelect<K> step = ExecutionStepFactory.tableMapValues(
        contextStacker,
        sourceTableStep,
        keyColumnNames,
        selectExpressions,
        InternalFormats.of(keyFormat, valueFormat)
    );

    return new SchemaKTable<>(
        step,
        resolveSchema(step),
        keyFormat,
        ksqlConfig,
        functionRegistry
    );
  }
@Test
  public void testSelectWithFunctions() {
    // Given:
    final String selectQuery =
        "SELECT col0, LEN(UCASE(col2)) FROM test2 WHERE col0 > 100 EMIT CHANGES;";
    final PlanNode logicalPlan = buildLogicalPlan(selectQuery);
    final ProjectNode projectNode = (ProjectNode) logicalPlan.getSources().get(0);
    initialSchemaKTable = buildSchemaKTableFromPlan(logicalPlan);

    // When:
    final SchemaKTable<?> projectedSchemaKStream = initialSchemaKTable.select(
        ImmutableList.of(),
        projectNode.getSelectExpressions(),
        childContextStacker,
        buildContext,
        internalFormats
    );

    // Then:
    assertThat(projectedSchemaKStream.getSchema(), is(LogicalSchema.builder()
        .keyColumn(ColumnName.of("COL0"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of("COL0"), SqlTypes.BIGINT)
        .valueColumn(ColumnName.of("KSQL_COL_0"), SqlTypes.INTEGER)
        .build()
    ));
  }