### Question:
SchemaKStream { @SuppressWarnings("unchecked") public SchemaKStream filter(final Expression filterExpression) { SqlPredicate predicate = new SqlPredicate(filterExpression, schema, false, functionRegistry); KStream<String, GenericRow> filteredKStream = kstream.filter(predicate.getPredicate()); return new SchemaKStream( schema, filteredKStream, keyField, Arrays.asList(this), Type.FILTER, functionRegistry, schemaRegistryClient ); } SchemaKStream(
      final Schema schema,
      final KStream<String, GenericRow> kstream,
      final Field keyField,
      final List<SchemaKStream> sourceSchemaKStreams,
      final Type type,
      final FunctionRegistry functionRegistry,
      final SchemaRegistryClient schemaRegistryClient
  ); QueuedSchemaKStream toQueue(Optional<Integer> limit); SchemaKStream into(
      final String kafkaTopicName,
      final Serde<GenericRow> topicValueSerDe,
      final Set<Integer> rowkeyIndexes
  ); @SuppressWarnings("unchecked") SchemaKStream filter(final Expression filterExpression); SchemaKStream select(final Schema selectSchema); SchemaKStream select(final List<Pair<String, Expression>> expressionPairList); @SuppressWarnings("unchecked") SchemaKStream leftJoin(
      final SchemaKTable schemaKTable,
      final Schema joinSchema,
      final Field joinKey,
      KsqlTopicSerDe joinSerDe,
      KsqlConfig ksqlConfig
  ); @SuppressWarnings("unchecked") SchemaKStream selectKey(final Field newKeyField, boolean updateRowKey); SchemaKGroupedStream groupBy(
      final Serde<String> keySerde, final Serde<GenericRow> valSerde,
      final List<Expression> groupByExpressions); Field getKeyField(); Schema getSchema(); KStream getKstream(); List<SchemaKStream> getSourceSchemaKStreams(); String getExecutionPlan(String indent); OutputNode outputNode(); void setOutputNode(final OutputNode output); Type getType(); FunctionRegistry getFunctionRegistry(); SchemaRegistryClient getSchemaRegistryClient();  }### 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); initialSchemaKStream = new SchemaKStream(logicalPlan.getTheSourceNode().getSchema(), kStream, ksqlStream.getKeyField(), new ArrayList<>(), SchemaKStream.Type.SOURCE, functionRegistry, new MockSchemaRegistryClient()); SchemaKStream filteredSchemaKStream = initialSchemaKStream.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) == initialSchemaKStream); }