### Question:
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 ); } SchemaKTable(
      final Schema schema,
      final KTable ktable,
      final Field keyField,
      final List<SchemaKStream> sourceSchemaKStreams,
      boolean isWindowed,
      Type type,
      final FunctionRegistry functionRegistry,
      final SchemaRegistryClient schemaRegistryClient
  ); @SuppressWarnings("unchecked") @Override SchemaKTable into(
      final String kafkaTopicName,
      final Serde<GenericRow> topicValueSerDe,
      Set<Integer> rowkeyIndexes
  ); @Override QueuedSchemaKStream toQueue(Optional<Integer> limit); @SuppressWarnings("unchecked") @Override SchemaKTable filter(final Expression filterExpression); @SuppressWarnings("unchecked") @Override SchemaKTable select(final List<Pair<String, Expression>> expressionPairList); @Override KStream getKstream(); KTable getKtable(); boolean isWindowed();  }### Answer:
@Test public void testFilter() throws Exception { String selectQuery = "SELECT col0, col2, col3 FROM test1 WHERE col0 > 100;"; PlanNode logicalPlan = planBuilder.buildLogicalPlan(selectQuery); FilterNode filterNode = (FilterNode) logicalPlan.getSources().get(0).getSources().get(0); initialSchemaKTable = new SchemaKTable(logicalPlan.getTheSourceNode().getSchema(), kTable, ksqlTable.getKeyField(), new ArrayList<>(), false, SchemaKStream.Type.SOURCE, functionRegistry, new MockSchemaRegistryClient()); SchemaKTable filteredSchemaKStream = initialSchemaKTable.filter(filterNode.getPredicate()); Assert.assertTrue(filteredSchemaKStream.getSchema().fields().size() == 8); Assert.assertTrue(filteredSchemaKStream.getSchema().field("TEST1.COL0") == filteredSchemaKStream.getSchema().fields().get(2)); Assert.assertTrue(filteredSchemaKStream.getSchema().field("TEST1.COL1") == filteredSchemaKStream.getSchema().fields().get(3)); Assert.assertTrue(filteredSchemaKStream.getSchema().field("TEST1.COL2") == filteredSchemaKStream.getSchema().fields().get(4)); Assert.assertTrue(filteredSchemaKStream.getSchema().field("TEST1.COL3") == filteredSchemaKStream.getSchema().fields().get(5)); Assert.assertTrue(filteredSchemaKStream.getSchema() .field("TEST1.COL0").schema().type() == Schema.Type.INT64); Assert.assertTrue(filteredSchemaKStream.getSchema() .field("TEST1.COL1").schema().type() == Schema.Type.STRING); Assert.assertTrue(filteredSchemaKStream.getSchema() .field("TEST1.COL2").schema().type() == Schema.Type.STRING); Assert.assertTrue(filteredSchemaKStream.getSchema() .field("TEST1.COL3").schema().type() == Schema.Type.FLOAT64); Assert.assertTrue(filteredSchemaKStream.getSourceSchemaKStreams().get(0) == initialSchemaKTable); }