class FilterNode extends SingleSourcePlanNode {

  @Override
  public SchemaKStream<?> buildStream(final KsqlQueryBuilder builder) {
    final Stacker contextStacker = builder.buildNodeContext(getId().toString());

    return getSource().buildStream(builder)
        .filter(
            getPredicate(),
            contextStacker
        );
  }

  public  FilterNode(
      final PlanNodeId id,
      final PlanNode source,
      final Expression predicate
  );

}