class ProjectNode extends SingleSourcePlanNode {

  @Override
  public SchemaKStream<?> buildStream(final KsqlQueryBuilder builder) {
    final SchemaKStream<?> stream = getSource().buildStream(builder);

    final List<ColumnName> keyColumnNames = getSchema().key().stream()
        .map(Column::name)
        .collect(Collectors.toList());

    return stream.select(
        keyColumnNames,
        getSelectExpressions(),
        builder.buildNodeContext(getId().toString()),
        builder
    );
  }

  public  ProjectNode(
      final PlanNodeId id,
      final PlanNode source
  );

}