class SchemaKTable extends SchemaKStream {

  @SuppressWarnings("unchecked")
  @Override
  public SchemaKTable select(final List<Pair<String, Expression>> expressionPairList) {

    final Pair<Schema, SelectValueMapper> schemaAndMapper
        = createSelectValueMapperAndSchema(expressionPairList);

    KTable projectedKTable = ktable.mapValues(schemaAndMapper.right);

    return new SchemaKTable(
        schemaAndMapper.left,
        projectedKTable,
        keyField,
        Collections.singletonList(this),
        isWindowed,
        Type.PROJECT,
        functionRegistry,
        schemaRegistryClient
    );
  }

  public  SchemaKTable(
      final Schema schema,
      final KTable ktable,
      final Field keyField,
      final List<SchemaKStream> sourceSchemaKStreams,
      boolean isWindowed,
      Type type,
      final FunctionRegistry functionRegistry,
      final SchemaRegistryClient schemaRegistryClient
  );

}