class SchemaKTable extends SchemaKStream {

  @SuppressWarnings("unchecked")
  @Override
  public SchemaKTable filter(final Expression filterExpression) {
    SqlPredicate predicate = new SqlPredicate(
        filterExpression,
        schema,
        isWindowed,
        functionRegistry
    );
    KTable filteredKTable = ktable.filter(predicate.getPredicate());
    return new SchemaKTable(
        schema,
        filteredKTable,
        keyField,
        Arrays.asList(this),
        isWindowed,
        Type.FILTER,
        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
  );

}